diff --git a/atlas/admin.py b/atlas/admin.py index 3b4f4e86..2e8fc42a 100755 --- a/atlas/admin.py +++ b/atlas/admin.py @@ -22,6 +22,7 @@ from .models import ( CaseDisplaySet, UserReportAnswer, CidReportAnswer, + APIToken, ) from django.forms import ModelForm @@ -142,3 +143,12 @@ class SeriesAdmin(VersionAdmin): admin.site.register(Series, SeriesAdmin) + + +class APITokenAdmin(admin.ModelAdmin): + list_display = ("user", "name", "revoked", "created", "expires", "last_used") + search_fields = ("user__username", "name") + readonly_fields = ("token_hash", "created", "last_used") + + +admin.site.register(APIToken, APITokenAdmin) diff --git a/atlas/api.py b/atlas/api.py index 1d5f18c2..08efd677 100644 --- a/atlas/api.py +++ b/atlas/api.py @@ -18,10 +18,13 @@ from ninja.security import django_auth from ninja import NinjaAPI, File from ninja.files import UploadedFile +import hashlib +import secrets +from django.utils import timezone from generic.models import Examination, Modality -from .models import Case, DuplicateDicom, Series, SeriesImage, UncategorisedDicom +from .models import Case, DuplicateDicom, Series, SeriesImage, UncategorisedDicom, APIToken from atlas.helpers import get_cases_available_to_user from loguru import logger @@ -30,6 +33,41 @@ from loguru import logger router = Router() +class TokenAuth: + """Simple bearer token auth for Ninja endpoints. + + Integrates with `APIToken` model; use as `auth=TokenAuth()` on router + endpoints that should accept tokens in `Authorization: Bearer `. + """ + 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] + token_hash = hashlib.sha256(token.encode()).hexdigest() + try: + api_token = APIToken.objects.get(token_hash=token_hash, revoked=False) + except APIToken.DoesNotExist: + return None + + if api_token.expires and api_token.expires < timezone.now(): + return None + + # mark last used (best-effort) + try: + api_token.mark_used() + except Exception: + pass + + # Attach token object to request for handlers + request.api_token = api_token + return api_token.user + + class SeriesSchema(ModelSchema): case_id: List[int] = [] @@ -54,7 +92,7 @@ class CaseSchema(ModelSchema): fields = ["id", "title"] -@router.post("/upload_dicom", auth=django_auth) +@router.post("/upload_dicom", auth=TokenAuth()) def upload_dicom(request, files: List[UploadedFile] = File(...)): uploaded = [] duplicate = [] @@ -87,7 +125,7 @@ def upload_dicom(request, files: List[UploadedFile] = File(...)): } -@router.post("/generate_image_hash", auth=django_auth) +@router.post("/generate_image_hash", auth=TokenAuth()) def generate_image_hash(request, id: int): s = SeriesImage.objects.get(pk=id) s.generate_hashes() @@ -99,7 +137,7 @@ def generate_image_hash(request, id: int): } -@router.post("/clear_dicoms", auth=django_auth) +@router.post("/clear_dicoms", auth=TokenAuth()) def clear_dicoms(request): if "selection" in request.POST: dicoms = UncategorisedDicom.objects.filter( @@ -114,7 +152,7 @@ def clear_dicoms(request): return True -@router.get("/uncategorised_dicoms", auth=django_auth) +@router.get("/uncategorised_dicoms", auth=TokenAuth()) def uncategorised_dicoms(request): dicoms = UncategorisedDicom.objects.filter(user=request.user) @@ -213,7 +251,7 @@ def import_dicoms_helper(request, case_id: int | None = None): @router.post( - "/import_dicoms", auth=django_auth, response=List[Tuple[SeriesSchema, str]] + "/import_dicoms", auth=TokenAuth(), response=List[Tuple[SeriesSchema, str]] ) def import_dicoms(request): return import_dicoms_helper(request) @@ -221,14 +259,14 @@ def import_dicoms(request): @router.post( "/import_dicoms/{case_id}", - auth=django_auth, + auth=TokenAuth(), response=List[Tuple[SeriesSchema, str]], ) def import_dicoms_case(request, case_id: int): return import_dicoms_helper(request, case_id=case_id) -@router.post("/upload_dicom_case/{case_id}", auth=django_auth, response=List[Tuple[SeriesSchema, str]]) +@router.post("/upload_dicom_case/{case_id}", auth=TokenAuth(), response=List[Tuple[SeriesSchema, str]]) def upload_dicom_case(request, case_id: int, files: List[UploadedFile] = File(...)): """Upload DICOM files and immediately import them into the given case. @@ -248,12 +286,12 @@ def upload_dicom_case(request, case_id: int, files: List[UploadedFile] = File(.. return import_dicoms_helper(request, case_id=case_id) -@router.get("/orphan_series", auth=django_auth, response=List[SeriesSchema]) +@router.get("/orphan_series", auth=TokenAuth(), response=List[SeriesSchema]) def orphan_series(request): return request.user.series.filter(case=None) -@router.get("/series_remove_duplicate_images", auth=django_auth) +@router.get("/series_remove_duplicate_images", auth=TokenAuth()) def series_remove_duplicate_images(request, series_id: int): series = get_object_or_404(Series, pk=series_id) @@ -270,19 +308,76 @@ def series_remove_duplicate_images(request, series_id: int): return len(dupes) -@router.get("/get_cases_user", auth=django_auth, response=List[CaseSchema]) +@router.get("/get_cases_user", auth=TokenAuth(), response=List[CaseSchema]) def get_cases_user(request): return Case.objects.filter(author=request.user) -@router.get("/get_cases_available", auth=django_auth, response=List[CaseSchema]) +@router.get("/get_cases_available", auth=TokenAuth(), response=List[CaseSchema]) def get_cases_available(request): """Return cases available to the authenticated user (via get_cases_available_to_user).""" qs = get_cases_available_to_user(request.user) return qs.order_by('-created_date')[:200] -@router.get("/check_image_hash/{hash}", auth=django_auth) +class APITokenOut(Schema): + id: int + name: str + scopes: str = "" + created: str + expires: str | None = None + revoked: bool = False + last_used: str | None = None + + +@router.get("/api_tokens", auth=TokenAuth(), response=List[APITokenOut]) +def list_api_tokens(request): + tokens = APIToken.objects.filter(user=request.user).order_by("-created") + out = [] + for t in tokens: + out.append( + { + "id": t.id, + "name": t.name, + "scopes": t.scopes, + "created": t.created.isoformat(), + "expires": t.expires.isoformat() if t.expires else None, + "revoked": t.revoked, + "last_used": t.last_used.isoformat() if t.last_used else None, + } + ) + return out + + +class CreateTokenIn(Schema): + name: str = Field("", description="Friendly name for token") + scopes: str = Field("", description="Space-separated scopes") + 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) + + 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} + + +@router.post("/api_tokens/{token_id}/revoke", auth=TokenAuth()) +def revoke_api_token(request, token_id: int): + try: + t = APIToken.objects.get(pk=token_id, user=request.user) + except APIToken.DoesNotExist: + return {"status": "not found"} + t.revoked = True + t.save(update_fields=["revoked"]) + return {"status": "revoked"} + + +@router.get("/check_image_hash/{hash}", auth=TokenAuth()) def check_image_hash(request, hash: str): try: series_image = SeriesImage.objects.get(image_blake3_hash=hash) @@ -297,7 +392,7 @@ def check_image_hash(request, hash: str): return data -@router.post("/check_image_hashes/", auth=django_auth) +@router.post("/check_image_hashes/", auth=TokenAuth()) def check_images_hashes(request, hashes: List[str]): """Checks a list of image hashes and returns the series id / url if found @@ -345,13 +440,13 @@ def check_images_hashes(request, hashes: List[str]): # print(series_image) -@router.get("/view_dicom_tags/{hash}", auth=django_auth) +@router.get("/view_dicom_tags/{hash}", auth=TokenAuth()) def view_dicom_tags(request, hash: str): item = SeriesImage.objects.get(image_blake3_hash=hash) return item.get_dicom_json() -@router.get("/series_split_by_dicom_tag/{series_id}/{dicom_tag}", auth=django_auth) +@router.get("/series_split_by_dicom_tag/{series_id}/{dicom_tag}", auth=TokenAuth()) def series_split_by_tag(request, series_id: int, dicom_tag: str): series = get_object_or_404(Series, pk=series_id) @@ -399,7 +494,7 @@ def series_split_by_tag(request, series_id: int, dicom_tag: str): return new_series -@router.get("/split_order_by_dicom_tag/{series_id}/{dicom_tag}", auth=django_auth) +@router.get("/split_order_by_dicom_tag/{series_id}/{dicom_tag}", auth=TokenAuth()) def series_order_by_tag(request, series_id: int, dicom_tag: str): series = get_object_or_404(Series, pk=series_id) diff --git a/atlas/migrations/0094_apitoken.py b/atlas/migrations/0094_apitoken.py new file mode 100644 index 00000000..5cb3ffc6 --- /dev/null +++ b/atlas/migrations/0094_apitoken.py @@ -0,0 +1,34 @@ +# Generated by Django 6.0.1 on 2026-02-23 11:51 + +import django.db.models.deletion +from django.conf import settings +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('atlas', '0093_procedure_case_procedures'), + migrations.swappable_dependency(settings.AUTH_USER_MODEL), + ] + + operations = [ + migrations.CreateModel( + name='APIToken', + fields=[ + ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('name', models.CharField(blank=True, help_text='Friendly name for this token', max_length=200)), + ('token_hash', models.CharField(db_index=True, max_length=64)), + ('scopes', models.CharField(blank=True, help_text='Space-separated scopes', max_length=255)), + ('created', models.DateTimeField(auto_now_add=True)), + ('expires', models.DateTimeField(blank=True, null=True)), + ('revoked', models.BooleanField(default=False)), + ('last_used', models.DateTimeField(blank=True, null=True)), + ('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='api_tokens', to=settings.AUTH_USER_MODEL)), + ], + options={ + 'verbose_name': 'API token', + 'verbose_name_plural': 'API tokens', + }, + ), + ] diff --git a/atlas/models.py b/atlas/models.py index a7fda4ce..0f2596dd 100644 --- a/atlas/models.py +++ b/atlas/models.py @@ -230,6 +230,44 @@ class SynMixin(object): return format_html("{}", self.get_absolute_url(), self.name) +class APIToken(models.Model): + """Personal Access Token for API access. + + Stores only a hash of the token. The raw token is returned once at + creation and never stored in plaintext. + """ + user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="api_tokens") + name = models.CharField(max_length=200, blank=True, help_text="Friendly name for this token") + token_hash = models.CharField(max_length=64, db_index=True) + scopes = models.CharField(max_length=255, blank=True, help_text="Space-separated scopes") + created = models.DateTimeField(auto_now_add=True) + expires = models.DateTimeField(null=True, blank=True) + revoked = models.BooleanField(default=False) + last_used = models.DateTimeField(null=True, blank=True) + + class Meta: + verbose_name = "API token" + verbose_name_plural = "API tokens" + + def __str__(self): + return f"APIToken(user={self.user}, name={self.name})" + + @classmethod + def create_token(cls, user, name="", scopes="", expires=None): + import secrets, hashlib + + token = secrets.token_urlsafe(32) + token_hash = hashlib.sha256(token.encode()).hexdigest() + obj = cls.objects.create(user=user, name=name, token_hash=token_hash, scopes=scopes, expires=expires) + return token, obj + + def mark_used(self): + from django.utils import timezone + + self.last_used = timezone.now() + self.save(update_fields=["last_used"]) + + class Finding(SynMixin, models.Model): name = models.CharField(max_length=255, unique=True) # New canonical/alias field: if set, this Finding is an alias and points diff --git a/atlas/templates/atlas/api_tokens.html b/atlas/templates/atlas/api_tokens.html new file mode 100644 index 00000000..8374f144 --- /dev/null +++ b/atlas/templates/atlas/api_tokens.html @@ -0,0 +1,64 @@ +{% extends 'base.html' %} +{% block content %} +
+

