From a3ac91437f859bebe015413d440364f523d91bbe Mon Sep 17 00:00:00 2001 From: Ross Date: Mon, 23 Feb 2026 12:35:29 +0000 Subject: [PATCH] Implement API token authentication: add BearerAuth and SessionOrBearer classes, create endpoints for token generation and validation --- atlas/api.py | 187 ++++++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 161 insertions(+), 26 deletions(-) diff --git a/atlas/api.py b/atlas/api.py index 08efd677..995180a4 100644 --- a/atlas/api.py +++ b/atlas/api.py @@ -2,6 +2,8 @@ from collections import defaultdict from typing import List, Tuple from django.shortcuts import get_object_or_404 from django.urls import reverse +import json +from django.http import JsonResponse, HttpResponse from ninja import ModelSchema, Router, Schema, Field from django.db import transaction, IntegrityError @@ -15,12 +17,17 @@ from generic.decorators import check_user_in_group from generic.constants import Group from ninja.security import django_auth +from ninja.security import HttpBearer, HttpError from ninja import NinjaAPI, File from ninja.files import UploadedFile import hashlib import secrets from django.utils import timezone +from django.contrib.auth import authenticate +from django.views.decorators.csrf import csrf_exempt +from django.views.decorators.csrf import csrf_exempt +from django.middleware.csrf import CsrfViewMiddleware from generic.models import Examination, Modality @@ -33,41 +40,63 @@ from loguru import logger router = Router() -class TokenAuth: - """Simple bearer token auth for Ninja endpoints. +class BearerAuth(HttpBearer): + """Ninja HttpBearer auth that validates APIToken entries. - Integrates with `APIToken` model; use as `auth=TokenAuth()` on router - endpoints that should accept tokens in `Authorization: Bearer `. + Returns the token owner user on success or raises HttpError(401) on failure. """ - def __call__(self, request): - # Extract header - auth = request.META.get("HTTP_AUTHORIZATION") - if not auth: - return None - parts = auth.split() - if len(parts) != 2 or parts[0].lower() != "bearer": - return None - token = parts[1] + def authenticate(self, request, token: str): + if not token: + raise HttpError(status_code=401) + token_hash = hashlib.sha256(token.encode()).hexdigest() try: - api_token = APIToken.objects.get(token_hash=token_hash, revoked=False) + api_token = APIToken.objects.get(token_hash=token_hash) except APIToken.DoesNotExist: - return None + raise HttpError(status_code=401) + + if api_token.revoked: + raise HttpError(status_code=401) if api_token.expires and api_token.expires < timezone.now(): - return None + raise HttpError(status_code=401) - # mark last used (best-effort) try: api_token.mark_used() except Exception: pass - # Attach token object to request for handlers + # Attach for downstream handlers request.api_token = api_token return api_token.user +class SessionOrBearer: + """Composite auth: prefer Django session (request.user), fall back to bearer token. + + Use as `auth=SessionOrBearer()` on Ninja router endpoints. + """ + def __call__(self, request): + # Session-backed auth already populated by Django middleware + if getattr(request, "user", None) and getattr(request.user, "is_authenticated", False): + return request.user + + # Try bearer token from header + auth = request.META.get("HTTP_AUTHORIZATION") + if not auth: + raise HttpError(status_code=401) + parts = auth.split() + if len(parts) != 2 or parts[0].lower() != "bearer": + raise HttpError(status_code=401) + token = parts[1] + # reuse BearerAuth logic + bearer = BearerAuth() + return bearer.authenticate(request, token) + +# Backwards-compatible alias used in existing decorators +TokenAuth = SessionOrBearer + + class SeriesSchema(ModelSchema): case_id: List[int] = [] @@ -355,15 +384,121 @@ class CreateTokenIn(Schema): expires_days: int | None = Field(None, description="Expire after N days") -@router.post("/api_tokens", auth=TokenAuth()) -def create_api_token(request, payload: CreateTokenIn): - expires = None - if payload.expires_days: - expires = timezone.now() + timezone.timedelta(days=payload.expires_days) +class TokenAuthIn(Schema): + username: str + password: str - token, obj = APIToken.create_token(request.user, name=payload.name, scopes=payload.scopes, expires=expires) - # Return raw token once — callers must store it - return {"token": token, "id": obj.id} + +@csrf_exempt +@router.post("/create_api_token") +def token_auth(request, payload: TokenAuthIn): + """Authenticate username/password and return a new API token. + + Returns JSON: {"token": "", "id": } + """ + user = authenticate(request, username=payload.username, password=payload.password) + if user is None: + return HttpResponse(status=401, content=json.dumps({"detail": "Invalid credentials"}), content_type="application/json") + + # Create a new token for this login (no scopes by default) + token, obj = APIToken.create_token(user, name=f"login-{timezone.now().isoformat()}") + + return {"token": token, "id": obj.id, "username": user.username} + + +@csrf_exempt +@router.post("/token_check") +def token_check(request): + """Check whether a token is valid. + + Accepts either `Authorization: Bearer ` header or POST JSON + body `{ "token": "" }`. + + Returns JSON with keys: `valid` (bool), and when valid: `id`, `username`, + `scopes`, `expires`, `revoked`. + """ + auth = request.META.get("HTTP_AUTHORIZATION") + token_obj = None + + # If header present, try TokenAuth path (this will set request.api_token) + if auth: + try: + user = TokenAuth()(request) + token_obj = getattr(request, "api_token", None) + except HttpError: + return JsonResponse({"valid": False}, status=401) + else: + # try JSON body + try: + data = json.loads(request.body.decode("utf-8")) if request.body else {} + except Exception: + data = {} + token = data.get("token") + if token: + token_hash = hashlib.sha256(token.encode()).hexdigest() + try: + t = APIToken.objects.get(token_hash=token_hash) + token_obj = t + except APIToken.DoesNotExist: + token_obj = None + + if not token_obj: + return JsonResponse({"valid": False}, status=401) + + # Validate token state + valid = True + if token_obj.revoked: + valid = False + if token_obj.expires and token_obj.expires < timezone.now(): + valid = False + + if not valid: + return JsonResponse({"valid": False, "revoked": token_obj.revoked}, status=401) + + return JsonResponse( + { + "valid": True, + "id": token_obj.id, + "username": token_obj.user.username, + "scopes": token_obj.scopes, + "expires": token_obj.expires.isoformat() if token_obj.expires else None, + "revoked": token_obj.revoked, + }, + status=200, + ) + + +#@csrf_exempt +#@router.post("/create_api_token") +#def create_api_token(request, payload: CreateTokenIn): +# """Create an API token. +# +# This endpoint accepts either normal session auth (`request.user`) or +# a bearer token. If no authenticated user is found a 403 is returned. +# """ +# # If caller is not using Authorization header, enforce CSRF via middleware +# if not request.META.get("HTTP_AUTHORIZATION"): +# mw = CsrfViewMiddleware() +# resp = mw.process_view(request, create_api_token, (), {}) +# if resp is not None: +# return resp +# # Determine authenticated user: prefer session auth, fall back to token auth +# user = None +# if getattr(request, "user", None) and getattr(request.user, "is_authenticated", False): +# user = request.user +# else: +# user = TokenAuth()(request) +# +# if not user: +# raise PermissionDenied() +# +# expires = None +# if payload.expires_days: +# expires = timezone.now() + timezone.timedelta(days=payload.expires_days) +# +# token, obj = APIToken.create_token(user, name=payload.name, scopes=payload.scopes, expires=expires) +# # Return raw token once — callers must store it +# return {"token": token, "id": obj.id} @router.post("/api_tokens/{token_id}/revoke", auth=TokenAuth())