diff --git a/assets b/assets index 3b3d948ad..63ce633bc 160000 --- a/assets +++ b/assets @@ -1 +1 @@ -Subproject commit 3b3d948ad2b8f64b3777b34b7009da521abaf67f +Subproject commit 63ce633bcd20f38dd63653e1235404a648bf58f6 diff --git a/registrations/admin.py b/registrations/admin.py index 49058544c..1caec1b5a 100644 --- a/registrations/admin.py +++ b/registrations/admin.py @@ -1,5 +1,7 @@ +from admin_auto_filters.filters import AutocompleteFilterFactory +from django import forms from django.conf import settings -from django.contrib import admin +from django.contrib import admin, messages from django.http import HttpResponseRedirect from django.template.loader import render_to_string from django_admin_listfilter_dropdown.filters import RelatedDropdownFilter @@ -177,6 +179,53 @@ class DomainWhitelistAdmin(CompareVersionAdmin): ordering = ("domain_name",) +# NOTE: External systems (eg. Montandon eoAPI) cache the revocation list, so re-enabling a +# disabled token is not applied instantly downstream. +RE_ENABLE_PROPAGATION_HINT = ( + "Re-enabling a disabled token may take some time to take effect: " + "external systems (eg. Montandon eoAPI) can keep it in their revoke list until their cache expires." +) + + +class UserExternalTokenAdminForm(forms.ModelForm): + class Meta: + model = models.UserExternalToken + fields = "__all__" + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + if "is_disabled" in self.fields: + self.fields["is_disabled"].help_text = RE_ENABLE_PROPAGATION_HINT + + +class UserExternalTokenAdmin(CompareVersionAdmin): + form = UserExternalTokenAdminForm + list_display = ("title", "user", "created_at", "expire_timestamp", "is_disabled", "is_old_token") + list_filter = (AutocompleteFilterFactory("User", "user"), "is_disabled", "is_old_token") + search_fields = ("title", "user__username", "user__email", "jti") + readonly_fields = ("jti", "created_at") + autocomplete_fields = ("user",) + actions = ("disable_tokens", "enable_tokens") + + def get_readonly_fields(self, request, obj=None): + readonly_fields = super().get_readonly_fields(request, obj) + if obj is not None: + # Owner/expiry are baked into the already issued JWT, and is_old_token is a + # migration marker, so none of them should be editable after creation + return readonly_fields + ("user", "expire_timestamp", "is_old_token") + return readonly_fields + + @admin.action(description="Disable selected tokens") + def disable_tokens(self, request, queryset): + updated = queryset.update(is_disabled=True) + self.message_user(request, f"{updated} token(s) disabled.") + + @admin.action(description="Enable selected tokens") + def enable_tokens(self, request, queryset): + updated = queryset.update(is_disabled=False) + self.message_user(request, f"{updated} token(s) enabled. {RE_ENABLE_PROPAGATION_HINT}", messages.WARNING) + + admin.site.register(models.Pending, PendingAdmin) admin.site.register(models.DomainWhitelist, DomainWhitelistAdmin) -admin.site.register(models.UserExternalToken) +admin.site.register(models.UserExternalToken, UserExternalTokenAdmin) diff --git a/registrations/migrations/0013_userexternaltoken_is_disabled.py b/registrations/migrations/0013_userexternaltoken_is_disabled.py new file mode 100644 index 000000000..7ebcae799 --- /dev/null +++ b/registrations/migrations/0013_userexternaltoken_is_disabled.py @@ -0,0 +1,18 @@ +# Generated by Django 5.2.14 on 2026-07-15 08:57 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('registrations', '0012_userexternaltoken_is_old_token'), + ] + + operations = [ + migrations.AddField( + model_name='userexternaltoken', + name='is_disabled', + field=models.BooleanField(default=False, help_text='Marks whether this token has been disabled/revoked', verbose_name='is disabled?'), + ), + ] diff --git a/registrations/models.py b/registrations/models.py index 790d1a86b..0c7eaecb8 100644 --- a/registrations/models.py +++ b/registrations/models.py @@ -98,8 +98,12 @@ class UserExternalToken(models.Model): default=False, help_text=_("Marks whether this is an old Montandon token"), ) - # @Note: Currently not used, but could be utilized for a blacklist feature. - # is_disabled = models.BooleanField(verbose_name=_('is disabled?'), default=False) + # @Note: Used as a blacklist/revocation flag. Disabled tokens are treated as inactive. + is_disabled = models.BooleanField( + verbose_name=_("is disabled?"), + default=False, + help_text=_("Marks whether this token has been disabled/revoked"), + ) class Meta: verbose_name = _("User External Token") @@ -112,6 +116,7 @@ def get_payload(self) -> dict: return { "jti": str(self.jti), "userId": self.user_id, + "iat": self.created_at, "exp": self.expire_timestamp, "inMovement": True, } diff --git a/registrations/serializers.py b/registrations/serializers.py index a28cbf5d3..5fcaece66 100644 --- a/registrations/serializers.py +++ b/registrations/serializers.py @@ -147,13 +147,21 @@ def save(self): class UserExternalTokenSerializer(serializers.ModelSerializer): - token = serializers.CharField(read_only=True) + token = serializers.SerializerMethodField() expire_timestamp = serializers.DateTimeField(required=False) is_old_token = serializers.BooleanField(read_only=True) + is_disabled = serializers.BooleanField(read_only=True) class Meta: model = UserExternalToken - fields = ["title", "token", "expire_timestamp", "created_at", "is_old_token"] + fields = ["id", "title", "token", "expire_timestamp", "created_at", "is_old_token", "is_disabled"] + + def get_token(self, obj) -> str | None: + # NOTE: The JWT token only exists on the creation path, where create() returns a dict + # containing "token". For stored model instances (list/retrieve/revoke) there is no token. + if isinstance(obj, dict): + return obj.get("token") + return None def validate_expire_timestamp(self, date): now = timezone.now() @@ -177,9 +185,22 @@ def create(self, validated_data): if not (settings.OIDC_RSA_PRIVATE_KEY and settings.OIDC_RSA_PUBLIC_KEY): raise serializers.ValidationError("Please contact system adminstrators to configurate private and public key.") instance = super().create(validated_data) + validated_data["id"] = instance.id validated_data["created_at"] = instance.created_at validated_data["token"] = jwt_encode_handler(instance.get_payload()) return validated_data def update(self, instance, validated_data): raise serializers.ValidationError("Update is not allowed") + + +class UserExternalTokenVerifySerializer(serializers.Serializer): + """Input serializer for the external-token verify endpoint.""" + + jti = serializers.UUIDField() + + +class UserExternalTokenVerifyResponseSerializer(serializers.Serializer): + """Output serializer for the external-token verify endpoint.""" + + active = serializers.BooleanField() diff --git a/registrations/test_views.py b/registrations/test_views.py index 792c977d5..949930b48 100644 --- a/registrations/test_views.py +++ b/registrations/test_views.py @@ -6,6 +6,8 @@ # 6. Use the admin token and new user username to query views.ValidateUser # 7. Confirm that a user without an official email is activated. +import uuid +from datetime import timedelta from unittest import mock from cryptography.hazmat.backends import default_backend @@ -13,12 +15,14 @@ from cryptography.hazmat.primitives.asymmetric import rsa from django.contrib.auth.models import User from django.test import override_settings +from django.utils import timezone from rest_framework.test import APITestCase from api.models import Country, Profile +from deployments.factories.user import UserFactory from main.test_case import APITestCase as GoAPITestCase -from .models import Pending +from .models import Pending, UserExternalToken class TwoGatekeepersTest(APITestCase): @@ -211,8 +215,147 @@ def test_external_token_with_key(self): ): response = self.client.post("/api/v2/external-token/", data, format="json") self.assertEqual(response.status_code, 201) + # get_token() returns the generated JWT on the creation path (a header.payload.signature string). + body = response.json() + self.assertTrue(body["token"]) + self.assertEqual(body["token"].count("."), 2) + # id is returned so clients can later revoke the token + self.assertEqual(body["id"], UserExternalToken.objects.get().id) def test_external_token_with_no_keys(self): self.client.force_authenticate(self.user) response = self.client.post("/api/v2/external-token/") self.assertEqual(response.status_code, 400) + + def test_verify_active_token(self): + token = UserExternalToken.objects.create( + title="active", + user=self.user, + expire_timestamp=timezone.now() + timedelta(days=1), + ) + # NOTE: verify is unauthenticated (server-to-server introspection) + response = self.client.post("/api/v2/external-token/verify/", {"jti": str(token.jti)}, format="json") + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json(), {"active": True}) + + def test_verify_disabled_token(self): + token = UserExternalToken.objects.create( + title="disabled", + user=self.user, + expire_timestamp=timezone.now() + timedelta(days=1), + is_disabled=True, + ) + response = self.client.post("/api/v2/external-token/verify/", {"jti": str(token.jti)}, format="json") + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json(), {"active": False}) + + def test_verify_expired_token(self): + token = UserExternalToken.objects.create( + title="expired", + user=self.user, + expire_timestamp=timezone.now() - timedelta(days=1), + ) + response = self.client.post("/api/v2/external-token/verify/", {"jti": str(token.jti)}, format="json") + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json(), {"active": False}) + + def test_verify_old_token(self): + token = UserExternalToken.objects.create( + title="old", + user=self.user, + expire_timestamp=timezone.now() + timedelta(days=1), + is_old_token=True, + ) + response = self.client.post("/api/v2/external-token/verify/", {"jti": str(token.jti)}, format="json") + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json(), {"active": False}) + + def test_verify_unknown_jti(self): + response = self.client.post("/api/v2/external-token/verify/", {"jti": str(uuid.uuid4())}, format="json") + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json(), {"active": False}) + + def test_verify_invalid_jti(self): + response = self.client.post("/api/v2/external-token/verify/", {"jti": "not-a-uuid"}, format="json") + self.assertEqual(response.status_code, 400) + + def test_revoke_own_token(self): + self.client.force_authenticate(self.user) + token = UserExternalToken.objects.create( + title="revoke-me", + user=self.user, + expire_timestamp=timezone.now() + timedelta(days=1), + ) + + # NOTE: verify is unauthenticated (server-to-server introspection) + # Token is active before revocation. + response = self.client.post("/api/v2/external-token/verify/", {"jti": str(token.jti)}, format="json") + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json(), {"active": True}) + + response = self.client.post(f"/api/v2/external-token/{token.id}/revoke/") + self.assertEqual(response.status_code, 200) + token.refresh_from_db() + self.assertIs(token.is_disabled, True) + + # Token is inactive after revocation. + response = self.client.post("/api/v2/external-token/verify/", {"jti": str(token.jti)}, format="json") + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json(), {"active": False}) + + def test_revoke_already_revoked_token_errors(self): + self.client.force_authenticate(self.user) + token = UserExternalToken.objects.create( + title="revoke-twice", + user=self.user, + expire_timestamp=timezone.now() + timedelta(days=1), + ) + response = self.client.post(f"/api/v2/external-token/{token.id}/revoke/") + self.assertEqual(response.status_code, 200) + + # A second revoke on the same token is a client error. + response = self.client.post(f"/api/v2/external-token/{token.id}/revoke/") + self.assertEqual(response.status_code, 400) + + def test_revoke_other_users_token_not_found(self): + other_user = UserFactory.create( + username="other@dave.com", + first_name="Other", + last_name="User", + password="test123", + email="other@dave.com", + ) + token = UserExternalToken.objects.create( + title="not-mine", + user=other_user, + expire_timestamp=timezone.now() + timedelta(days=1), + ) + self.client.force_authenticate(self.user) + response = self.client.post(f"/api/v2/external-token/{token.id}/revoke/") + self.assertEqual(response.status_code, 404) + token.refresh_from_db() + self.assertIs(token.is_disabled, False) + + def test_revoke_requires_auth(self): + token = UserExternalToken.objects.create( + title="revoke-unauth", + user=self.user, + expire_timestamp=timezone.now() + timedelta(days=1), + ) + response = self.client.post(f"/api/v2/external-token/{token.id}/revoke/") + self.assertEqual(response.status_code, 401) + + def test_list_includes_is_disabled(self): + self.client.force_authenticate(self.user) + UserExternalToken.objects.create( + title="listed", + user=self.user, + expire_timestamp=timezone.now() + timedelta(days=1), + ) + response = self.client.get("/api/v2/external-token/") + self.assertEqual(response.status_code, 200) + results = response.json()["results"] + self.assertEqual(len(results), 1) + self.assertIn("is_disabled", results[0]) + # get_token() returns None for stored instances, so the JWT is never re-exposed on list. + self.assertIsNone(results[0]["token"]) diff --git a/registrations/views.py b/registrations/views.py index 366c76abe..7bfc3599b 100644 --- a/registrations/views.py +++ b/registrations/views.py @@ -6,12 +6,19 @@ from django.http import HttpResponse from django.template.loader import render_to_string from django.utils import timezone +from drf_spectacular.utils import extend_schema from rest_framework import permissions, viewsets +from rest_framework.decorators import action +from rest_framework.response import Response from rest_framework.views import APIView from api.views import bad_http_request, bad_request from notifications.notification import send_notification -from registrations.serializers import UserExternalTokenSerializer +from registrations.serializers import ( + UserExternalTokenSerializer, + UserExternalTokenVerifyResponseSerializer, + UserExternalTokenVerifySerializer, +) from .models import Pending, UserExternalToken from .utils import getRegionalAdmins, is_valid_domain @@ -160,3 +167,41 @@ def get_queryset(self): def destroy(self, request, *args, **kwargs): return bad_request("Delete method not allowed") + + @extend_schema(request=None, responses=UserExternalTokenSerializer) + @action( + detail=True, + methods=["post"], + # NOTE: Allows a user to self-revoke one of their own tokens. get_object() is scoped to + # get_queryset (user=request.user), so other users' tokens resolve to 404. Revoking an + # already-revoked token is a client error (400). + url_path="revoke", + ) + def revoke(self, request, *args, **kwargs): + instance = self.get_object() + if instance.is_disabled: + return bad_request("Token is already revoked") + instance.is_disabled = True + instance.save(update_fields=["is_disabled"]) + return Response(self.get_serializer(instance).data) + + @extend_schema(responses=UserExternalTokenVerifyResponseSerializer) + @action( + detail=False, + methods=["post"], + # NOTE: This is a lightweight server-to-server introspection endpoint called by the + # external STAC auth proxy. It only returns a boolean and never leaks user data, so it + # is intentionally left unauthenticated (the jti itself is an unguessable UUID). + permission_classes=[permissions.AllowAny], + serializer_class=UserExternalTokenVerifySerializer, + ) + def verify(self, request, *args, **kwargs): + serializer = self.get_serializer(data=request.data) + serializer.is_valid(raise_exception=True) + active = UserExternalToken.objects.filter( + jti=serializer.validated_data["jti"], + is_disabled=False, + is_old_token=False, + expire_timestamp__gt=timezone.now(), + ).exists() + return Response({"active": active})