Implement API token authentication: add BearerAuth and SessionOrBearer classes, create endpoints for token generation and validation

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:
+class BearerAuth(HttpBearer):
-    """Simple bearer token auth for Ninja endpoints.
+    """Ninja HttpBearer auth that validates APIToken entries.
 
-    Integrates with `APIToken` model; use as `auth=TokenAuth()` on router
+    Returns the token owner user on success or raises HttpError(401) on failure.
-    endpoints that should accept tokens in `Authorization: Bearer <token>`.
     """
-    def __call__(self, request):
+    def authenticate(self, request, token: str):
-        # Extract header
+        if not token:
-        auth = request.META.get("HTTP_AUTHORIZATION")
+            raise HttpError(status_code=401)
-        if not auth:
+
-            return None
-        parts = auth.split()
-        if len(parts) != 2 or parts[0].lower() != "bearer":
-            return None
-        token = parts[1]
         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())
+class TokenAuthIn(Schema):
-def create_api_token(request, payload: CreateTokenIn):
+    username: str
-    expires = None
+    password: str
-    if payload.expires_days:
-        expires = timezone.now() + timezone.timedelta(days=payload.expires_days)
 
-    token, obj = APIToken.create_token(request.user, name=payload.name, scopes=payload.scopes, expires=expires)
+
-    # Return raw token once — callers must store it
+@csrf_exempt
-    return {"token": token, "id": obj.id}
+@router.post("/create_api_token")
+def token_auth(request, payload: TokenAuthIn):
+    """Authenticate username/password and return a new API token.
+
+    Returns JSON: {"token": "<raw-token>", "id": <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 <token>` header or POST JSON
+    body `{ "token": "<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())