API Tokens

+
+
+
+ {% csrf_token %} +
+
+ +
+
+ +
+
+ +
+
+ +
+
+
+
+
+ + {% if created_token %} +
Created token. Save it now — it won't be shown again. + 
{{ created_token }}
+
+ {% endif %} + +
+
Your tokens
+
+ + + + {% for t in tokens %} + + + + + + + + + {% empty %} + + {% endfor %} + +
NameScopesCreatedExpiresRevoked
{{ t.name }}{{ t.scopes }}{{ t.created }}{{ t.expires }}{{ t.revoked }} + {% if not t.revoked %} +
+ {% csrf_token %} + + +
+ {% endif %} +
No tokens
+
+
+
+{% endblock %} diff --git a/atlas/urls.py b/atlas/urls.py index 07db0ee4..a6d94758 100755 --- a/atlas/urls.py +++ b/atlas/urls.py @@ -520,6 +520,7 @@ urlpatterns = [ views.SeriesImagesZipView.as_view(), name="series_download", ), + path("api_tokens/", views.api_tokens_page, name="api_tokens"), path( "series//order_dicom_instance", views.series_order_dicom_instance, diff --git a/atlas/views.py b/atlas/views.py index 511677d1..75c75807 100755 --- a/atlas/views.py +++ b/atlas/views.py @@ -119,6 +119,7 @@ from .models import ( UncategorisedDicom, UserReportAnswer, PrerequisiteRequired + , APIToken ) from .models import Procedure from .models import NormalCase @@ -519,6 +520,52 @@ def case_search_page(request): }, ) + + +@login_required +def api_tokens_page(request): + """Simple UI to create and revoke API tokens for the logged-in user.""" + created_token = None + + if request.method == "POST": + action = request.POST.get("action") + if action == "create": + name = request.POST.get("name", "") + scopes = request.POST.get("scopes", "") + expires_days = request.POST.get("expires_days") + expires = None + if expires_days: + try: + expires = timezone.now() + timezone.timedelta(days=int(expires_days)) + except Exception: + expires = None + + token, obj = APIToken.create_token(request.user, name=name, scopes=scopes, expires=expires) + created_token = token + elif action == "revoke": + revoke_id = request.POST.get("revoke_id") + try: + t = APIToken.objects.get(pk=revoke_id, user=request.user) + t.revoked = True + t.save(update_fields=["revoked"]) + except APIToken.DoesNotExist: + pass + + tokens_qs = APIToken.objects.filter(user=request.user).order_by("-created") + tokens = [ + { + "id": t.id, + "name": t.name, + "scopes": t.scopes, + "created": t.created, + "expires": t.expires, + "revoked": t.revoked, + } + for t in tokens_qs + ] + + return render(request, "atlas/api_tokens.html", {"tokens": tokens, "created_token": created_token}) + @login_required @user_has_case_view_access @require_http_methods(["GET", "POST"]) diff --git a/templates/profile.html b/templates/profile.html index 2994cd76..75e5a3c0 100644 --- a/templates/profile.html +++ b/templates/profile.html @@ -90,6 +90,9 @@ Update user details {% endif %} Edit profile + {% if request.user == user %} + API tokens + {% endif %} {% if request.user.is_superuser %} Admin edit {% endif %}