🧑‍💻(demo) add a token exchange test page

This must only be used in development mode to test
proper behavior of the token exchange API.

src/backend/conftest.py

@@ -1,9 +1,16 @@
 """Global fixtures for the backend tests."""
 
 import pytest
+from rest_framework.test import APIClient
 from urllib3.connectionpool import HTTPConnectionPool
 
 
+@pytest.fixture
+def api_client():
+    """Create API client."""
+    return APIClient()
+
+
 @pytest.fixture(autouse=True)
 def no_http_requests(monkeypatch):
     """

src/backend/demo/forms.py

@@ -0,0 +1,68 @@
+"""Forms for the demo helpers."""
+
+from django import forms
+
+
+class TokenExchangeDemoForm(forms.Form):
+    """Simple form to craft a token exchange request payload."""
+
+    client_id = forms.CharField(
+        required=False,
+        label="client_id (Basic Auth)",
+        widget=forms.TextInput(attrs={"autocomplete": "off"}),
+        help_text="Used only to build the Authorization header",
+    )
+    client_secret = forms.CharField(
+        required=False,
+        label="client_secret (Basic Auth)",
+        widget=forms.PasswordInput(render_value=True, attrs={"autocomplete": "off"}),
+        help_text="Used only to build the Authorization header",
+    )
+    grant_type = forms.CharField(
+        required=True,
+        initial="urn:ietf:params:oauth:grant-type:token-exchange",
+        widget=forms.TextInput(attrs={"size": 60}),
+    )
+    subject_token = forms.CharField(
+        required=True,
+        widget=forms.Textarea(
+            attrs={"rows": 3, "placeholder": "Access token to exchange"}
+        ),
+    )
+    subject_token_type = forms.CharField(
+        required=True,
+        initial="urn:ietf:params:oauth:token-type:access_token",
+        widget=forms.TextInput(attrs={"size": 60}),
+    )
+    requested_token_type = forms.CharField(
+        required=False,
+        widget=forms.TextInput(attrs={"size": 60}),
+        help_text="Optional: e.g. urn:ietf:params:oauth:token-type:jwt",
+    )
+    audience = forms.CharField(
+        required=False,
+        widget=forms.TextInput(attrs={"size": 60}),
+        help_text="Optional: space separated audience identifiers",
+    )
+    scope = forms.CharField(
+        required=False,
+        widget=forms.TextInput(attrs={"size": 60}),
+        help_text="Optional: space separated scopes",
+    )
+    actor_token = forms.CharField(
+        required=False,
+        widget=forms.Textarea(attrs={"rows": 2}),
+    )
+    actor_token_type = forms.CharField(
+        required=False,
+        widget=forms.TextInput(attrs={"size": 60}),
+    )
+    resource = forms.CharField(
+        required=False,
+        widget=forms.TextInput(attrs={"size": 60}),
+    )
+    expires_in = forms.IntegerField(
+        required=False,
+        min_value=1,
+        help_text="Optional: lifetime in seconds",
+    )

src/backend/demo/management/commands/create_demo.py

@@ -26,6 +26,7 @@ from mailbox_manager.enums import (
     MailDomainRoleChoices,
     MailDomainStatusChoices,
 )
+from token_exchange import models as token_exchange_models
 
 fake = Faker()
 
@@ -467,6 +468,75 @@ def create_demo(stdout):  # pylint: disable=too-many-branches too-many-statement
         create_oidc_people_idp_client()
         create_oidc_people_idp_client_user()
 
+    # Create token exchange base configuration
+    stdout.write("Creating token exchange base configuration")
+
+    people_service_provider, _created = (
+        token_exchange_models.ServiceProvider.objects.get_or_create(
+            name="People",  # yeah that's me!
+            audience_id="people",
+        )
+    )
+    docs_service_provider, _created = (
+        token_exchange_models.ServiceProvider.objects.get_or_create(
+            name="Docs",
+            audience_id="docs",
+        )
+    )
+
+    token_exchange_models.ServiceProviderCredentials.objects.get_or_create(
+        service_provider=people_service_provider,
+        client_id="client_id",
+        client_secret="client_secret",
+    )
+
+    people_to_people_rule, _created = (
+        token_exchange_models.TokenExchangeRule.objects.get_or_create(
+            source_service=people_service_provider,
+            target_service=people_service_provider,
+        )
+    )
+    token_exchange_models.ScopeGrant.objects.get_or_create(
+        rule=people_to_people_rule,
+        source_scope="openid",
+        granted_scope="openid",
+    )
+    token_exchange_models.ScopeGrant.objects.get_or_create(
+        rule=people_to_people_rule,
+        source_scope="openid",
+        granted_scope="docs:read",
+    )
+
+    people_to_docs_rule, _created = (
+        token_exchange_models.TokenExchangeRule.objects.get_or_create(
+            source_service=people_service_provider,
+            target_service=docs_service_provider,
+        )
+    )
+    token_exchange_models.ScopeGrant.objects.get_or_create(
+        rule=people_to_docs_rule,
+        source_scope="openid",
+        granted_scope="openid",
+    )
+
+    simple_action, _created = token_exchange_models.ActionScope.objects.get_or_create(
+        name="action:create-docs-for-groups",
+    )
+    token_exchange_models.ActionScopeGrant.objects.get_or_create(
+        action=simple_action,
+        target_service=docs_service_provider,
+        granted_scope="docs:write",
+    )
+    token_exchange_models.ActionScopeGrant.objects.get_or_create(
+        action=simple_action,
+        target_service=people_service_provider,
+        granted_scope="teams:read",
+    )
+    token_exchange_models.TokenExchangeActionPermission.objects.get_or_create(
+        rule=people_to_docs_rule,
+        action=simple_action,
+    )
+
 
 class Command(BaseCommand):
     """A management command to create a demo database."""

src/backend/demo/templates/demo/token_exchange_form.html

@@ -0,0 +1,175 @@
+<!doctype html>
+<html lang="en">
+<head>
+  <meta charset="utf-8" />
+  <title>Token exchange demo</title>
+  <style>
+    body { font-family: sans-serif; margin: 2rem; max-width: 960px; }
+    form { display: grid; grid-template-columns: 1fr; gap: 0.5rem; }
+    .field { display: flex; flex-direction: column; gap: 0.15rem; }
+    label { font-weight: 600; }
+    input[type="text"], input[type="password"], textarea { padding: 0.35rem; font-size: 0.95rem; }
+    small { color: #555; }
+    button { margin-top: 0.5rem; padding: 0.6rem 1.2rem; font-size: 1rem; }
+    .response { margin-top: 1.5rem; }
+    pre { background: #f4f4f4; padding: 1rem; overflow: auto; }
+    .jwt { margin-top: 1rem; }
+    .jwt h2 { margin: 0 0 0.35rem; font-size: 1rem; }
+    .jwt .row { display: grid; grid-template-columns: 1fr 1fr; gap: 0.75rem; }
+    .hidden { display: none; }
+  </style>
+</head>
+<body>
+  <h1>Token exchange demo</h1>
+  <p>Use this form to POST to <code>/auth/token/exchange/</code>. Basic Auth is built from the client_id and client_secret fields.</p>
+  <form id="token-exchange-form" data-target="{{ token_exchange_url }}">
+    {% csrf_token %}
+    <div class="field">{{ form.client_id.label_tag }}{{ form.client_id }}<small>{{ form.client_id.help_text }}</small></div>
+    <div class="field">{{ form.client_secret.label_tag }}{{ form.client_secret }}<small>{{ form.client_secret.help_text }}</small></div>
+    <div class="field">{{ form.grant_type.label_tag }}{{ form.grant_type }}</div>
+    <div class="field">{{ form.subject_token.label_tag }}{{ form.subject_token }}</div>
+    <div class="field">{{ form.subject_token_type.label_tag }}{{ form.subject_token_type }}</div>
+    <div class="field">{{ form.requested_token_type.label_tag }}{{ form.requested_token_type }}<small>{{ form.requested_token_type.help_text }}</small></div>
+    <div class="field">{{ form.audience.label_tag }}{{ form.audience }}<small>{{ form.audience.help_text }}</small></div>
+    <div class="field">{{ form.scope.label_tag }}{{ form.scope }}<small>{{ form.scope.help_text }}</small></div>
+    <div class="field">{{ form.actor_token.label_tag }}{{ form.actor_token }}</div>
+    <div class="field">{{ form.actor_token_type.label_tag }}{{ form.actor_token_type }}</div>
+    <div class="field">{{ form.resource.label_tag }}{{ form.resource }}</div>
+    <div class="field">{{ form.expires_in.label_tag }}{{ form.expires_in }}<small>{{ form.expires_in.help_text }}</small></div>
+    <button type="submit">Send exchange request</button>
+  </form>
+
+  <div class="response">
+    <div id="response-status"></div>
+    <pre id="exchange-response"></pre>
+  </div>
+
+  <div class="response jwt hidden" id="jwt-section">
+    <h2>Decoded JWT</h2>
+    <div id="jwt-error"></div>
+    <div class="row">
+      <div>
+        <strong>Header</strong>
+        <pre id="jwt-header"></pre>
+      </div>
+      <div>
+        <strong>Payload</strong>
+        <pre id="jwt-payload"></pre>
+      </div>
+    </div>
+  </div>
+
+  <script>
+    (() => {
+      const form = document.getElementById("token-exchange-form");
+      const statusEl = document.getElementById("response-status");
+      const outputEl = document.getElementById("exchange-response");
+      const jwtSection = document.getElementById("jwt-section");
+      const jwtHeaderEl = document.getElementById("jwt-header");
+      const jwtPayloadEl = document.getElementById("jwt-payload");
+      const jwtErrorEl = document.getElementById("jwt-error");
+
+      const resetJwt = () => {
+        if (!jwtSection) return;
+        jwtSection.classList.add("hidden");
+        jwtHeaderEl.textContent = "";
+        jwtPayloadEl.textContent = "";
+        jwtErrorEl.textContent = "";
+      };
+
+      const base64UrlDecode = (segment) => {
+        const normalized = segment.replace(/-/g, "+").replace(/_/g, "/");
+        const padded = normalized + "=".repeat((4 - (normalized.length % 4)) % 4);
+        return atob(padded);
+      };
+
+      const decodeJwt = (token) => {
+        const parts = token.split(".");
+        if (parts.length < 2) {
+          return { error: "Token does not look like a JWT (missing segments)." };
+        }
+        try {
+          return {
+            header: JSON.parse(base64UrlDecode(parts[0])),
+            payload: JSON.parse(base64UrlDecode(parts[1])),
+          };
+        } catch (err) {
+          return { error: `Unable to decode JWT: ${err}` };
+        }
+      };
+
+      const renderJwtIfAny = (responseData, requestedType) => {
+        resetJwt();
+        if (!jwtSection || !responseData || typeof responseData !== "object") return;
+
+        const issuedType = (responseData.issued_token_type || "").toString().toLowerCase();
+        const requested = (requestedType || "").toString().toLowerCase();
+        const wantsJwt = issuedType === "urn:ietf:params:oauth:token-type:jwt" || requested === "urn:ietf:params:oauth:token-type:jwt";
+        const token = responseData.access_token || responseData.id_token || responseData.token;
+        if (!wantsJwt || !token) return;
+
+        const decoded = decodeJwt(token);
+        jwtSection.classList.remove("hidden");
+        if (decoded.error) {
+          jwtErrorEl.textContent = decoded.error;
+          return;
+        }
+
+        jwtHeaderEl.textContent = JSON.stringify(decoded.header, null, 2);
+        jwtPayloadEl.textContent = JSON.stringify(decoded.payload, null, 2);
+      };
+
+      const buildPayload = (formData) => {
+        const body = new URLSearchParams();
+        for (const [key, value] of formData.entries()) {
+          if (!value) continue;
+          body.append(key, value);
+        }
+        return body;
+      };
+
+      form.addEventListener("submit", async (event) => {
+        event.preventDefault();
+
+        statusEl.textContent = "Sending...";
+        outputEl.textContent = "";
+        resetJwt();
+
+        const formData = new FormData(form);
+        const clientId = formData.get("client_id") || "";
+        const clientSecret = formData.get("client_secret") || "";
+
+        formData.delete("client_id");
+        formData.delete("client_secret");
+
+        const headers = { "Content-Type": "application/x-www-form-urlencoded" };
+        if (clientId || clientSecret) {
+          headers.Authorization = `Basic ${btoa(`${clientId}:${clientSecret}`)}`;
+        }
+
+        const response = await fetch(form.dataset.target, {
+          method: "POST",
+          headers,
+          body: buildPayload(formData),
+          credentials: "include",
+        }).catch((error) => {
+          statusEl.textContent = "Network error";
+          outputEl.textContent = error;
+          throw error;
+        });
+
+        const text = await response.text();
+        statusEl.textContent = `Status: ${response.status} ${response.statusText}`;
+
+        try {
+          const parsed = JSON.parse(text);
+          outputEl.textContent = JSON.stringify(parsed, null, 2);
+          renderJwtIfAny(parsed, formData.get("requested_token_type"));
+        } catch (err) {
+          outputEl.textContent = text;
+        }
+      });
+    })();
+  </script>
+</body>
+</html>

src/backend/demo/urls.py

@@ -0,0 +1,15 @@
+"""URL configuration for the demo helpers."""
+
+from django.urls import path
+
+from .views import TokenExchangeDemoView
+
+app_name = "demo"
+
+urlpatterns = [
+    path(
+        "token-exchange/",
+        TokenExchangeDemoView.as_view(),
+        name="token-exchange-demo",
+    ),
+]

src/backend/demo/views.py

@@ -0,0 +1,30 @@
+"""Views for demo application."""
+
+from django.urls import reverse_lazy
+from django.views.generic import TemplateView
+
+from .forms import TokenExchangeDemoForm
+
+
+class TokenExchangeDemoView(TemplateView):
+    """Demo view to help test token exchange flows."""
+
+    template_name = "demo/token_exchange_form.html"
+    form_class = TokenExchangeDemoForm
+    http_method_names = ["get"]
+
+    def get_context_data(self, **kwargs):
+        """Add form and token exchange URL to the context."""
+        context = super().get_context_data(**kwargs)
+        context["form"] = self.form_class(
+            initial={
+                "client_id": "client_id",
+                "client_secret": "client_secret",
+                "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
+                "subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
+                "subject_token": self.request.session.get("oidc_access_token"),
+                "requested_token_type": "urn:ietf:params:oauth:token-type:jwt",
+            }
+        )
+        context["token_exchange_url"] = reverse_lazy("token-exchange")
+        return context

src/backend/people/settings.py

@@ -228,6 +228,7 @@ class Base(Configuration):
     INSTALLED_APPS = [
         # People
         "admin.apps.PeopleAdminConfig",  # replaces 'django.contrib.admin'
+        "token_exchange.apps.TokenExchangeConfig",
         "core",
         "demo",
         "mailbox_manager.apps.MailboxManagerConfig",
@@ -299,6 +300,11 @@ class Base(Configuration):
                 environ_name="BURST_THROTTLE_RATES",
                 environ_prefix=None,
             ),
+            "token_exchange": values.Value(
+                default="20/minute",
+                environ_name="TOKEN_EXCHANGE_THROTTLE_RATES",
+                environ_prefix=None,
+            ),
         },
     }
 
@@ -535,6 +541,58 @@ class Base(Configuration):
         True, environ_name="OIDC_VERIFY_SSL", environ_prefix=None
     )
 
+    # Token Exchange (RFC 8693) settings
+    TOKEN_EXCHANGE_ENABLED = values.BooleanValue(
+        default=False,
+        environ_name="TOKEN_EXCHANGE_ENABLED",
+        environ_prefix=None,
+    )
+    TOKEN_EXCHANGE_MULTI_AUDIENCES_ALLOWED = values.BooleanValue(
+        default=False,
+        environ_name="TOKEN_EXCHANGE_MULTI_AUDIENCES_ALLOWED",
+        environ_prefix=None,
+    )
+    TOKEN_EXCHANGE_DEFAULT_EXPIRES_IN = values.IntegerValue(
+        default=3600,  # 1 hour
+        environ_name="TOKEN_EXCHANGE_DEFAULT_EXPIRES_IN",
+        environ_prefix=None,
+    )
+    TOKEN_EXCHANGE_MAX_EXPIRES_IN = values.IntegerValue(
+        default=86400,  # 24 hours
+        environ_name="TOKEN_EXCHANGE_MAX_EXPIRES_IN",
+        environ_prefix=None,
+    )
+    TOKEN_EXCHANGE_ALLOWED_TOKEN_TYPES = values.ListValue(
+        default=["access_token", "jwt"],
+        environ_name="TOKEN_EXCHANGE_ALLOWED_TOKEN_TYPES",
+        environ_prefix=None,
+    )
+    TOKEN_EXCHANGE_JWT_SIGNING_KEYS = values.DictValue(
+        default={},
+        environ_name="TOKEN_EXCHANGE_JWT_SIGNING_KEYS",
+        environ_prefix=None,
+    )
+    TOKEN_EXCHANGE_JWT_CURRENT_KID = values.Value(
+        default=None,
+        environ_name="TOKEN_EXCHANGE_JWT_CURRENT_KID",
+        environ_prefix=None,
+    )
+    TOKEN_EXCHANGE_JWT_ALGORITHM = values.Value(
+        default="RS256",
+        environ_name="TOKEN_EXCHANGE_JWT_ALGORITHM",
+        environ_prefix=None,
+    )
+    TOKEN_EXCHANGE_MAX_ACTIVE_TOKENS_PER_USER = values.IntegerValue(
+        default=100,
+        environ_name="TOKEN_EXCHANGE_MAX_ACTIVE_TOKENS_PER_USER",
+        environ_prefix=None,
+    )
+    TOKEN_EXCHANGE_ALLOWED_SCHEMES = values.ListValue(
+        default=["http", "https"],
+        environ_name="TOKEN_EXCHANGE_ALLOWED_SCHEMES",
+        environ_prefix=None,
+    )
+
     OIDC_TIMEOUT = values.Value(None, environ_name="OIDC_TIMEOUT", environ_prefix=None)
     OIDC_FALLBACK_TO_EMAIL_FOR_IDENTIFICATION = values.BooleanValue(
         default=True,
@@ -542,6 +600,21 @@ class Base(Configuration):
         environ_prefix=None,
     )
 
+    # required to call a resource server
+    OIDC_STORE_ACCESS_TOKEN = values.BooleanValue(
+        default=False,
+        environ_name="OIDC_STORE_ACCESS_TOKEN",
+        environ_prefix=None,
+    )
+    OIDC_STORE_REFRESH_TOKEN = values.BooleanValue(
+        default=False, environ_name="OIDC_STORE_REFRESH_TOKEN", environ_prefix=None
+    )
+    OIDC_STORE_REFRESH_TOKEN_KEY = values.Value(
+        default=None,
+        environ_name="OIDC_STORE_REFRESH_TOKEN_KEY",
+        environ_prefix=None,
+    )
+
     ACCOUNT_SERVICE_SCOPES = values.ListValue(
         default=[],
         environ_name="ACCOUNT_SERVICE_SCOPES",
@@ -904,6 +977,7 @@ class Test(Base):
     MAIL_PROVISIONING_API_CREDENTIALS = "bGFfcmVnaWU6cGFzc3dvcmQ="
 
     OIDC_ORGANIZATION_REGISTRATION_ID_FIELD = "siret"
+    OIDC_RS_BACKEND_CLASS = "lasuite.oidc_resource_server.backend.ResourceServerBackend"
 
     ORGANIZATION_REGISTRATION_ID_VALIDATORS = [
         {

src/backend/people/urls.py

@@ -26,6 +26,7 @@ urlpatterns = (
     [
         path("admin/", admin.site.urls),
         path("o/", include(oauth2_urls)),
+        path("auth/", include("token_exchange.urls")),
     ]
     + api_urls.urlpatterns
     + resource_server_urls.urlpatterns
@@ -40,6 +41,12 @@ if settings.DEBUG:
         + debug_urls.urlpatterns
     )
 
+    # We double-check here just in case
+    if settings.ENVIRONMENT != "production":
+        urlpatterns += [
+            path("demo/", include("demo.urls"))
+        ]
+
 if settings.USE_SWAGGER or settings.DEBUG:
     urlpatterns += [
         path(

src/backend/pyproject.toml

@@ -84,6 +84,7 @@ dev = [
     "pytest==9.0.2",
     "pytest-icdiff==0.9",
     "pytest-xdist==3.8.0",
+    "pytest-unordered==0.7.0",
     "responses==0.25.8",
     "ruff==0.15.0",
     "types-requests==2.32.4.20260107",

src/backend/token_exchange/__init__.py

@@ -0,0 +1 @@
+"""Token Exchange application for RFC 8693."""

src/backend/token_exchange/admin.py

@@ -0,0 +1,355 @@
+"""Admin configuration for the token_exchange application."""
+
+from django.contrib import admin
+from django.utils.html import format_html
+from django.utils.translation import gettext_lazy as _
+
+from .models import (
+    ActionScope,
+    ActionScopeGrant,
+    ExchangedToken,
+    ScopeGrant,
+    TokenExchangeActionPermission,
+    TokenExchangeRule,
+)
+
+
+@admin.register(ExchangedToken)
+class ExchangedTokenAdmin(admin.ModelAdmin):
+    """Admin interface for ExchangedToken model."""
+
+    list_display = [
+        "token_display",
+        "subject_sub",
+        "subject_email",
+        "token_type",
+        "audiences_display",
+        "scope_display",
+        "expires_at",
+        "revoked_at",
+        "is_valid_display",
+        "created_at",
+    ]
+
+    list_filter = [
+        "token_type",
+        ("revoked_at", admin.EmptyFieldListFilter),
+        "expires_at",
+    ]
+
+    search_fields = [
+        "token",
+        "subject_sub",
+        "subject_email",
+        "subject_token_jti",
+        "audiences",
+    ]
+
+    readonly_fields = [
+        "token",
+        "jwt_kid",
+        "actor_token",
+        "may_act",
+        "subject_token_jti",
+        "subject_token_scope",
+        "created_at",
+        "is_valid_display",
+    ]
+
+    date_hierarchy = "created_at"
+
+    fieldsets = [
+        (
+            _("Token Information"),
+            {
+                "fields": [
+                    "token",
+                    "token_type",
+                    "jwt_kid",
+                ]
+            },
+        ),
+        (
+            _("User & Scopes"),
+            {
+                "fields": [
+                    "subject_sub",
+                    "subject_email",
+                    "audiences",
+                    "scope",
+                    "subject_token_scope",
+                ]
+            },
+        ),
+        (
+            _("Validity"),
+            {
+                "fields": [
+                    "expires_at",
+                    "revoked_at",
+                    "is_valid_display",
+                    "created_at",
+                ]
+            },
+        ),
+        (
+            _("Advanced (RFC 8693)"),
+            {
+                "fields": [
+                    "actor_token",
+                    "may_act",
+                    "subject_token_jti",
+                ],
+                "classes": ["collapse"],
+            },
+        ),
+    ]
+
+    actions = ["revoke_selected_tokens"]
+
+    def has_add_permission(self, request):
+        """Disable manual creation - tokens should only be created via API."""
+        return False
+
+    def token_display(self, obj):
+        """Display truncated token."""
+        if len(obj.token) > 50:
+            return f"{obj.token[:50]}..."
+        return obj.token
+
+    token_display.short_description = _("Token")
+
+    def audiences_display(self, obj):
+        """Display audiences as comma-separated list."""
+        if not obj.audiences:
+            return "-"
+        return ", ".join(obj.audiences)
+
+    audiences_display.short_description = _("Audiences")
+
+    def scope_display(self, obj):
+        """Display scope truncated."""
+        if not obj.scope:
+            return "-"
+        if len(obj.scope) > 50:
+            return f"{obj.scope[:50]}..."
+        return obj.scope
+
+    scope_display.short_description = _("Scope")
+
+    def is_valid_display(self, obj):
+        """Display validity status with icon."""
+        if obj.is_valid():
+            return format_html(
+                '<img src="/static/admin/img/icon-yes.svg" alt="Valid"> Valid'
+            )
+        if obj.is_revoked():
+            return format_html(
+                '<img src="/static/admin/img/icon-no.svg" alt="Revoked"> Revoked'
+            )
+
+        return format_html(
+            '<img src="/static/admin/img/icon-no.svg" alt="Expired"> Expired'
+        )
+
+    is_valid_display.short_description = _("Status")
+
+    @admin.action(description=_("Revoke selected tokens"))
+    def revoke_selected_tokens(self, request, queryset):
+        """Bulk action to revoke selected tokens."""
+        count = 0
+        for token in queryset:
+            if not token.is_revoked():
+                token.revoke()
+                count += 1
+
+        self.message_user(
+            request,
+            _(f"{count} token(s) have been revoked."),
+        )
+
+
+class ScopeGrantInline(admin.TabularInline):
+    """Inline admin for ScopeGrant."""
+
+    model = ScopeGrant
+    extra = 1
+    fields = ["source_scope", "granted_scope", "throttle_rate"]
+
+
+class TokenExchangeActionPermissionInline(admin.TabularInline):
+    """Inline admin for TokenExchangeActionPermission."""
+
+    model = TokenExchangeActionPermission
+    extra = 1
+    fields = ["action", "required_source_scope"]
+    autocomplete_fields = ["action"]
+
+
+@admin.register(TokenExchangeRule)
+class TokenExchangeRuleAdmin(admin.ModelAdmin):
+    """Admin interface for TokenExchangeRule model."""
+
+    list_display = [
+        "id",
+        "source_service",
+        "target_service",
+        "exchanged_token_duration",
+        "is_active",
+        "created_at",
+    ]
+
+    list_filter = [
+        "is_active",
+        "created_at",
+    ]
+
+    search_fields = [
+        "source_service__name",
+        "target_service__name",
+    ]
+
+    autocomplete_fields = ["source_service", "target_service"]
+
+    inlines = [ScopeGrantInline, TokenExchangeActionPermissionInline]
+
+    fieldsets = [
+        (
+            _("Services"),
+            {
+                "fields": [
+                    "source_service",
+                    "target_service",
+                ]
+            },
+        ),
+        (
+            _("Configuration"),
+            {
+                "fields": [
+                    "exchanged_token_duration",
+                    "is_active",
+                ]
+            },
+        ),
+    ]
+
+
+@admin.register(ScopeGrant)
+class ScopeGrantAdmin(admin.ModelAdmin):
+    """Admin interface for ScopeGrant model."""
+
+    list_display = [
+        "id",
+        "rule",
+        "source_scope",
+        "granted_scope",
+        "throttle_rate",
+    ]
+
+    list_filter = [
+        "rule__source_service",
+        "rule__target_service",
+    ]
+
+    search_fields = [
+        "source_scope",
+        "granted_scope",
+        "rule__source_service__name",
+        "rule__target_service__name",
+    ]
+
+    autocomplete_fields = ["rule"]
+
+
+class ActionScopeGrantInline(admin.TabularInline):
+    """Inline admin for ActionScopeGrant."""
+
+    model = ActionScopeGrant
+    extra = 1
+    fields = ["target_service", "granted_scope", "throttle_rate"]
+    autocomplete_fields = ["target_service"]
+
+
+@admin.register(ActionScope)
+class ActionScopeAdmin(admin.ModelAdmin):
+    """Admin interface for ActionScope model."""
+
+    list_display = [
+        "id",
+        "name",
+        "description",
+        "created_at",
+    ]
+
+    search_fields = [
+        "name",
+        "description",
+    ]
+
+    inlines = [ActionScopeGrantInline]
+
+    fieldsets = [
+        (
+            None,
+            {
+                "fields": [
+                    "name",
+                    "description",
+                ]
+            },
+        ),
+    ]
+
+
+@admin.register(ActionScopeGrant)
+class ActionScopeGrantAdmin(admin.ModelAdmin):
+    """Admin interface for ActionScopeGrant model."""
+
+    list_display = [
+        "id",
+        "action",
+        "target_service",
+        "granted_scope",
+        "throttle_rate",
+    ]
+
+    list_filter = [
+        "target_service",
+        "action",
+    ]
+
+    search_fields = [
+        "action__name",
+        "target_service__name",
+        "granted_scope",
+    ]
+
+    autocomplete_fields = ["action", "target_service"]
+
+
+@admin.register(TokenExchangeActionPermission)
+class TokenExchangeActionPermissionAdmin(admin.ModelAdmin):
+    """Admin interface for TokenExchangeActionPermission model."""
+
+    list_display = [
+        "id",
+        "rule",
+        "action",
+        "required_source_scope",
+    ]
+
+    list_filter = [
+        "rule__source_service",
+        "rule__target_service",
+        "action",
+    ]
+
+    search_fields = [
+        "action__name",
+        "rule__source_service__name",
+        "rule__target_service__name",
+        "required_source_scope",
+    ]
+
+    autocomplete_fields = ["rule", "action"]

src/backend/token_exchange/apps.py

@@ -0,0 +1,22 @@
+"""App configuration for the token_exchange application."""
+
+import logging
+
+from django.apps import AppConfig
+
+logger = logging.getLogger(__name__)
+
+
+class TokenExchangeConfig(AppConfig):
+    """Configuration for the token_exchange application."""
+
+    name = "token_exchange"
+    verbose_name = "Token Exchange (RFC 8693)"
+
+    def ready(self):
+        """Import tasks when app is ready."""
+        # Import tasks to register periodic tasks
+        try:
+            import token_exchange.tasks  # noqa: PLC0415 # pylint: disable=unused-import, import-outside-toplevel
+        except ImportError:
+            logger.error("Failed to import tasks for token_exchange")

src/backend/token_exchange/authentication.py

@@ -0,0 +1,40 @@
+"""Authentication backends for the token exchanges API endpoints."""
+
+from rest_framework.authentication import BasicAuthentication
+from rest_framework.exceptions import AuthenticationFailed
+
+from . import models
+
+
+class ServiceProviderBasicAuthentication(BasicAuthentication):
+    """
+    Authentication backend for the token exchanges API endpoints.
+
+    A Service Provider can authenticate to have access to the endpoints.
+    """
+
+    www_authenticate_realm = "token-exchange"
+
+    def authenticate_credentials(self, client_id, client_secret, request=None):
+        """
+        Authenticate a service provider.
+        """
+        if client_id is None or client_secret is None:
+            raise AuthenticationFailed("No credentials provided.")
+
+        try:
+            credentials = models.ServiceProviderCredentials.objects.select_related(
+                "service_provider"
+            ).get(
+                client_id=client_id,
+                client_secret=client_secret,
+            )
+        except models.ServiceProviderCredentials.DoesNotExist as exc:
+            raise AuthenticationFailed("Service provider does not exist.") from exc
+
+        if not credentials.is_active:
+            raise AuthenticationFailed("Service provider inactive or deleted.")
+
+        credentials.service_provider.is_authenticated = True
+
+        return (credentials.service_provider, None)

src/backend/token_exchange/factories.py

@@ -0,0 +1,105 @@
+"""Factories for creating token exchange related model instances for testing."""
+
+import factory.fuzzy
+from faker import Faker
+from pytz import UTC
+
+from . import models
+
+fake = Faker()
+
+
+class ExchangedTokenFactory(factory.django.DjangoModelFactory):
+    """A factory to create ExchangedToken instances for testing."""
+
+    class Meta:
+        model = models.ExchangedToken
+
+    token = factory.LazyFunction(lambda: fake.uuid4())
+    token_type = factory.fuzzy.FuzzyChoice(models.TokenTypeChoices.values)
+    jwt_kid = None
+    subject_sub = factory.LazyFunction(lambda: fake.uuid4())
+    subject_email = factory.LazyFunction(lambda: fake.email())
+    audiences = factory.LazyFunction(lambda: [fake.domain_word() for _ in range(3)])
+    scope = "openid email profile"
+    grants = None
+    expires_at = factory.LazyFunction(
+        lambda: fake.date_time_between(start_date="+1d", end_date="+30d", tzinfo=UTC)
+    )
+    subject_token_jti = factory.LazyFunction(lambda: fake.uuid4())
+    subject_token_scope = "openid email profile"  # noqa: S105
+
+
+class ExpiredExchangedTokenFactory(ExchangedTokenFactory):
+    """A factory to create expired ExchangedToken instances for testing."""
+
+    expires_at = factory.LazyFunction(
+        lambda: fake.date_time_between(start_date="-30d", end_date="-1d", tzinfo=UTC)
+    )
+
+
+class ServiceProviderCredentialsFactory(factory.django.DjangoModelFactory):
+    """Allow a ServiceProvider to authenticate against the token exchange endpoint."""
+
+    class Meta:
+        model = models.ServiceProviderCredentials
+
+    service_provider = factory.SubFactory("core.factories.ServiceProviderFactory")
+    client_id = factory.LazyFunction(lambda: fake.uuid4())
+    client_secret = factory.LazyFunction(lambda: fake.uuid4())
+
+
+class TokenExchangeRuleFactory(factory.django.DjangoModelFactory):
+    """Factory for creating TokenExchangeRule instances."""
+
+    class Meta:
+        model = models.TokenExchangeRule
+
+    source_service = factory.SubFactory("core.factories.ServiceProviderFactory")
+    target_service = factory.SubFactory("core.factories.ServiceProviderFactory")
+    is_active = True
+
+
+class ScopeGrantFactory(factory.django.DjangoModelFactory):
+    """Factory for creating ScopeGrant instances."""
+
+    class Meta:
+        model = models.ScopeGrant
+
+    rule = factory.SubFactory(TokenExchangeRuleFactory)
+    source_scope = factory.LazyFunction(lambda: fake.word())
+    granted_scope = factory.LazyFunction(lambda: fake.word())
+    throttle_rate = None
+
+
+class ActionScopeFactory(factory.django.DjangoModelFactory):
+    """Factory for creating ActionScope instances."""
+
+    class Meta:
+        model = models.ActionScope
+
+    name = factory.Sequence(lambda n: f"action_{n}")
+    description = factory.LazyFunction(lambda: fake.sentence())
+
+
+class ActionScopeGrantFactory(factory.django.DjangoModelFactory):
+    """Factory for creating ActionScopeGrant instances."""
+
+    class Meta:
+        model = models.ActionScopeGrant
+
+    action = factory.SubFactory(ActionScopeFactory)
+    target_service = factory.SubFactory("core.factories.ServiceProviderFactory")
+    granted_scope = factory.LazyFunction(lambda: fake.word())
+    throttle_rate = None
+
+
+class TokenExchangeActionPermissionFactory(factory.django.DjangoModelFactory):
+    """Factory for creating TokenExchangeActionPermission instances."""
+
+    class Meta:
+        model = models.TokenExchangeActionPermission
+
+    rule = factory.SubFactory(TokenExchangeRuleFactory)
+    action = factory.SubFactory(ActionScopeFactory)
+    required_source_scope = ""

src/backend/token_exchange/management/__init__.py

@@ -0,0 +1 @@
+"""Auth management module."""

src/backend/token_exchange/management/commands/__init__.py

@@ -0,0 +1 @@
+"""Auth management commands module."""

src/backend/token_exchange/management/commands/cleanup_expired_tokens.py

@@ -0,0 +1,74 @@
+"""Management command to clean up expired exchanged tokens."""
+
+import logging
+from datetime import timedelta
+
+from django.core.management.base import BaseCommand
+from django.utils import timezone
+
+from token_exchange.models import ExchangedToken
+
+logger = logging.getLogger(__name__)
+
+
+class Command(BaseCommand):
+    """Management command to clean up expired exchanged tokens."""
+
+    help = "Clean up expired exchanged tokens"
+
+    def add_arguments(self, parser):
+        """Add command arguments."""
+        parser.add_argument(
+            "--days",
+            type=int,
+            default=7,
+            help="Delete tokens expired for more than N days (default: 7)",
+        )
+        parser.add_argument(
+            "--dry-run",
+            action="store_true",
+            help="Simulate the cleanup without actually deleting tokens",
+        )
+
+    def handle(self, *args, **options):
+        """Execute the command."""
+        days = options["days"]
+        dry_run = options["dry_run"]
+
+        # Calculate the threshold date
+        threshold = timezone.now() - timedelta(days=days)
+
+        # Find expired tokens older than the threshold
+        expired_tokens = ExchangedToken.objects.filter(expires_at__lt=threshold)
+        count = expired_tokens.count()
+
+        if count == 0:
+            self.stdout.write(
+                self.style.SUCCESS(f"No expired tokens found older than {days} days")
+            )
+            return
+
+        if dry_run:
+            self.stdout.write(
+                self.style.WARNING(
+                    f"[DRY RUN] Would delete {count} expired token(s) older than {days} days"
+                )
+            )
+            logger.info(
+                "[DRY RUN] Would clean up %d expired tokens older than %d days",
+                count,
+                days,
+            )
+        else:
+            # Delete the tokens
+            expired_tokens.delete()
+            self.stdout.write(
+                self.style.SUCCESS(
+                    f"Successfully deleted {count} expired token(s) older than {days} days"
+                )
+            )
+            logger.info(
+                "Cleaned up %d expired tokens older than %d days",
+                count,
+                days,
+            )

src/backend/token_exchange/management/commands/rotate_jwt_key.py

@@ -0,0 +1,115 @@
+"""Management command to rotate JWT signing keys for token exchange."""
+
+import logging
+
+from django.core.management.base import BaseCommand
+
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import rsa
+
+logger = logging.getLogger(__name__)
+
+
+class Command(BaseCommand):
+    """Management command to generate a new JWT signing key."""
+
+    help = "Generate a new JWT signing key for token exchange"
+
+    def add_arguments(self, parser):
+        """Add command arguments."""
+        parser.add_argument(
+            "--kid",
+            type=str,
+            required=True,
+            help="Key ID for the new signing key (e.g., 'key-2024-12')",
+        )
+        parser.add_argument(
+            "--set-current",
+            action="store_true",
+            help="Indicate that this should be set as the current signing key",
+        )
+
+    def handle(self, *args, **options):
+        """Execute the command."""
+        kid = options["kid"]
+        set_current = options["set_current"]
+
+        self.stdout.write(
+            self.style.WARNING(f"\nGenerating new RSA key pair with kid='{kid}'...\n")
+        )
+
+        # Generate RSA key pair (2048 bits, exponent 65537)
+        private_key = rsa.generate_private_key(
+            public_exponent=65537,
+            key_size=2048,
+            backend=default_backend(),
+        )
+
+        # Serialize private key to PEM format
+        private_pem = private_key.private_bytes(
+            encoding=serialization.Encoding.PEM,
+            format=serialization.PrivateFormat.PKCS8,
+            encryption_algorithm=serialization.NoEncryption(),
+        ).decode("utf-8")
+
+        # Extract public key
+        public_key = private_key.public_key()
+        public_pem = public_key.public_bytes(
+            encoding=serialization.Encoding.PEM,
+            format=serialization.PublicFormat.SubjectPublicKeyInfo,
+        ).decode("utf-8")
+
+        # Display the keys
+        self.stdout.write(self.style.SUCCESS("\n✓ Key pair generated successfully!\n"))
+
+        self.stdout.write(
+            self.style.WARNING("=== PUBLIC KEY (distribute to consumers) ===")
+        )
+        self.stdout.write(public_pem)
+
+        self.stdout.write(self.style.WARNING("\n=== PRIVATE KEY (keep secret!) ==="))
+        self.stdout.write(private_pem)
+
+        # Display configuration instructions
+        self.stdout.write(self.style.SUCCESS("\n=== CONFIGURATION INSTRUCTIONS ===\n"))
+
+        self.stdout.write(
+            "1. Add this key to TOKEN_EXCHANGE_JWT_SIGNING_KEYS in your settings or environment:\n"
+        )
+
+        # Escape the private key for JSON/env var
+        private_pem_escaped = private_pem.replace("\n", "\\n")
+
+        self.stdout.write(
+            f'   TOKEN_EXCHANGE_JWT_SIGNING_KEYS={{"{kid}": "{private_pem_escaped}"}}\n'
+        )
+
+        self.stdout.write(
+            "   Note: If you already have keys, add this to the existing dictionary:\n"
+        )
+        self.stdout.write(
+            f'   {{"existing-kid": "...", "{kid}": "{private_pem_escaped}"}}\n'
+        )
+
+        if set_current:
+            self.stdout.write(f"\n2. Set this as the current signing key:\n")
+            self.stdout.write(f"   TOKEN_EXCHANGE_JWT_CURRENT_KID={kid}\n")
+        else:
+            self.stdout.write(f"\n2. To use this key for new tokens, set:\n")
+            self.stdout.write(f"   TOKEN_EXCHANGE_JWT_CURRENT_KID={kid}\n")
+
+        self.stdout.write(
+            "\n3. IMPORTANT: Keep old keys in TOKEN_EXCHANGE_JWT_SIGNING_KEYS to validate existing tokens!\n"
+        )
+
+        self.stdout.write(
+            "   Do NOT remove old keys immediately. They are needed to verify tokens signed with those keys.\n"
+        )
+
+        # Log the key generation
+        logger.info("Generated new JWT key with kid=%s", kid)
+
+        self.stdout.write(
+            self.style.SUCCESS(f"\n✓ Key rotation process completed for kid='{kid}'\n")
+        )

src/backend/token_exchange/migrations/0001_initial.py

@@ -0,0 +1,151 @@
+# Generated by Django 5.2.9 on 2026-02-04 14:19
+
+import datetime
+import django.contrib.postgres.fields
+import django.db.models.deletion
+import token_exchange.models
+import uuid
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+        ('core', '0017_teamwebhook_protocol'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='ActionScope',
+            fields=[
+                ('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')),
+                ('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created at')),
+                ('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated at')),
+                ('name', models.CharField(help_text="Unique identifier for this action scope, starts with 'action:'", max_length=255, unique=True, validators=[token_exchange.models.validate_action_scope_name], verbose_name='Action Name')),
+                ('description', models.TextField(blank=True, help_text='Human-readable description of this action', verbose_name='Description')),
+            ],
+            options={
+                'verbose_name': 'Action Scope',
+                'verbose_name_plural': 'Action Scopes',
+                'db_table': 'token_exchange_action_scope',
+            },
+        ),
+        migrations.CreateModel(
+            name='ExchangedToken',
+            fields=[
+                ('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')),
+                ('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated at')),
+                ('token', models.CharField(db_index=True, help_text='The actual token string (opaque or JWT)', unique=True, verbose_name='Token')),
+                ('token_type', models.CharField(choices=[('access_token', 'Access Token'), ('jwt', 'JWT')], default='access_token', help_text='The type of token as per RFC 8693', max_length=50, verbose_name='Token Type')),
+                ('jwt_kid', models.CharField(blank=True, db_index=True, help_text='The key ID used to sign the JWT token', max_length=255, null=True, verbose_name='JWT Key ID')),
+                ('subject_sub', models.CharField(blank=True, db_index=True, help_text='The subject identifier (sub) from the introspection', max_length=255, null=True, verbose_name='Subject Sub')),
+                ('subject_email', models.EmailField(blank=True, db_index=True, help_text='The subject email from the introspection', max_length=255, null=True, verbose_name='Subject Email')),
+                ('audiences', django.contrib.postgres.fields.ArrayField(base_field=models.CharField(max_length=255), default=list, help_text='The audiences for this token (RFC 8693 supports multiple)', size=None, verbose_name='Audiences')),
+                ('scope', models.TextField(blank=True, default='', help_text='Space-separated scopes for this token', verbose_name='Scope')),
+                ('grants', models.JSONField(blank=True, default=list, help_text='List of grant objects with scope and throttling information', null=True, verbose_name='Grants')),
+                ('expires_at', models.DateTimeField(db_index=True, help_text='When this token expires', verbose_name='Expires At')),
+                ('revoked_at', models.DateTimeField(blank=True, db_index=True, help_text='When this token was revoked (null if not revoked)', null=True, verbose_name='Revoked At')),
+                ('actor_token', models.TextField(blank=True, help_text='The actor token for delegation (RFC 8693)', null=True, verbose_name='Actor Token')),
+                ('may_act', models.JSONField(blank=True, help_text='The may_act claim for delegation (RFC 8693)', null=True, verbose_name='May Act')),
+                ('subject_token_jti', models.CharField(db_index=True, help_text='The JTI of the original SSO token for traceability', max_length=255, verbose_name='Subject Token JTI')),
+                ('subject_token_scope', models.TextField(blank=True, default='', help_text='The original scopes from the SSO token', verbose_name='Subject Token Scope')),
+                ('created_at', models.DateTimeField(auto_now_add=True, help_text='When this token was created', verbose_name='Created At')),
+            ],
+            options={
+                'verbose_name': 'Exchanged Token',
+                'verbose_name_plural': 'Exchanged Tokens',
+                'db_table': 'token_exchange_exchanged_token',
+                'ordering': ['-created_at'],
+                'indexes': [models.Index(fields=['subject_sub', 'created_at'], name='token_excha_subject_633725_idx'), models.Index(fields=['subject_email', 'created_at'], name='token_excha_subject_c5c60f_idx'), models.Index(fields=['expires_at', 'revoked_at'], name='token_excha_expires_7f22e1_idx')],
+            },
+        ),
+        migrations.CreateModel(
+            name='ServiceProviderCredentials',
+            fields=[
+                ('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')),
+                ('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created at')),
+                ('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated at')),
+                ('client_id', models.CharField(db_index=True, max_length=255, unique=True, verbose_name='Client ID')),
+                ('client_secret', models.CharField(db_index=True, max_length=255, unique=True, verbose_name='Client secret')),
+                ('allowed_origins', models.TextField(blank=True, default='', help_text='Allowed origins list to enable CORS, space separated')),
+                ('is_active', models.BooleanField(default=True)),
+                ('service_provider', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='token_exchange_credentials', to='core.serviceprovider')),
+            ],
+            options={
+                'abstract': False,
+            },
+        ),
+        migrations.CreateModel(
+            name='TokenExchangeRule',
+            fields=[
+                ('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')),
+                ('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created at')),
+                ('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated at')),
+                ('exchanged_token_duration', models.DurationField(blank=True, default=datetime.timedelta(seconds=300), help_text='Duration of the generated token')),
+                ('is_active', models.BooleanField(default=True, verbose_name='Is Active')),
+                ('source_service', models.ForeignKey(help_text='Service provider initiating the token exchange', on_delete=django.db.models.deletion.CASCADE, related_name='exchange_out', to='core.serviceprovider', verbose_name='Source Service')),
+                ('target_service', models.ForeignKey(help_text='Service provider receiving the exchanged token', on_delete=django.db.models.deletion.CASCADE, related_name='exchange_in', to='core.serviceprovider', verbose_name='Target Service')),
+            ],
+            options={
+                'verbose_name': 'Token Exchange Rule',
+                'verbose_name_plural': 'Token Exchange Rules',
+                'db_table': 'token_exchange_rule',
+                'unique_together': {('source_service', 'target_service')},
+            },
+        ),
+        migrations.CreateModel(
+            name='ActionScopeGrant',
+            fields=[
+                ('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')),
+                ('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created at')),
+                ('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated at')),
+                ('granted_scope', models.CharField(help_text='Scope granted on the target service when action is used', max_length=255, verbose_name='Granted Scope')),
+                ('throttle_rate', models.CharField(blank=True, help_text="Optional throttle rate (e.g., '5/h', '100/day')", max_length=255, null=True, verbose_name='Throttle Rate')),
+                ('action', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='grants', to='token_exchange.actionscope', verbose_name='Action Scope')),
+                ('target_service', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='action_grants', to='core.serviceprovider', verbose_name='Target Service')),
+            ],
+            options={
+                'verbose_name': 'Action Scope Grant',
+                'verbose_name_plural': 'Action Scope Grants',
+                'db_table': 'token_exchange_action_scope_grant',
+                'unique_together': {('action', 'target_service', 'granted_scope')},
+            },
+        ),
+        migrations.CreateModel(
+            name='TokenExchangeActionPermission',
+            fields=[
+                ('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')),
+                ('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created at')),
+                ('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated at')),
+                ('required_source_scope', models.CharField(blank=True, help_text='Scope required in source token to use this action (empty = no requirement)', max_length=255, verbose_name='Required Source Scope')),
+                ('action', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='permissions', to='token_exchange.actionscope', verbose_name='Action Scope')),
+                ('rule', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='action_permissions', to='token_exchange.tokenexchangerule', verbose_name='Exchange Rule')),
+            ],
+            options={
+                'verbose_name': 'Token Exchange Action Permission',
+                'verbose_name_plural': 'Token Exchange Action Permissions',
+                'db_table': 'token_exchange_action_permission',
+                'unique_together': {('rule', 'action')},
+            },
+        ),
+        migrations.CreateModel(
+            name='ScopeGrant',
+            fields=[
+                ('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')),
+                ('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created at')),
+                ('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated at')),
+                ('source_scope', models.CharField(help_text='Scope from the source token', max_length=255, verbose_name='Source Scope')),
+                ('granted_scope', models.CharField(help_text='Scope granted on the target service', max_length=255, verbose_name='Granted Scope')),
+                ('throttle_rate', models.CharField(blank=True, help_text="Optional throttle rate (e.g., '5/h', '100/day')", max_length=255, null=True, verbose_name='Throttle Rate')),
+                ('rule', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='scope_grants', to='token_exchange.tokenexchangerule', verbose_name='Exchange Rule')),
+            ],
+            options={
+                'verbose_name': 'Scope Grant',
+                'verbose_name_plural': 'Scope Grants',
+                'db_table': 'token_exchange_scope_grant',
+                'unique_together': {('rule', 'source_scope', 'granted_scope')},
+            },
+        ),
+    ]

src/backend/token_exchange/migrations/__init__.py

@@ -0,0 +1,2 @@
+"""Auth migrations module."""
+

src/backend/token_exchange/models.py

@@ -0,0 +1,522 @@
+"""Models for the auth application."""
+
+import datetime
+import logging
+
+from django.conf import settings
+from django.contrib.auth import get_user_model
+from django.contrib.postgres.fields import ArrayField
+from django.core.exceptions import ValidationError
+from django.core.validators import URLValidator
+from django.db import models
+from django.db.models import JSONField
+from django.utils import timezone
+from django.utils.translation import gettext_lazy as _
+
+from core.models import BaseModel, ServiceProvider
+
+logger = logging.getLogger(__name__)
+
+User = get_user_model()
+
+
+def validate_action_scope_name(value):
+    if not value.startswith("action:"):
+        raise ValidationError(_("Action name must start with 'action:'"))
+
+
+class TokenTypeChoices(models.TextChoices):
+    """Token type choices for RFC 8693."""
+
+    ACCESS_TOKEN = "access_token", _("Access Token")
+    JWT = "jwt", _("JWT")
+
+
+class ServiceProviderCredentials(BaseModel):
+    """
+    Allow to define credentials for a Service Provider.
+
+    We could (should?) inherit from AbstractApplication and swap Application, but
+    for now we prefer to keep thing simple.
+    """
+
+    service_provider = models.ForeignKey(
+        ServiceProvider,
+        on_delete=models.CASCADE,
+        related_name="token_exchange_credentials",
+        blank=False,
+        null=False,
+    )
+    client_id = models.CharField(
+        max_length=255,
+        unique=True,
+        db_index=True,
+        verbose_name=_("Client ID"),
+    )
+    client_secret = models.CharField(  # Should be hashed
+        max_length=255,
+        unique=True,
+        db_index=True,
+        verbose_name=_("Client secret"),
+    )
+
+    allowed_origins = models.TextField(
+        blank=True,
+        help_text=_("Allowed origins list to enable CORS, space separated"),
+        default="",
+    )
+    is_active = models.BooleanField(
+        default=True,
+    )
+
+    def clean(self):
+        """Validate allowed origins URLs."""
+        super().clean()
+
+        if allowed_origins := self.allowed_origins.strip().split():
+            validator = URLValidator(
+                settings.TOKEN_EXCHANGE_ALLOWED_SCHEMES, "allowed origin"
+            )
+            for url in allowed_origins:
+                validator(url)
+
+
+class TokenExchangeRule(BaseModel):
+    """
+    Defines a token exchange rule from one service provider to another.
+
+    This model represents the authorization for a source service to exchange
+    tokens for accessing a target service.
+    """
+
+    source_service = models.ForeignKey(
+        ServiceProvider,
+        related_name="exchange_out",
+        on_delete=models.CASCADE,
+        verbose_name=_("Source Service"),
+        help_text=_("Service provider initiating the token exchange"),
+    )
+    target_service = models.ForeignKey(
+        ServiceProvider,
+        related_name="exchange_in",
+        on_delete=models.CASCADE,
+        verbose_name=_("Target Service"),
+        help_text=_("Service provider receiving the exchanged token"),
+    )
+
+    exchanged_token_duration = models.DurationField(
+        blank=True,
+        default=datetime.timedelta(minutes=5),
+        help_text=_("Duration of the generated token"),
+    )
+
+    is_active = models.BooleanField(
+        default=True,
+        verbose_name=_("Is Active"),
+    )
+
+    class Meta:
+        unique_together = ("source_service", "target_service")
+        db_table = "token_exchange_rule"
+        verbose_name = _("Token Exchange Rule")
+        verbose_name_plural = _("Token Exchange Rules")
+
+    def __str__(self):
+        return f"{self.source_service} → {self.target_service}"
+
+
+class ScopeGrant(BaseModel):
+    """
+    Defines a scope grant within a token exchange rule.
+
+    Maps a source scope to a granted scope on the target service,
+    with optional throttling constraints. Can perform downscoping
+    (reducing privileges) or upscoping (elevating privileges)
+    depending on the configuration.
+    """
+
+    rule = models.ForeignKey(
+        TokenExchangeRule,
+        on_delete=models.CASCADE,
+        related_name="scope_grants",
+        verbose_name=_("Exchange Rule"),
+    )
+
+    source_scope = models.CharField(
+        max_length=255,
+        verbose_name=_("Source Scope"),
+        help_text=_("Scope from the source token"),
+    )
+    granted_scope = models.CharField(
+        max_length=255,
+        verbose_name=_("Granted Scope"),
+        help_text=_("Scope granted on the target service"),
+    )
+
+    throttle_rate = models.CharField(
+        null=True,
+        blank=True,
+        max_length=255,
+        verbose_name=_("Throttle Rate"),
+        help_text=_("Optional throttle rate (e.g., '5/h', '100/day')"),
+    )
+
+    class Meta:
+        db_table = "token_exchange_scope_grant"
+        verbose_name = _("Scope Grant")
+        verbose_name_plural = _("Scope Grants")
+        unique_together = ("rule", "source_scope", "granted_scope")
+
+    def __str__(self):
+        return f"{self.source_scope} → {self.granted_scope} (rule: {self.rule_id})"
+
+
+class ActionScope(BaseModel):
+    """
+    Defines an action scope that can grant access to multiple services.
+
+    An action scope is a special scope that, when requested, can grant
+    access to specific scopes on different target services.
+    """
+
+    name = models.CharField(
+        max_length=255,
+        unique=True,
+        validators=[validate_action_scope_name],
+        verbose_name=_("Action Name"),
+        help_text=_("Unique identifier for this action scope, starts with 'action:'"),
+    )
+    description = models.TextField(
+        blank=True,
+        verbose_name=_("Description"),
+        help_text=_("Human-readable description of this action"),
+    )
+
+    class Meta:
+        db_table = "token_exchange_action_scope"
+        verbose_name = _("Action Scope")
+        verbose_name_plural = _("Action Scopes")
+
+    def __str__(self):
+        return self.name
+
+    def save(self, *args, **kwargs):
+        """Ensure the action name is stored in lowercase."""
+        if self.name:
+            self.name = self.name.lower()
+        super().save(*args, **kwargs)
+
+
+class ActionScopeGrant(BaseModel):
+    """
+    Defines a scope grant for an action on a specific target service.
+
+    This model maps an action to specific scopes on a target service,
+    with optional throttling constraints.
+    """
+
+    action = models.ForeignKey(
+        ActionScope,
+        on_delete=models.CASCADE,
+        related_name="grants",
+        verbose_name=_("Action Scope"),
+    )
+
+    target_service = models.ForeignKey(
+        ServiceProvider,
+        on_delete=models.CASCADE,
+        related_name="action_grants",
+        verbose_name=_("Target Service"),
+    )
+    granted_scope = models.CharField(
+        max_length=255,
+        verbose_name=_("Granted Scope"),
+        help_text=_("Scope granted on the target service when action is used"),
+    )
+
+    throttle_rate = models.CharField(
+        null=True,
+        blank=True,
+        max_length=255,
+        verbose_name=_("Throttle Rate"),
+        help_text=_("Optional throttle rate (e.g., '5/h', '100/day')"),
+    )
+
+    class Meta:
+        db_table = "token_exchange_action_scope_grant"
+        verbose_name = _("Action Scope Grant")
+        verbose_name_plural = _("Action Scope Grants")
+        unique_together = ("action", "target_service", "granted_scope")
+
+    def __str__(self):
+        return f"{self.action.name} → {self.granted_scope} on {self.target_service}"
+
+
+class TokenExchangeActionPermission(BaseModel):
+    """
+    Defines permissions for using an action within a token exchange rule.
+
+    This model specifies which actions can be used within a token exchange
+    and what source scopes are required to use them.
+    """
+
+    rule = models.ForeignKey(
+        TokenExchangeRule,
+        on_delete=models.CASCADE,
+        related_name="action_permissions",
+        verbose_name=_("Exchange Rule"),
+    )
+
+    action = models.ForeignKey(
+        ActionScope,
+        on_delete=models.CASCADE,
+        related_name="permissions",
+        verbose_name=_("Action Scope"),
+    )
+
+    required_source_scope = models.CharField(
+        max_length=255,
+        blank=True,
+        verbose_name=_("Required Source Scope"),
+        help_text=_(
+            "Scope required in source token to use this action (empty = no requirement)"
+        ),
+    )
+
+    class Meta:
+        db_table = "token_exchange_action_permission"
+        verbose_name = _("Token Exchange Action Permission")
+        verbose_name_plural = _("Token Exchange Action Permissions")
+        unique_together = ("rule", "action")
+
+    def __str__(self):
+        return f"{self.action.name} in rule {self.rule_id}"
+
+
+class ExchangedToken(BaseModel):
+    """
+    Model representing an exchanged token according to RFC 8693.
+
+    This model stores tokens that have been exchanged from an external SSO
+    via the token exchange endpoint. It supports both opaque tokens and
+    signed JWT tokens with key rotation.
+    """
+
+    token = models.CharField(
+        max_length=None,
+        unique=True,
+        db_index=True,
+        verbose_name=_("Token"),
+        help_text=_("The actual token string (opaque or JWT)"),
+    )
+    token_type = models.CharField(
+        max_length=50,
+        choices=TokenTypeChoices.choices,
+        default=TokenTypeChoices.ACCESS_TOKEN,
+        verbose_name=_("Token Type"),
+        help_text=_("The type of token as per RFC 8693"),
+    )
+    jwt_kid = models.CharField(
+        max_length=255,
+        null=True,
+        blank=True,
+        db_index=True,
+        verbose_name=_("JWT Key ID"),
+        help_text=_("The key ID used to sign the JWT token"),
+    )
+    subject_sub = models.CharField(
+        max_length=255,
+        null=True,
+        blank=True,
+        db_index=True,
+        verbose_name=_("Subject Sub"),
+        help_text=_("The subject identifier (sub) from the introspection"),
+    )
+    subject_email = models.EmailField(
+        max_length=255,
+        null=True,
+        blank=True,
+        db_index=True,
+        verbose_name=_("Subject Email"),
+        help_text=_("The subject email from the introspection"),
+    )
+    audiences = ArrayField(
+        models.CharField(max_length=255),
+        default=list,
+        verbose_name=_("Audiences"),
+        help_text=_("The audiences for this token (RFC 8693 supports multiple)"),
+    )
+    scope = models.TextField(
+        blank=True,
+        default="",
+        verbose_name=_("Scope"),
+        help_text=_("Space-separated scopes for this token"),
+    )
+    grants = JSONField(
+        null=True,
+        blank=True,
+        default=list,
+        verbose_name=_("Grants"),
+        help_text=_("List of grant objects with scope and throttling information"),
+    )
+    expires_at = models.DateTimeField(
+        db_index=True,
+        verbose_name=_("Expires At"),
+        help_text=_("When this token expires"),
+    )
+    revoked_at = models.DateTimeField(
+        null=True,
+        blank=True,
+        db_index=True,
+        verbose_name=_("Revoked At"),
+        help_text=_("When this token was revoked (null if not revoked)"),
+    )
+    actor_token = models.TextField(
+        null=True,
+        blank=True,
+        verbose_name=_("Actor Token"),
+        help_text=_("The actor token for delegation (RFC 8693)"),
+    )
+    may_act = models.JSONField(
+        null=True,
+        blank=True,
+        verbose_name=_("May Act"),
+        help_text=_("The may_act claim for delegation (RFC 8693)"),
+    )
+    subject_token_jti = models.CharField(
+        max_length=255,
+        db_index=True,
+        verbose_name=_("Subject Token JTI"),
+        help_text=_("The JTI of the original SSO token for traceability"),
+    )
+    subject_token_scope = models.TextField(
+        blank=True,
+        default="",
+        verbose_name=_("Subject Token Scope"),
+        help_text=_("The original scopes from the SSO token"),
+    )
+    created_at = models.DateTimeField(
+        auto_now_add=True,
+        verbose_name=_("Created At"),
+        help_text=_("When this token was created"),
+    )
+
+    class Meta:
+        db_table = "token_exchange_exchanged_token"
+        ordering = ["-created_at"]
+        verbose_name = _("Exchanged Token")
+        verbose_name_plural = _("Exchanged Tokens")
+        indexes = [
+            models.Index(fields=["subject_sub", "created_at"]),
+            models.Index(fields=["subject_email", "created_at"]),
+            models.Index(fields=["expires_at", "revoked_at"]),
+        ]
+
+    def __str__(self):
+        """String representation of the token."""
+        identity = self.subject_email or self.subject_sub or "unknown"
+        return f"{self.token_type} for {identity} (expires {self.expires_at})"
+
+    def is_expired(self):
+        """Check if the token has expired."""
+        return timezone.now() >= self.expires_at
+
+    def is_revoked(self):
+        """Check if the token has been revoked."""
+        return self.revoked_at is not None
+
+    def is_valid(self):
+        """Check if the token is valid (not expired and not revoked)."""
+        return not self.is_expired() and not self.is_revoked()
+
+    def revoke(self):
+        """Revoke this token."""
+        if not self.is_revoked():
+            self.revoked_at = timezone.now()
+            self.save(update_fields=["revoked_at", "updated_at"])
+
+    def to_introspection_response(self):
+        """
+        Convert this token to an RFC 7662 introspection response.
+
+        Returns:
+            dict: The introspection response payload
+        """
+        if not self.is_valid():
+            return {"active": False}
+
+        return {
+            "active": True,
+            "scope": self.scope,
+            "username": self.subject_email or self.subject_sub,
+            "token_type": self.token_type,
+            "exp": int(self.expires_at.timestamp()),
+            "iat": int(self.created_at.timestamp()),
+            "sub": self.subject_sub,
+            "email": self.subject_email,
+            "aud": self.audiences,
+            "jti": self._get_jti(),
+            "client_id": settings.OIDC_RS_CLIENT_ID,
+        }
+
+    def _get_jti(self):
+        """Extract or generate a JTI for this token."""
+        if self.token_type == TokenTypeChoices.JWT:
+            # For JWT tokens, we could extract the jti from the token itself
+            # For now, we use a stable identifier
+            return self.subject_token_jti
+        # For opaque tokens, use the token itself as identifier
+        return self.token[:50]  # Truncate for readability
+
+    def save(self, *args, **kwargs):
+        """Override save to enforce token limit per user when creating a new token.
+
+        This ensures that both tokens created through the view and tokens created
+        directly in tests/fixtures will respect the per-user limit.
+        """
+        is_new = self._state.adding
+        super().save(*args, **kwargs)
+
+        if is_new:
+            # Enforce token limit after creation to ensure created_at is set
+            try:
+                max_tokens = settings.TOKEN_EXCHANGE_MAX_ACTIVE_TOKENS_PER_USER
+            except AttributeError:
+                # Setting not present — nothing to enforce
+                return
+
+            if max_tokens is None:
+                return
+
+            # Filter by subject_sub or subject_email
+            if self.subject_sub:
+                active_tokens_qs = ExchangedToken.objects.filter(
+                    subject_sub=self.subject_sub,
+                    expires_at__gt=timezone.now(),
+                    revoked_at__isnull=True,
+                )
+            elif self.subject_email:
+                active_tokens_qs = ExchangedToken.objects.filter(
+                    subject_email=self.subject_email,
+                    expires_at__gt=timezone.now(),
+                    revoked_at__isnull=True,
+                )
+            else:
+                # No identity to group by
+                return
+
+            active_tokens_qs = active_tokens_qs.order_by("created_at")
+
+            active_tokens_count = active_tokens_qs.count()
+            if active_tokens_count > max_tokens:
+                # Number to delete
+                to_delete = active_tokens_count - max_tokens
+                oldest = active_tokens_qs[:to_delete]
+                ids = [t.id for t in oldest]
+                ExchangedToken.objects.filter(id__in=ids).delete()
+                logger.info(
+                    "Enforced token limit for sub=%s/email=%s: deleted %d oldest tokens",
+                    self.subject_sub,
+                    self.subject_email,
+                    to_delete,
+                )

src/backend/token_exchange/permissions.py

@@ -0,0 +1,19 @@
+"""Permissions for token exchange app."""
+
+from rest_framework.permissions import BasePermission
+
+from core.models import ServiceProvider
+
+
+class IsServiceProviderAuthenticated(BasePermission):
+    """
+    Allows access only to authenticated ServiceProvider.
+    """
+
+    def has_permission(self, request, view):
+        """Check if the user is an authenticated ServiceProvider."""
+        return bool(
+            isinstance(request.user, ServiceProvider)
+            and request.user
+            and request.user.is_authenticated
+        )

src/backend/token_exchange/serializers.py

@@ -0,0 +1,142 @@
+"""Serializers for the token_exchange application."""
+
+from django.conf import settings
+
+from rest_framework import serializers
+
+
+class TokenExchangeSerializer(serializers.Serializer):
+    """
+    Serializer for RFC 8693 token exchange requests.
+
+    Validates the token exchange request parameters according to RFC 8693.
+    """
+
+    grant_type = serializers.CharField(required=True)
+    subject_token = serializers.CharField(required=True)
+    subject_token_type = serializers.CharField(required=True)
+
+    requested_token_type = serializers.CharField(required=False, allow_blank=True)
+    audience = serializers.CharField(required=False, allow_blank=True)
+    scope = serializers.CharField(required=False, allow_blank=True)
+    actor_token = serializers.CharField(required=False, allow_blank=True)
+    actor_token_type = serializers.CharField(required=False, allow_blank=True)
+    resource = serializers.CharField(required=False, allow_blank=True)
+    expires_in = serializers.IntegerField(required=False, min_value=1)
+
+    def validate_grant_type(self, value):
+        """Validate that grant_type is the correct RFC 8693 value."""
+        expected = "urn:ietf:params:oauth:grant-type:token-exchange"
+        if value != expected:
+            raise serializers.ValidationError(
+                f"Invalid grant_type. Expected '{expected}', got '{value}'"
+            )
+        return value
+
+    def validate_requested_token_type(self, value):
+        """Validate that the requested token type is allowed."""
+        if not value:
+            return value
+
+        allowed_types = settings.TOKEN_EXCHANGE_ALLOWED_TOKEN_TYPES
+
+        # Map URN types to simple types
+        urn_mapping = {
+            "urn:ietf:params:oauth:token-type:access_token": "access_token",
+            "urn:ietf:params:oauth:token-type:jwt": "jwt",
+            "urn:ietf:params:oauth:token-type:refresh_token": "refresh_token",
+        }
+
+        # Check if it's a URN or simple type
+        simple_type = urn_mapping.get(value, value)
+
+        if simple_type == "refresh_token":
+            raise serializers.ValidationError(
+                "refresh_token type is not supported for token exchange"
+            )
+
+        if simple_type not in allowed_types:
+            raise serializers.ValidationError(
+                f"Token type '{value}' is not allowed. "
+                f"Allowed types: {', '.join(allowed_types)}"
+            )
+
+        return simple_type
+
+    def validate_expires_in(self, value):
+        """Validate that expires_in is within acceptable bounds."""
+        if value is None:
+            return value
+
+        max_expires_in = settings.TOKEN_EXCHANGE_MAX_EXPIRES_IN
+        if value > max_expires_in:
+            raise serializers.ValidationError(
+                f"expires_in must not exceed {max_expires_in} seconds"
+            )
+
+        return value
+
+    def validate_scope(self, value):
+        """Validate and parse scopes."""
+        if not value:
+            return ""
+
+        scopes = [scope.strip() for scope in value.split() if scope.strip()]
+
+        action_count = len([scope for scope in scopes if scope.startswith("action:")])
+        if action_count > 1:
+            raise serializers.ValidationError("Only one action scope is allowed")
+
+        if action_count and len(scopes) != 1:
+            raise serializers.ValidationError(
+                "Actions cannot be combined with other scopes"
+            )
+
+        # Return as-is, will be validated against subject_token_scope in the view
+        return scopes
+
+    def validate_audience(self, value):
+        """Parse multiple audiences separated by spaces."""
+        if not value:
+            return []
+        # Split by spaces and filter empty strings
+        audiences = [aud.strip() for aud in value.split() if aud.strip()]
+        return audiences
+
+    def validate_subject_token_type(self, value):
+        """Validate that subject_token_type is access_token."""
+        expected = "urn:ietf:params:oauth:token-type:access_token"
+        if value != expected:
+            raise serializers.ValidationError(
+                f"Invalid subject_token_type. Expected '{expected}', got '{value}'"
+            )
+        return value
+
+    def validate(self, attrs):
+        """Additional cross-field validation."""
+        # If actor_token is provided, actor_token_type should also be provided
+        if attrs.get("actor_token") and not attrs.get("actor_token_type"):
+            raise serializers.ValidationError(
+                {
+                    "actor_token_type": "actor_token_type is required when actor_token is provided"
+                }
+            )
+
+        return attrs
+
+
+class TokenRevocationSerializer(serializers.Serializer):
+    """
+    Serializer for RFC 7009 token revocation requests.
+
+    Validates the token revocation request parameters.
+    """
+
+    token = serializers.CharField(required=True)
+    token_type_hint = serializers.CharField(required=False, allow_blank=True)
+
+    def validate_token(self, value):
+        """Validate that token is not empty."""
+        if not value or not value.strip():
+            raise serializers.ValidationError("Token cannot be empty")
+        return value.strip()

src/backend/token_exchange/tasks.py

@@ -0,0 +1,26 @@
+"""Celery tasks for the token_exchange application."""
+
+from django.core.management import call_command
+
+from celery import Celery
+from celery.schedules import crontab
+
+from people.celery_app import app as celery_app
+
+
+@celery_app.on_after_finalize.connect
+def setup_periodic_tasks(sender: Celery, **kwargs):
+    """Setup periodic tasks for token_exchange module."""
+    # Daily cleanup of expired tokens at 2:00 AM
+    sender.add_periodic_task(
+        crontab(hour="2", minute="0"),
+        cleanup_expired_tokens_task.s(),
+        name="cleanup_expired_tokens_daily",
+        serializer="json",
+    )
+
+
+@celery_app.task
+def cleanup_expired_tokens_task():
+    """Task to clean up expired exchanged tokens."""
+    call_command("cleanup_expired_tokens")

src/backend/token_exchange/tests/__init__.py

@@ -0,0 +1 @@
+"""Tests for the token_exchange application."""

src/backend/token_exchange/tests/helpers.py

@@ -0,0 +1,57 @@
+"""Helper function to create token exchange rules and grants for tests."""
+
+from token_exchange.factories import (
+    ScopeGrantFactory,
+    TokenExchangeRuleFactory,
+)
+
+
+class DummyBackend:
+    """Dummy backend for mocking resource server introspection in tests."""
+
+    # Class attributes to store test data
+    _test_user_info = None
+    _test_origin = None
+
+    def __init__(self):
+        """Initialize DummyBackend with test data from class attributes."""
+        user_info = self._test_user_info or {}
+        origin = self._test_origin
+        self.user_info = user_info
+        self.token_origin_audience = origin or user_info.get("aud")
+        self._scopes = []
+
+    def get_user_info_with_introspection(self, token):
+        """Return the mocked user info."""
+        return self.user_info
+
+
+def create_token_exchange_rule_with_scopes(source_sp, target_sp, scopes):
+    """
+    Create a TokenExchangeRule with ScopeGrants for the given scopes.
+
+    Each scope in the source token is mapped to the same scope in the target service
+    to quickly build symmetric grants for tests.
+
+    Args:
+        source_sp: Source ServiceProvider
+        target_sp: Target ServiceProvider
+        scopes: List of scope strings
+
+    Returns:
+        The created TokenExchangeRule
+    """
+
+    rule = TokenExchangeRuleFactory(
+        source_service=source_sp,
+        target_service=target_sp,
+    )
+
+    for scope in scopes:
+        ScopeGrantFactory(
+            rule=rule,
+            source_scope=scope,
+            granted_scope=scope,
+        )
+
+    return rule

src/backend/token_exchange/tests/test_cleanup_tokens.py

@@ -0,0 +1,169 @@
+"""Tests for cleanup_expired_tokens management command."""
+
+import logging
+from datetime import timedelta
+from io import StringIO
+from unittest import mock
+
+from django.core.management import call_command
+from django.utils import timezone
+
+import pytest
+
+from token_exchange.factories import ExchangedTokenFactory
+from token_exchange.models import ExchangedToken
+from token_exchange.tasks import cleanup_expired_tokens_task
+
+pytestmark = pytest.mark.django_db
+
+
+def test_cleanup_deletes_old_expired_tokens():
+    """Test that cleanup deletes tokens expired more than N days ago."""
+    valid_token = ExchangedTokenFactory()
+    expired_recent_token = ExchangedTokenFactory(
+        expires_at=timezone.now() - timedelta(days=3),
+    )
+    expired_old_token = ExchangedTokenFactory(
+        expires_at=timezone.now() - timedelta(days=10),
+    )
+
+    out = StringIO()
+
+    # Default is 7 days, should delete only expired_old_token (10 days)
+    call_command("cleanup_expired_tokens", stdout=out)
+
+    # Check that old expired token was deleted
+    assert not ExchangedToken.objects.filter(id=expired_old_token.id).exists()
+
+    # Check that recent expired token still exists (only 3 days old)
+    assert ExchangedToken.objects.filter(id=expired_recent_token.id).exists()
+
+    # Check that valid token still exists
+    assert ExchangedToken.objects.filter(id=valid_token.id).exists()
+
+
+def test_cleanup_preserves_non_expired_tokens():
+    """Test that cleanup preserves non-expired tokens."""
+    valid_token = ExchangedTokenFactory()
+
+    out = StringIO()
+
+    call_command("cleanup_expired_tokens", stdout=out)
+
+    # Valid token should still exist
+    assert ExchangedToken.objects.filter(id=valid_token.id).exists()
+
+
+def test_cleanup_with_custom_days_parameter():
+    """Test cleanup with custom --days parameter."""
+    expired_recent_token = ExchangedTokenFactory(
+        expires_at=timezone.now() - timedelta(days=3),
+    )
+    expired_old_token = ExchangedTokenFactory(
+        expires_at=timezone.now() - timedelta(days=10),
+    )
+
+    out = StringIO()
+
+    # Set days=2, should delete both tokens (10 days and 3 days)
+    call_command("cleanup_expired_tokens", "--days=2", stdout=out)
+
+    # Both expired tokens should be deleted
+    assert not ExchangedToken.objects.filter(id=expired_old_token.id).exists()
+    assert not ExchangedToken.objects.filter(id=expired_recent_token.id).exists()
+
+
+def test_cleanup_dry_run_does_not_delete():
+    """Test that --dry-run option doesn't actually delete tokens."""
+    expired_old_token = ExchangedTokenFactory(
+        expires_at=timezone.now() - timedelta(days=10),
+    )
+    out = StringIO()
+
+    call_command("cleanup_expired_tokens", "--dry-run", stdout=out)
+
+    # Token should still exist after dry-run
+    assert ExchangedToken.objects.filter(id=expired_old_token.id).exists()
+
+    # Output should indicate dry-run
+    assert "DRY RUN" in out.getvalue() or "Would delete" in out.getvalue()
+
+
+def test_cleanup_displays_correct_count():
+    """Test that cleanup displays correct count of deleted tokens."""
+    out = StringIO()
+
+    # Create multiple old expired tokens
+    ExchangedTokenFactory.create_batch(
+        6,
+        expires_at=timezone.now() - timedelta(days=10),
+    )
+
+    call_command("cleanup_expired_tokens", stdout=out)
+
+    output = out.getvalue()
+    # Should mention deletion of 6 tokens
+    assert "6" in output or "six" in output.lower()
+
+
+def test_cleanup_with_no_expired_tokens():
+    """Test cleanup when there are no expired tokens to delete."""
+    valid_token = ExchangedTokenFactory()
+
+    out = StringIO()
+
+    call_command("cleanup_expired_tokens", stdout=out)
+
+    output = out.getvalue()
+    assert "No expired tokens" in output or "0" in output
+
+
+def test_cleanup_preserves_revoked_but_not_expired_tokens():
+    """Test that cleanup doesn't delete revoked tokens that haven't expired."""
+    # Create a revoked but not expired token
+    token = ExchangedTokenFactory()
+    token.revoke()
+
+    out = StringIO()
+    call_command("cleanup_expired_tokens", stdout=out)
+
+    # Token should still exist (not expired)
+    assert ExchangedToken.objects.filter(id=token.id).exists()
+
+
+def test_cleanup_logs_deletion(caplog):
+    """Test that cleanup logs the deletion."""
+    _expired_old_token = ExchangedTokenFactory(
+        expires_at=timezone.now() - timedelta(days=10),
+    )
+
+    caplog.set_level(logging.INFO)
+
+    out = StringIO()
+    call_command("cleanup_expired_tokens", stdout=out)
+
+    assert "Cleaned up" in caplog.text
+    assert "expired tokens" in caplog.text
+
+
+def test_cleanup_dry_run_logs_simulation(caplog):
+    """Test that dry-run logs simulation message."""
+    _expired_old_token = ExchangedTokenFactory(
+        expires_at=timezone.now() - timedelta(days=10),
+    )
+
+    caplog.set_level(logging.INFO)
+
+    out = StringIO()
+    call_command("cleanup_expired_tokens", "--dry-run", stdout=out)
+
+    assert "DRY RUN" in caplog.text or "Would clean up" in caplog.text
+
+
+@mock.patch("token_exchange.tasks.call_command")
+def test_celery_task_calls_cleanup_command(mock_call_command):
+    """Test that Celery task calls the cleanup command."""
+
+    cleanup_expired_tokens_task()
+
+    mock_call_command.assert_called_once_with("cleanup_expired_tokens")

src/backend/token_exchange/tests/test_models.py

@@ -0,0 +1,377 @@
+"""Tests for new token exchange models with ActionScopes and throttling."""
+
+import base64
+import json
+
+from django.urls import reverse
+
+import pytest
+import responses
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import rsa
+from joserfc import jwt
+from joserfc.jwk import RSAKey
+from lasuite.oidc_resource_server.backend import ResourceServerBackend
+from pytest_unordered import unordered
+from rest_framework import status
+
+from core.factories import ServiceProviderFactory
+
+from token_exchange.factories import (
+    ActionScopeFactory,
+    ActionScopeGrantFactory,
+    ScopeGrantFactory,
+    ServiceProviderCredentialsFactory,
+    TokenExchangeActionPermissionFactory,
+    TokenExchangeRuleFactory,
+)
+from token_exchange.models import ExchangedToken, TokenTypeChoices
+
+pytestmark = pytest.mark.django_db
+
+
+@pytest.fixture(autouse=True)
+def relax_backend(monkeypatch):
+    """Disable strict scope verification from the ResourceServerBackend for these tests."""
+    monkeypatch.setattr(
+        ResourceServerBackend, "_verify_user_info", lambda self, claims: claims
+    )
+
+
+@responses.activate
+def test_exchange_with_scope_grant_and_throttling(api_client, settings, monkeypatch):
+    """Test token exchange with scope grant that includes throttling."""
+    settings.TOKEN_EXCHANGE_ENABLED = True
+    settings.OIDC_OP_INTROSPECTION_ENDPOINT = "http://mock-oidc/introspect"
+    settings.OIDC_RS_CLIENT_ID = "service-a"
+    settings.OIDC_RS_AUDIENCE_CLAIM = "aud"
+    settings.OIDC_OP_URL = "http://mock-oidc"
+    settings.OIDC_RS_REQUIRED_SCOPES = []
+
+    # Setup: Service A → Service B with throttled scope
+    service_a = ServiceProviderFactory(audience_id="service-a")
+    service_b = ServiceProviderFactory(audience_id="service-b")
+    credentials = ServiceProviderCredentialsFactory(service_provider=service_a)
+    credentials_b64 = base64.b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    # Create rule with scope grants
+    rule = TokenExchangeRuleFactory(source_service=service_a, target_service=service_b)
+    ScopeGrantFactory(
+        rule=rule,
+        source_scope="orders.read",
+        granted_scope="orders.read",
+        throttle_rate=None,  # No throttling
+    )
+    ScopeGrantFactory(
+        rule=rule,
+        source_scope="payments.write",
+        granted_scope="payments.write",
+        throttle_rate="5/h",  # Throttled
+    )
+    ScopeGrantFactory(
+        rule=rule,
+        source_scope="payments.write",
+        granted_scope="payments.refund",
+        throttle_rate="5/h",  # Throttled
+    )
+
+    # Mock introspection
+    responses.add(
+        responses.POST,
+        "http://mock-oidc/introspect",
+        json={
+            "active": True,
+            "iss": "http://mock-oidc",
+            "sub": "user-123",
+            "email": "test@example.com",
+            "scope": "orders.read payments.write",
+            "jti": "token-jti",
+            "aud": "service-a",
+        },
+        status=200,
+    )
+
+    # Exchange token
+    url = reverse("token-exchange")
+    data = {
+        "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
+        "subject_token": "external-sso-token",
+        "subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
+        "audience": "service-b",
+    }
+
+    response = api_client.post(
+        url, data, headers={"Authorization": f"Basic {credentials_b64}"}, format="json"
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+    assert "access_token" in response.data
+
+    # Verify the exchanged token contains grants
+    token = ExchangedToken.objects.get(subject_sub="user-123")
+    assert "orders.read" in token.scope
+    assert "payments.write" in token.scope
+    assert "payments.refund" not in token.scope  # no escalation without requesting it
+
+    assert token.grants == {
+        'service-b': unordered([
+            {'scope': 'orders.read', 'throttle': {}},
+            {'scope': 'payments.write', 'throttle': {'rate': '5/h'}}
+        ]),
+    }
+
+
+@responses.activate
+def test_exchange_with_action_scope(api_client, settings):
+    """Test token exchange with action scope that grants multiple scopes."""
+    settings.TOKEN_EXCHANGE_ENABLED = True
+    settings.OIDC_OP_INTROSPECTION_ENDPOINT = "http://mock-oidc/introspect"
+    settings.OIDC_RS_CLIENT_ID = "service-a"
+    settings.OIDC_RS_AUDIENCE_CLAIM = "aud"
+    settings.OIDC_OP_URL = "http://mock-oidc"
+    settings.OIDC_RS_REQUIRED_SCOPES = []
+
+    # Setup: Service A → Service B with action scope
+    service_a = ServiceProviderFactory(audience_id="service-a")
+    service_b = ServiceProviderFactory(audience_id="service-b")
+    credentials = ServiceProviderCredentialsFactory(service_provider=service_a)
+    credentials_b64 = base64.b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    # Create an action scope
+    action = ActionScopeFactory(
+        name="action:upload-transcript",
+        description="Upload and process student transcript",
+    )
+
+    # Action grants multiple scopes on service B
+    ActionScopeGrantFactory(
+        action=action,
+        target_service=service_b,
+        granted_scope="files.write",
+        throttle_rate="10/h",
+    )
+    ActionScopeGrantFactory(
+        action=action,
+        target_service=service_b,
+        granted_scope="transcripts.create",
+        throttle_rate="5/h",
+    )
+
+    # Create rule and permission
+    rule = TokenExchangeRuleFactory(source_service=service_a, target_service=service_b)
+    TokenExchangeActionPermissionFactory(
+        rule=rule,
+        action=action,
+        required_source_scope="admin",  # Requires admin scope to use this action
+    )
+
+    # Mock introspection with admin scope
+    responses.add(
+        responses.POST,
+        "http://mock-oidc/introspect",
+        json={
+            "active": True,
+            "iss": "http://mock-oidc",
+            "sub": "user-456",
+            "email": "admin@example.com",
+            "scope": "admin",
+            "jti": "token-jti-2",
+            "aud": "service-a",
+        },
+        status=200,
+    )
+
+    # Exchange token requesting the action
+    url = reverse("token-exchange")
+    data = {
+        "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
+        "subject_token": "external-sso-token",
+        "subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
+        "audience": "service-b",
+        "scope": "action:upload-transcript",
+    }
+
+    response = api_client.post(
+        url, data, headers={"Authorization": f"Basic {credentials_b64}"}, format="json"
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+
+    # Verify the exchanged token contains all action scopes
+    token = ExchangedToken.objects.get(subject_sub="user-456")
+    assert "files.write" in token.scope
+    assert "transcripts.create" in token.scope
+
+    # Should have grants with throttling organized by audience_id
+    assert "service-b" in token.grants
+    grants_for_service_b = token.grants["service-b"]
+    assert len(grants_for_service_b) == 2
+    throttle_rates = {g["scope"]: g["throttle"]["rate"] for g in grants_for_service_b}
+    assert throttle_rates["files.write"] == "10/h"
+    assert throttle_rates["transcripts.create"] == "5/h"
+
+
+@responses.activate
+def test_exchange_action_scope_requires_permission(api_client, settings):
+    """Test that action scope requires proper source scope."""
+    settings.TOKEN_EXCHANGE_ENABLED = True
+    settings.OIDC_OP_INTROSPECTION_ENDPOINT = "http://mock-oidc/introspect"
+    settings.OIDC_RS_CLIENT_ID = "service-a"
+    settings.OIDC_RS_AUDIENCE_CLAIM = "aud"
+    settings.OIDC_OP_URL = "http://mock-oidc"
+    settings.OIDC_RS_REQUIRED_SCOPES = []
+
+    # Setup
+    service_a = ServiceProviderFactory(audience_id="service-a")
+    service_b = ServiceProviderFactory(audience_id="service-b")
+    credentials = ServiceProviderCredentialsFactory(service_provider=service_a)
+    credentials_b64 = base64.b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    # Create an action that requires admin scope
+    action = ActionScopeFactory(name="action:delete-account")
+    ActionScopeGrantFactory(
+        action=action,
+        target_service=service_b,
+        granted_scope="accounts.delete",
+    )
+
+    rule = TokenExchangeRuleFactory(source_service=service_a, target_service=service_b)
+    TokenExchangeActionPermissionFactory(
+        rule=rule,
+        action=action,
+        required_source_scope="admin",  # Requires admin
+    )
+
+    # Mock introspection WITHOUT admin scope
+    responses.add(
+        responses.POST,
+        "http://mock-oidc/introspect",
+        json={
+            "active": True,
+            "iss": "http://mock-oidc",
+            "sub": "user-789",
+            "email": "user@example.com",
+            "scope": "read",  # Only has read, not admin
+            "jti": "token-jti-3",
+            "aud": "service-a",
+        },
+        status=200,
+    )
+
+    # Try to exchange with action scope
+    url = reverse("token-exchange")
+    data = {
+        "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
+        "subject_token": "external-sso-token",
+        "subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
+        "audience": "service-b",
+        "scope": "action:delete-account",
+    }
+
+    response = api_client.post(
+        url, data, headers={"Authorization": f"Basic {credentials_b64}"}, format="json"
+    )
+
+    # Should fail because user doesn't have required admin scope
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert response.data == {
+        "error": "invalid_target",
+        "error_description": "Invalid target audience",
+    }
+
+
+@responses.activate
+def test_exchange_jwt_contains_grants_in_payload(api_client, settings):
+    """Test that JWT tokens contain grants in the payload."""
+    settings.TOKEN_EXCHANGE_ENABLED = True
+    settings.OIDC_OP_INTROSPECTION_ENDPOINT = "http://mock-oidc/introspect"
+    settings.OIDC_RS_CLIENT_ID = "service-a"
+    settings.OIDC_RS_AUDIENCE_CLAIM = "aud"
+    settings.OIDC_OP_URL = "http://mock-oidc"
+    settings.OIDC_RS_REQUIRED_SCOPES = []
+
+    private_key = rsa.generate_private_key(
+        public_exponent=65537, key_size=2048, backend=default_backend()
+    )
+    private_pem = private_key.private_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=serialization.PrivateFormat.PKCS8,
+        encryption_algorithm=serialization.NoEncryption(),
+    ).decode("utf-8")
+
+    settings.TOKEN_EXCHANGE_JWT_SIGNING_KEYS = {"test-key-1": private_pem}
+    settings.TOKEN_EXCHANGE_JWT_CURRENT_KID = "test-key-1"
+    settings.TOKEN_EXCHANGE_JWT_ALGORITHM = "RS256"
+
+    # Setup
+    service_a = ServiceProviderFactory(audience_id="service-a")
+    service_b = ServiceProviderFactory(audience_id="service-b")
+    credentials = ServiceProviderCredentialsFactory(service_provider=service_a)
+    credentials_b64 = base64.b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    # Create rule with throttled grant
+    rule = TokenExchangeRuleFactory(source_service=service_a, target_service=service_b)
+    ScopeGrantFactory(
+        rule=rule,
+        source_scope="api.access",
+        granted_scope="api.access",
+        throttle_rate="100/day",
+    )
+
+    # Mock introspection
+    responses.add(
+        responses.POST,
+        "http://mock-oidc/introspect",
+        json={
+            "active": True,
+            "iss": "http://mock-oidc",
+            "sub": "user-jwt",
+            "email": "jwt@example.com",
+            "scope": "api.access",
+            "jti": "token-jti-jwt",
+            "aud": "service-a",
+        },
+        status=200,
+    )
+
+    # Request JWT token
+    url = reverse("token-exchange")
+    data = {
+        "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
+        "subject_token": "external-sso-token",
+        "subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
+        "requested_token_type": "urn:ietf:params:oauth:token-type:jwt",
+        "audience": "service-b",
+    }
+
+    response = api_client.post(
+        url, data, headers={"Authorization": f"Basic {credentials_b64}"}, format="json"
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+    assert response.data["issued_token_type"] == "urn:ietf:params:oauth:token-type:jwt"
+
+    # Decode JWT to verify grants are in the payload
+    token_value = response.data["access_token"]
+    key = RSAKey.import_key(private_pem)
+    claims = jwt.decode(token_value, key).claims
+
+    # Verify grants in JWT payload
+    assert "grants" in claims
+    assert isinstance(claims["grants"], dict)
+    assert "service-b" in claims["grants"]
+    service_b_grants = claims["grants"]["service-b"]
+    assert len(service_b_grants) == 1
+    assert service_b_grants[0]["scope"] == "api.access"
+    assert service_b_grants[0]["throttle"]["rate"] == "100/day"
+    assert claims["aud"] == ["service-b"]
+    assert "api.access" in claims["scope"]

src/backend/token_exchange/tests/test_token_exchange.py

@@ -0,0 +1,971 @@
+"""Tests for token exchange endpoint (RFC 8693)."""
+
+import base64
+import logging
+
+from django.urls import reverse
+from django.utils import timezone
+
+import pytest
+import responses
+from rest_framework import status
+
+from core.factories import ServiceProviderFactory
+
+from token_exchange.factories import (
+    ScopeGrantFactory,
+    ServiceProviderCredentialsFactory,
+    TokenExchangeRuleFactory,
+)
+from token_exchange.models import (
+    ExchangedToken,
+    TokenTypeChoices,
+)
+from token_exchange.tests.helpers import (
+    DummyBackend,
+    create_token_exchange_rule_with_scopes,
+)
+
+pytestmark = pytest.mark.django_db
+
+
+def _mock_introspection_backend(settings, user_info, origin=None):
+    """Configure the resource server backend to use a dummy backend for testing.
+
+    Instead of patching the function, we change OIDC_RS_BACKEND_CLASS setting
+    to use a custom DummyBackend that returns fixed introspection data.
+    """
+
+    # Create a factory function that returns a configured DummyBackend instance
+    def backend_factory():
+        return DummyBackend(user_info=user_info, origin=origin)
+
+    # Set the backend class to our factory (it will be called by get_resource_server_introspection_backend)
+    # We need to use the full module path for the backend class
+    settings.OIDC_RS_BACKEND_CLASS = "token_exchange.tests.helpers.DummyBackend"
+
+    # Store the user_info and origin as class attributes so they're accessible
+    # when the backend is instantiated
+    DummyBackend._test_user_info = user_info
+    DummyBackend._test_origin = origin
+
+
+@responses.activate
+def test_exchange_access_token_opaque_success(api_client, settings):
+    """Test successful exchange for opaque access_token."""
+    # Enable feature
+    settings.TOKEN_EXCHANGE_ENABLED = True
+    settings.OIDC_OP_INTROSPECTION_ENDPOINT = "http://mock-oidc/introspect"
+    settings.OIDC_RS_CLIENT_ID = "aud1"
+    settings.OIDC_RS_AUDIENCE_CLAIM = "aud"
+    settings.OIDC_OP_URL = "http://mock-oidc"
+
+    # Mock authentication
+    sp_audience = "aud1"
+    credentials = ServiceProviderCredentialsFactory(
+        service_provider__audience_id=sp_audience
+    )
+    credentials_b64 = base64.b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    # Create token exchange rule and grants
+    target_sp = ServiceProviderFactory(audience_id="aud2")
+    rule = TokenExchangeRuleFactory(
+        source_service=credentials.service_provider,
+        target_service=target_sp,
+    )
+    ScopeGrantFactory(rule=rule, source_scope="openid", granted_scope="openid")
+    ScopeGrantFactory(rule=rule, source_scope="email", granted_scope="email")
+
+    # Mock introspection response
+    responses.add(
+        responses.POST,
+        "http://mock-oidc/introspect",
+        json={
+            "active": True,
+            "iss": "http://mock-oidc",
+            "sub": "user-123",
+            "email": "test@example.com",
+            "scope": "openid email",
+            "jti": "sso-token-id",
+            "aud": sp_audience,
+        },
+        status=200,
+    )
+
+    # Define validation payload
+    # auth_payload = {
+    #     "sub": "user-123",
+    #     "email": "test@example.com",
+    #     "scope": "openid email",
+    #     "jti": "sso-token-id",
+    #     "aud": sp_audience # Matches requesting SP
+    # }
+
+    url = reverse("token-exchange")
+    data = {
+        "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
+        "subject_token": "external-sso-token",
+        "subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
+        "requested_token_type": "urn:ietf:params:oauth:token-type:access_token",
+        "audience": "aud2",
+    }
+
+    response = api_client.post(
+        url, data, headers={"Authorization": f"Basic {credentials_b64}"}, format="json"
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+    assert "access_token" in response.data
+    assert response.data["token_type"] == "Bearer"
+    assert (
+        response.data["issued_token_type"]
+        == "urn:ietf:params:oauth:token-type:access_token"
+    )
+    assert "expires_in" in response.data
+    assert "scope" in response.data
+
+    # Verify token was created in DB
+    assert ExchangedToken.objects.filter(subject_sub="user-123").count() == 1
+    token = ExchangedToken.objects.get(subject_sub="user-123")
+    assert token.token_type == TokenTypeChoices.ACCESS_TOKEN
+    assert token.subject_email == "test@example.com"
+
+
+@responses.activate
+def test_exchange_with_custom_expires_in(api_client, settings):
+    """Test exchange with custom expires_in parameter."""
+    settings.TOKEN_EXCHANGE_ENABLED = True
+    settings.TOKEN_EXCHANGE_MAX_EXPIRES_IN = 7200
+    settings.OIDC_OP_INTROSPECTION_ENDPOINT = "http://mock-oidc/introspect"
+    settings.OIDC_RS_CLIENT_ID = "aud1"
+    settings.OIDC_RS_AUDIENCE_CLAIM = "aud"
+    settings.OIDC_OP_URL = "http://mock-oidc"
+
+    # auth_payload = {
+    #     "sub": "user-123",
+    #     "email": "test@example.com",
+    #     "scope": "openid email",
+    #     "aud": "aud1"
+    # }
+
+    url = reverse("token-exchange")
+    data = {
+        "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
+        "subject_token": "external-sso-token",
+        "subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
+        "expires_in": 1800,  # 30 minutes
+        "audience": "aud1",
+    }
+
+    target_sp = ServiceProviderFactory(audience_id="aud1")
+    credentials = ServiceProviderCredentialsFactory(service_provider=target_sp)
+    credentials_b64 = base64.b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    # Create token exchange rule
+    create_token_exchange_rule_with_scopes(
+        credentials.service_provider,
+        target_sp,
+        ["openid", "email"],
+    )
+
+    responses.add(
+        responses.POST,
+        "http://mock-oidc/introspect",
+        json={
+            "active": True,
+            "iss": "http://mock-oidc",
+            "sub": "user-123",
+            "email": "test@example.com",
+            "scope": "openid email",
+            "aud": "aud1",
+        },
+        status=200,
+    )
+
+    response = api_client.post(
+        url, data, headers={"Authorization": f"Basic {credentials_b64}"}, format="json"
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+    assert response.data["expires_in"] == 1800
+
+
+@responses.activate
+def test_exchange_with_multiple_audiences(api_client, settings):
+    """Test exchange with multiple audiences separated by spaces."""
+    settings.TOKEN_EXCHANGE_ENABLED = True
+    settings.TOKEN_EXCHANGE_MULTI_AUDIENCES_ALLOWED = True
+    settings.OIDC_OP_INTROSPECTION_ENDPOINT = "http://mock-oidc/introspect"
+    settings.OIDC_RS_CLIENT_ID = "original-aud"
+    settings.OIDC_RS_AUDIENCE_CLAIM = "aud"
+    settings.OIDC_OP_URL = "http://mock-oidc"
+
+    # auth_payload = {
+    #     "sub": "user-123",
+    #     "email": "test@example.com",
+    #     "aud": "original-aud",
+    #     "scope": "openid email"
+    # }
+
+    url = reverse("token-exchange")
+    data = {
+        "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
+        "subject_token": "external-sso-token",
+        "subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
+        "audience": "aud1 aud2 aud3",
+    }
+
+    credentials = ServiceProviderCredentialsFactory(
+        service_provider__audience_id="original-aud"
+    )
+    credentials_b64 = base64.b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    # Create token exchange rules
+    sp1 = ServiceProviderFactory(audience_id="aud1")
+    sp2 = ServiceProviderFactory(audience_id="aud2")
+    sp3 = ServiceProviderFactory(audience_id="aud3")
+
+    create_token_exchange_rule_with_scopes(
+        credentials.service_provider, sp1, ["openid", "email"]
+    )
+    create_token_exchange_rule_with_scopes(
+        credentials.service_provider, sp2, ["openid", "email"]
+    )
+    create_token_exchange_rule_with_scopes(
+        credentials.service_provider, sp3, ["openid", "email"]
+    )
+
+    responses.add(
+        responses.POST,
+        "http://mock-oidc/introspect",
+        json={
+            "active": True,
+            "iss": "http://mock-oidc",
+            "sub": "user-123",
+            "email": "test@example.com",
+            "scope": "openid email",
+            "aud": "original-aud",
+        },
+        status=200,
+    )
+
+    response = api_client.post(
+        url, data, headers={"Authorization": f"Basic {credentials_b64}"}, format="json"
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+
+    # Verify audiences were parsed correctly
+    token = ExchangedToken.objects.get(subject_sub="user-123")
+    assert len(token.audiences) == 3
+    assert "aud1" in token.audiences
+    assert "aud2" in token.audiences
+    assert "aud3" in token.audiences
+
+
+@responses.activate
+def test_exchange_with_scope_subset(api_client, settings):
+    """Test exchange with requested scopes that are subset of available."""
+    settings.TOKEN_EXCHANGE_ENABLED = True
+    settings.OIDC_OP_INTROSPECTION_ENDPOINT = "http://mock-oidc/introspect"
+    settings.OIDC_RS_CLIENT_ID = "aud1"
+    settings.OIDC_RS_AUDIENCE_CLAIM = "aud"
+    settings.OIDC_OP_URL = "http://mock-oidc"
+
+    # auth_payload = {
+    #     "sub": "user-123",
+    #     "email": "test@example.com",
+    #     "scope": "openid email groups profile",
+    #     "aud": "aud1"
+    # }
+
+    url = reverse("token-exchange")
+    data = {
+        "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
+        "subject_token": "external-sso-token",
+        "subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
+        "scope": "openid email",  # Subset of available
+        "audience": "aud1",
+    }
+
+    target_sp = ServiceProviderFactory(audience_id="aud1")
+    credentials = ServiceProviderCredentialsFactory(service_provider=target_sp)
+    credentials_b64 = base64.b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    # Create token exchange rule
+    create_token_exchange_rule_with_scopes(
+        credentials.service_provider,
+        target_sp,
+        ["openid", "email", "groups", "profile"],
+    )
+
+    responses.add(
+        responses.POST,
+        "http://mock-oidc/introspect",
+        json={
+            "active": True,
+            "iss": "http://mock-oidc",
+            "sub": "user-123",
+            "email": "test@example.com",
+            "scope": "openid email groups profile",
+            "aud": "aud1",
+        },
+        status=200,
+    )
+
+    response = api_client.post(
+        url, data, headers={"Authorization": f"Basic {credentials_b64}"}, format="json"
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+    assert set(response.data["scope"].split()) == {"openid", "email"}
+
+
+def test_exchange_rejects_refresh_token(api_client, settings):
+    """Test that refresh_token type is explicitly rejected."""
+    settings.TOKEN_EXCHANGE_ENABLED = True
+    settings.OIDC_OP_URL = "http://mock-oidc"
+    settings.OIDC_OP_INTROSPECTION_ENDPOINT = "http://mock-oidc/introspect"
+
+    auth_payload = {"sub": "user-123", "email": "test@example.com", "aud": "aud1"}
+
+    url = reverse("token-exchange")
+    data = {
+        "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
+        "subject_token": "external-sso-token",
+        "subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
+        "requested_token_type": "urn:ietf:params:oauth:token-type:refresh_token",
+        "audience": "aud1",
+    }
+
+    credentials = ServiceProviderCredentialsFactory()
+    credentials_b64 = base64.b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    # Create token exchange rule (though it might fail validation before needing it, safe to have)
+    target_sp = ServiceProviderFactory(audience_id="aud1")
+    create_token_exchange_rule_with_scopes(
+        credentials.service_provider,
+        target_sp,
+        ["openid", "email"],
+    )
+
+    response = api_client.post(
+        url, data, headers={"Authorization": f"Basic {credentials_b64}"}, format="json"
+    )
+
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert "error" in response.data
+    assert "refresh_token" in str(response.data)
+
+
+def test_exchange_rejects_invalid_grant_type(api_client, settings):
+    """Test that invalid grant_type is rejected."""
+    settings.TOKEN_EXCHANGE_ENABLED = True
+    settings.OIDC_OP_URL = "http://mock-oidc"
+    settings.OIDC_OP_INTROSPECTION_ENDPOINT = "http://mock-oidc/introspect"
+
+    auth_payload = {"sub": "user-123", "email": "test@example.com", "aud": "aud1"}
+
+    url = reverse("token-exchange")
+    data = {
+        "grant_type": "invalid-grant-type",
+        "subject_token": "external-sso-token",
+        "subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
+        "audience": "aud1",
+    }
+
+    credentials = ServiceProviderCredentialsFactory()
+    credentials_b64 = base64.b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    target_sp = ServiceProviderFactory(audience_id="aud1")
+    create_token_exchange_rule_with_scopes(
+        credentials.service_provider,
+        target_sp,
+        ["openid", "email"],
+    )
+
+    response = api_client.post(
+        url, data, headers={"Authorization": f"Basic {credentials_b64}"}, format="json"
+    )
+
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+
+
+@responses.activate
+def test_exchange_rejects_scope_elevation(api_client, settings):
+    """Test that scope elevation (requesting additional scopes) is rejected."""
+    settings.TOKEN_EXCHANGE_ENABLED = True
+    settings.OIDC_OP_INTROSPECTION_ENDPOINT = "http://mock-oidc/introspect"
+    settings.OIDC_RS_CLIENT_ID = "aud1"
+    settings.OIDC_RS_AUDIENCE_CLAIM = "aud"
+    settings.OIDC_OP_URL = "http://mock-oidc"
+
+    # auth_payload = {
+    #     "sub": "user-123",
+    #     "email": "test@example.com",
+    #     "scope": "openid email",
+    #     "aud": "aud1"
+    # }
+
+    url = reverse("token-exchange")
+    data = {
+        "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
+        "subject_token": "external-sso-token",
+        "subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
+        "scope": "openid email profile",  # Requesting extra scope
+        "audience": "aud1",
+    }
+
+    target_sp = ServiceProviderFactory(audience_id="aud1")
+    credentials = ServiceProviderCredentialsFactory(service_provider=target_sp)
+    credentials_b64 = base64.b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    create_token_exchange_rule_with_scopes(
+        credentials.service_provider,
+        target_sp,
+        ["openid", "email"],
+    )
+
+    responses.add(
+        responses.POST,
+        "http://mock-oidc/introspect",
+        json={
+            "active": True,
+            "iss": "http://mock-oidc",
+            "sub": "user-123",
+            "email": "test@example.com",
+            "scope": "openid email",
+            "aud": "aud1",
+        },
+        status=200,
+    )
+
+    response = api_client.post(
+        url, data, headers={"Authorization": f"Basic {credentials_b64}"}, format="json"
+    )
+
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert response.data == {
+        "error": "invalid_target",
+        "error_description": "Invalid target audience",
+    }
+
+
+def test_exchange_rejects_expires_in_exceeding_max(api_client, settings):
+    """Test that expires_in exceeding max is rejected."""
+    settings.TOKEN_EXCHANGE_ENABLED = True
+    settings.TOKEN_EXCHANGE_MAX_EXPIRES_IN = 3600
+    settings.OIDC_OP_URL = "http://mock-oidc"
+    settings.OIDC_OP_INTROSPECTION_ENDPOINT = "http://mock-oidc/introspect"
+
+    auth_payload = {
+        "sub": "user-123",
+        "email": "test@example.com",
+        "scope": "openid email",
+        "aud": "aud1",
+    }
+
+    url = reverse("token-exchange")
+    data = {
+        "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
+        "subject_token": "external-sso-token",
+        "subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
+        "expires_in": 7200,  # Exceeds max of 3600
+        "audience": "aud1",
+    }
+
+    credentials = ServiceProviderCredentialsFactory()
+    credentials_b64 = base64.b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    target_sp = ServiceProviderFactory(audience_id="aud1")
+    create_token_exchange_rule_with_scopes(
+        credentials.service_provider,
+        target_sp,
+        ["openid", "email"],
+    )
+
+    response = api_client.post(
+        url, data, headers={"Authorization": f"Basic {credentials_b64}"}, format="json"
+    )
+
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+
+
+def test_exchange_requires_feature_enabled(api_client, settings):
+    """Test that exchange requires TOKEN_EXCHANGE_ENABLED=True."""
+    settings.TOKEN_EXCHANGE_ENABLED = False
+
+    url = reverse("token-exchange")
+    data = {
+        "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
+        "subject_token": "external-sso-token",
+        "subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
+        "audience": "aud1",
+    }
+
+    credentials = ServiceProviderCredentialsFactory()
+    credentials_b64 = base64.b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    target_sp = ServiceProviderFactory(audience_id="aud1")
+    create_token_exchange_rule_with_scopes(
+        credentials.service_provider,
+        target_sp,
+        ["openid", "email"],
+    )
+
+    # Even with valid auth, it should fail
+    response = api_client.post(
+        url, data, headers={"Authorization": f"Basic {credentials_b64}"}, format="json"
+    )
+
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert "token_exchange_disabled" in response.data["error"]
+
+
+@responses.activate
+def test_exchange_enforces_token_limit(api_client, settings):
+    """Test that token limit is enforced and oldest tokens are deleted."""
+    settings.TOKEN_EXCHANGE_ENABLED = True
+    settings.TOKEN_EXCHANGE_MAX_ACTIVE_TOKENS_PER_USER = 3
+    settings.OIDC_OP_INTROSPECTION_ENDPOINT = "http://mock-oidc/introspect"
+    settings.OIDC_RS_CLIENT_ID = "aud1"
+    settings.OIDC_RS_AUDIENCE_CLAIM = "aud"
+    settings.OIDC_OP_URL = "http://mock-oidc"
+
+    # auth_payload = {"sub": "user-123", "email": "test@example.com", "scope": "openid email", "aud": "aud1"}
+
+    url = reverse("token-exchange")
+    data = {
+        "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
+        "subject_token": "external-sso-token",
+        "subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
+        "audience": "aud1",
+    }
+
+    target_sp = ServiceProviderFactory(audience_id="aud1")
+    credentials = ServiceProviderCredentialsFactory(service_provider=target_sp)
+    credentials_b64 = base64.b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    create_token_exchange_rule_with_scopes(
+        credentials.service_provider,
+        target_sp,
+        ["openid", "email"],
+    )
+
+    responses.add(
+        responses.POST,
+        "http://mock-oidc/introspect",
+        json={
+            "active": True,
+            "iss": "http://mock-oidc",
+            "sub": "user-123",
+            "email": "test@example.com",
+            "scope": "openid email",
+            "aud": "aud1",
+        },
+        status=200,
+    )
+
+    # Create 4 tokens (exceeds limit of 3)
+    for _i in range(4):
+        response = api_client.post(
+            url,
+            data,
+            headers={"Authorization": f"Basic {credentials_b64}"},
+            format="json",
+        )
+        assert response.status_code == status.HTTP_200_OK
+
+    # Should only have 3 tokens (oldest one deleted)
+    assert (
+        ExchangedToken.objects.filter(
+            subject_sub="user-123",
+            expires_at__gt=timezone.now(),
+            revoked_at__isnull=True,
+        ).count()
+        == 3
+    )
+
+
+@responses.activate
+def test_exchange_logs_token_creation(api_client, caplog, settings):
+    """Test that token exchange logs appropriate information."""
+
+    caplog.set_level(logging.INFO)
+
+    settings.TOKEN_EXCHANGE_ENABLED = True
+    settings.OIDC_OP_INTROSPECTION_ENDPOINT = "http://mock-oidc/introspect"
+    settings.OIDC_RS_CLIENT_ID = "aud1"
+    settings.OIDC_RS_AUDIENCE_CLAIM = "aud"
+    settings.OIDC_OP_URL = "http://mock-oidc"
+
+    # auth_payload = {
+    #     "sub": "user-123",
+    #     "email": "logtest@example.com",
+    #     "scope": "openid email",
+    #     "aud": "aud1"
+    # }
+
+    url = reverse("token-exchange")
+    data = {
+        "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
+        "subject_token": "external-sso-token",
+        "subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
+        "audience": "aud1",
+    }
+
+    target_sp = ServiceProviderFactory(audience_id="aud1")
+    credentials = ServiceProviderCredentialsFactory(service_provider=target_sp)
+    credentials_b64 = base64.b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    create_token_exchange_rule_with_scopes(
+        credentials.service_provider,
+        target_sp,
+        ["openid", "email"],
+    )
+
+    responses.add(
+        responses.POST,
+        "http://mock-oidc/introspect",
+        json={
+            "active": True,
+            "iss": "http://mock-oidc",
+            "sub": "user-123",
+            "email": "logtest@example.com",
+            "scope": "openid email",
+            "aud": "aud1",
+        },
+        status=200,
+    )
+
+    response = api_client.post(
+        url, data, headers={"Authorization": f"Basic {credentials_b64}"}, format="json"
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+
+    # Check that logging occurred
+    assert "Token exchanged" in caplog.text
+    assert "logtest@example.com" in caplog.text
+    assert "aud1" in caplog.text
+
+
+@responses.activate
+def test_exchange_rejects_unknown_audience(api_client, settings):
+    """Test that unknown audience is rejected."""
+    settings.TOKEN_EXCHANGE_ENABLED = True
+    settings.OIDC_RS_AUDIENCE_CLAIM = "aud"
+
+    credentials = ServiceProviderCredentialsFactory(
+        service_provider__audience_id="source-aud",
+    )
+    credentials_b64 = base64.b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    _mock_introspection_backend(
+        settings,
+        {
+            "active": True,
+            "sub": "user-123",
+            "email": "user@example.com",
+            "scope": "openid",
+            "aud": "source-aud",
+            "jti": "jti-1",
+        },
+    )
+
+    response = api_client.post(
+        reverse("token-exchange"),
+        {
+            "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
+            "subject_token": "external-sso-token",
+            "subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
+            "audience": "unknown-aud",
+        },
+        headers={"Authorization": f"Basic {credentials_b64}"},
+        format="json",
+    )
+
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert response.data == {
+        "error": "invalid_target",
+        "error_description": "Invalid target audience",
+    }
+
+
+def test_exchange_rejects_inactive_rule(api_client, settings):
+    """Test that inactive rule is rejected."""
+    settings.TOKEN_EXCHANGE_ENABLED = True
+
+    target_sp = ServiceProviderFactory(audience_id="aud-inactive")
+    credentials = ServiceProviderCredentialsFactory()
+    credentials_b64 = base64.b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    TokenExchangeRuleFactory(
+        source_service=credentials.service_provider,
+        target_service=target_sp,
+        is_active=False,
+    )
+
+    response = api_client.post(
+        reverse("token-exchange"),
+        {
+            "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
+            "subject_token": "external-sso-token",
+            "subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
+            "audience": target_sp.audience_id,
+        },
+        headers={"Authorization": f"Basic {credentials_b64}"},
+        format="json",
+    )
+
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert response.data["error"] == "invalid_target"
+
+
+def test_exchange_invalid_when_identity_missing(api_client, settings):
+    """Test that missing identity is rejected."""
+    settings.TOKEN_EXCHANGE_ENABLED = True
+    settings.OIDC_RS_AUDIENCE_CLAIM = "aud"
+
+    target_sp = ServiceProviderFactory(audience_id="aud-target")
+    credentials = ServiceProviderCredentialsFactory(
+        service_provider__audience_id="aud-source",
+    )
+    credentials_b64 = base64.b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    create_token_exchange_rule_with_scopes(
+        credentials.service_provider,
+        target_sp,
+        ["openid"],
+    )
+
+    _mock_introspection_backend(
+        settings,
+        {
+            "active": True,
+            "scope": "openid",
+            "aud": "aud-source",
+            "jti": "jti-2",
+        },
+    )
+
+    response = api_client.post(
+        reverse("token-exchange"),
+        {
+            "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
+            "subject_token": "external-sso-token",
+            "subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
+            "audience": target_sp.audience_id,
+        },
+        headers={"Authorization": f"Basic {credentials_b64}"},
+        format="json",
+    )
+
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert response.data["error"] == "invalid_token"
+
+
+def test_exchange_rejects_mismatched_token_origin(api_client, settings):
+    """Test that mismatched token origin is rejected."""
+    settings.TOKEN_EXCHANGE_ENABLED = True
+    settings.OIDC_RS_AUDIENCE_CLAIM = "aud"
+
+    target_sp = ServiceProviderFactory(audience_id="aud-target")
+    credentials = ServiceProviderCredentialsFactory(
+        service_provider__audience_id="aud-source",
+    )
+    credentials_b64 = base64.b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    create_token_exchange_rule_with_scopes(
+        credentials.service_provider,
+        target_sp,
+        ["openid"],
+    )
+
+    _mock_introspection_backend(
+        settings,
+        {
+            "active": True,
+            "sub": "user-456",
+            "email": "another@example.com",
+            "scope": "openid",
+            "aud": "aud-source",
+            "jti": "jti-3",
+        },
+        origin="other-aud",
+    )
+
+    response = api_client.post(
+        reverse("token-exchange"),
+        {
+            "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
+            "subject_token": "external-sso-token",
+            "subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
+            "audience": target_sp.audience_id,
+        },
+        headers={"Authorization": f"Basic {credentials_b64}"},
+        format="json",
+    )
+
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+
+
+def test_exchange_action_scope_requires_permission(api_client, settings):
+    """Test that action scope requires permission."""
+    settings.TOKEN_EXCHANGE_ENABLED = True
+    settings.OIDC_RS_AUDIENCE_CLAIM = "aud"
+
+    target_sp = ServiceProviderFactory(audience_id="aud-action")
+    credentials = ServiceProviderCredentialsFactory(
+        service_provider__audience_id="aud-source",
+    )
+    credentials_b64 = base64.b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    TokenExchangeRuleFactory(
+        source_service=credentials.service_provider,
+        target_service=target_sp,
+    )
+
+    _mock_introspection_backend(
+        settings,
+        {
+            "active": True,
+            "sub": "user-action",
+            "email": "action@example.com",
+            "scope": "openid",
+            "aud": "aud-source",
+            "jti": "jti-4",
+        },
+    )
+
+    response = api_client.post(
+        reverse("token-exchange"),
+        {
+            "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
+            "subject_token": "external-sso-token",
+            "subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
+            "scope": "action:do-something",
+            "audience": target_sp.audience_id,
+        },
+        headers={"Authorization": f"Basic {credentials_b64}"},
+        format="json",
+    )
+
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert response.data["error"] == "invalid_target"
+
+
+def test_exchange_requires_actor_token_type_when_actor_token_present(
+    api_client, settings
+):
+    """Test that actor_token_type is required when actor_token is present."""
+    settings.TOKEN_EXCHANGE_ENABLED = True
+
+    credentials = ServiceProviderCredentialsFactory()
+    credentials_b64 = base64.b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    response = api_client.post(
+        reverse("token-exchange"),
+        {
+            "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
+            "subject_token": "external-sso-token",
+            "subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
+            "audience": "aud-solo",
+            "actor_token": "actor-token-value",
+        },
+        headers={"Authorization": f"Basic {credentials_b64}"},
+        format="json",
+    )
+
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert "actor_token_type" in response.data.get("error_description", "")
+
+
+def test_exchange_multi_audience_disabled_keeps_first(api_client, settings):
+    """Test when multiple audiences are request but not allowed."""
+    settings.TOKEN_EXCHANGE_ENABLED = True
+    settings.TOKEN_EXCHANGE_MULTI_AUDIENCES_ALLOWED = False
+    settings.OIDC_RS_AUDIENCE_CLAIM = "aud"
+
+    credentials = ServiceProviderCredentialsFactory(
+        service_provider__audience_id="aud-source",
+    )
+    credentials_b64 = base64.b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    target_sp_1 = ServiceProviderFactory(audience_id="aud-1")
+    target_sp_2 = ServiceProviderFactory(audience_id="aud-2")
+
+    create_token_exchange_rule_with_scopes(
+        credentials.service_provider,
+        target_sp_1,
+        ["openid"],
+    )
+    create_token_exchange_rule_with_scopes(
+        credentials.service_provider,
+        target_sp_2,
+        ["openid"],
+    )
+
+    _mock_introspection_backend(
+        settings,
+        {
+            "active": True,
+            "sub": "user-multi",
+            "email": "multi@example.com",
+            "scope": "openid",
+            "aud": "aud-source",
+            "jti": "jti-5",
+        },
+    )
+
+    response = api_client.post(
+        reverse("token-exchange"),
+        {
+            "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
+            "subject_token": "external-sso-token",
+            "subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
+            "audience": f"{target_sp_1.audience_id} {target_sp_2.audience_id}",
+        },
+        headers={"Authorization": f"Basic {credentials_b64}"},
+        format="json",
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+    token = ExchangedToken.objects.get(subject_sub="user-multi")
+    assert token.audiences == [target_sp_1.audience_id]

src/backend/token_exchange/tests/test_token_introspection.py

@@ -0,0 +1,268 @@
+"""Tests for token introspection endpoint (RFC 7662)."""
+
+import logging
+from base64 import b64encode
+from datetime import timedelta
+
+from django.urls import reverse
+from django.utils import timezone
+
+import pytest
+from rest_framework import status
+
+from token_exchange.factories import (
+    ExchangedTokenFactory,
+    ExpiredExchangedTokenFactory,
+    ServiceProviderCredentialsFactory,
+)
+from token_exchange.models import ExchangedToken, TokenTypeChoices
+from token_exchange.token_generator import TokenGenerator
+
+pytestmark = pytest.mark.django_db
+
+
+def test_introspect_unauthenticated(api_client):
+    """Test that introspection requires authentication."""
+    url = reverse("token-introspect")
+    valid_opaque_token = ExchangedTokenFactory(
+        token_type=TokenTypeChoices.ACCESS_TOKEN,
+        scope="openid email",
+        audiences=["test-audience"],
+        subject_email="test@example.com",
+        subject_sub="user-123",
+    )
+    data = {"token": valid_opaque_token.token}
+
+    response = api_client.post(url, data, format="json")
+
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+    assert response.data["detail"] == "Authentication credentials were not provided."
+
+
+def test_introspect_valid_opaque_token(api_client):
+    """Test introspection of valid opaque token returns active=true with all fields."""
+    credentials = ServiceProviderCredentialsFactory()
+    credentials_b64 = b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    url = reverse("token-introspect")
+    valid_opaque_token = ExchangedTokenFactory(
+        token_type=TokenTypeChoices.ACCESS_TOKEN,
+        scope="openid email",
+        audiences=["test-audience"],
+        subject_email="test@example.com",
+        subject_sub="user-123",
+    )
+    data = {"token": valid_opaque_token.token}
+
+    response = api_client.post(
+        url, data, format="json", headers={"Authorization": f"Basic {credentials_b64}"}
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+    assert response.data["active"] is True
+    assert response.data["scope"] == "openid email"
+    assert response.data["username"] == "test@example.com"
+    assert response.data["token_type"] == TokenTypeChoices.ACCESS_TOKEN
+    assert "exp" in response.data
+    assert "iat" in response.data
+    assert response.data["sub"] == "user-123"
+    assert response.data["email"] == "test@example.com"
+    assert response.data["aud"] == ["test-audience"]
+    assert "jti" in response.data
+
+
+def test_introspect_with_multiple_audiences(api_client):
+    """Test introspection returns array for multiple audiences."""
+    credentials = ServiceProviderCredentialsFactory()
+    credentials_b64 = b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    token = ExchangedToken.objects.create(
+        token=TokenGenerator.generate_opaque_token(),
+        token_type=TokenTypeChoices.ACCESS_TOKEN,
+        subject_email="test-aud@example.com",
+        subject_sub="user-aud-123",
+        audiences=["aud1", "aud2", "aud3"],
+        scope="openid email",
+        expires_at=timezone.now() + timedelta(hours=1),
+        subject_token_jti="test-jti",
+        subject_token_scope="openid email",
+    )
+
+    url = reverse("token-introspect")
+    data = {"token": token.token}
+
+    response = api_client.post(
+        url, data, format="json", headers={"Authorization": f"Basic {credentials_b64}"}
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+    assert response.data["active"] is True
+    assert response.data["aud"] == ["aud1", "aud2", "aud3"]
+
+
+def test_introspect_no_auth_required(api_client):
+    """Test that introspection endpoint allows unauthenticated access."""
+    credentials = ServiceProviderCredentialsFactory()
+    credentials_b64 = b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    valid_opaque_token = ExchangedTokenFactory(
+        token_type=TokenTypeChoices.ACCESS_TOKEN,
+    )
+
+    # Don't authenticate
+    url = reverse("token-introspect")
+    data = {"token": valid_opaque_token.token}
+
+    response = api_client.post(
+        url, data, format="json", headers={"Authorization": f"Basic {credentials_b64}"}
+    )
+
+    # Should succeed even without authentication
+    assert response.status_code == status.HTTP_200_OK
+    assert response.data["active"] is True
+
+
+def test_introspect_expired_token(api_client):
+    """Test introspection of expired token returns active=false."""
+    credentials = ServiceProviderCredentialsFactory()
+    credentials_b64 = b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    url = reverse("token-introspect")
+    expired_token = ExpiredExchangedTokenFactory()
+    data = {"token": expired_token.token}
+
+    response = api_client.post(
+        url, data, format="json", headers={"Authorization": f"Basic {credentials_b64}"}
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+    assert response.data["active"] is False
+
+
+def test_introspect_revoked_token(api_client):
+    """Test introspection of revoked token returns active=false."""
+    credentials = ServiceProviderCredentialsFactory()
+    credentials_b64 = b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    url = reverse("token-introspect")
+    revoked_token = ExchangedTokenFactory(revoked_at=timezone.now())
+    data = {"token": revoked_token.token}
+
+    response = api_client.post(
+        url, data, format="json", headers={"Authorization": f"Basic {credentials_b64}"}
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+    assert response.data["active"] is False
+
+
+def test_introspect_nonexistent_token(api_client):
+    """Test introspection of non-existent token returns active=false."""
+    credentials = ServiceProviderCredentialsFactory()
+    credentials_b64 = b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    url = reverse("token-introspect")
+    data = {"token": "nonexistent-token-12345"}
+
+    response = api_client.post(
+        url, data, format="json", headers={"Authorization": f"Basic {credentials_b64}"}
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+    assert response.data["active"] is False
+    assert len(response.data) == 1  # Only 'active' field
+
+
+def test_introspect_empty_token(api_client):
+    """Test introspection with empty token returns active=false."""
+    credentials = ServiceProviderCredentialsFactory()
+    credentials_b64 = b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    url = reverse("token-introspect")
+    data = {"token": ""}
+
+    response = api_client.post(
+        url, data, format="json", headers={"Authorization": f"Basic {credentials_b64}"}
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+    assert response.data["active"] is False
+
+
+def test_introspect_missing_token(api_client):
+    """Test introspection without token parameter returns active=false."""
+    credentials = ServiceProviderCredentialsFactory()
+    credentials_b64 = b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    url = reverse("token-introspect")
+    data = {}
+
+    response = api_client.post(
+        url, data, format="json", headers={"Authorization": f"Basic {credentials_b64}"}
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+    assert response.data["active"] is False
+
+
+def test_introspect_logs_valid_token(api_client, caplog):
+    """Test that introspection logs for valid tokens."""
+    credentials = ServiceProviderCredentialsFactory()
+    credentials_b64 = b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    valid_opaque_token = ExchangedTokenFactory(
+        token_type=TokenTypeChoices.ACCESS_TOKEN,
+    )
+
+    caplog.set_level(logging.INFO)
+
+    url = reverse("token-introspect")
+    data = {"token": valid_opaque_token.token}
+
+    response = api_client.post(
+        url, data, format="json", headers={"Authorization": f"Basic {credentials_b64}"}
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+    assert "Token introspected" in caplog.text
+    assert "active=True" in caplog.text or "active=true" in caplog.text.lower()
+
+
+def test_introspect_logs_invalid_token(api_client, caplog):
+    """Test that introspection logs for invalid tokens."""
+    credentials = ServiceProviderCredentialsFactory()
+    credentials_b64 = b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    caplog.set_level(logging.INFO)
+
+    url = reverse("token-introspect")
+    expired_token = ExpiredExchangedTokenFactory()
+    data = {"token": expired_token.token}
+
+    response = api_client.post(
+        url, data, format="json", headers={"Authorization": f"Basic {credentials_b64}"}
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+    assert "Token introspected" in caplog.text
+    assert "active=False" in caplog.text or "active=false" in caplog.text.lower()

src/backend/token_exchange/tests/test_token_limit.py

@@ -0,0 +1,168 @@
+"""Tests for token limit per user."""
+
+import logging
+
+from django.utils import timezone
+
+import pytest
+
+from token_exchange.factories import ExchangedTokenFactory, ExpiredExchangedTokenFactory
+from token_exchange.models import ExchangedToken
+
+pytestmark = pytest.mark.django_db
+
+
+def test_allows_tokens_up_to_limit(settings):
+    """Test that tokens can be created up to the max limit."""
+    settings.TOKEN_EXCHANGE_MAX_ACTIVE_TOKENS_PER_USER = 5
+    sub = "user-limit-1"
+
+    # Create 5 tokens (at limit)
+    for _i in range(5):
+        ExchangedTokenFactory(subject_sub=sub)
+
+    # All 5 should exist
+    assert (
+        ExchangedToken.objects.filter(
+            subject_sub=sub, expires_at__gt=timezone.now(), revoked_at__isnull=True
+        ).count()
+        == 5
+    )
+
+
+def test_deletes_oldest_when_exceeding_limit(settings):
+    """Test that oldest tokens are deleted when limit is exceeded."""
+    settings.TOKEN_EXCHANGE_MAX_ACTIVE_TOKENS_PER_USER = 3
+    sub = "user-limit-2"
+
+    tokens = []
+    for _i in range(5):
+        token = ExchangedTokenFactory(subject_sub=sub)
+        tokens.append(token)
+
+    # Should only have 3 tokens (newest ones)
+    remaining_tokens = ExchangedToken.objects.filter(
+        subject_sub=sub, expires_at__gt=timezone.now(), revoked_at__isnull=True
+    ).order_by("created_at")
+
+    assert remaining_tokens.count() == 3
+
+    # First two tokens should be deleted
+    assert not ExchangedToken.objects.filter(id=tokens[0].id).exists()
+    assert not ExchangedToken.objects.filter(id=tokens[1].id).exists()
+
+    # Last three tokens should still exist
+    assert ExchangedToken.objects.filter(id=tokens[2].id).exists()
+    assert ExchangedToken.objects.filter(id=tokens[3].id).exists()
+    assert ExchangedToken.objects.filter(id=tokens[4].id).exists()
+
+
+def test_preserves_recent_tokens_when_enforcing_limit(settings):
+    """Test that recently created tokens are preserved when enforcing limit."""
+    settings.TOKEN_EXCHANGE_MAX_ACTIVE_TOKENS_PER_USER = 2
+    sub = "user-limit-3"
+
+    # Create 3 tokens
+    token1 = ExchangedTokenFactory(subject_sub=sub)
+    token2 = ExchangedTokenFactory(subject_sub=sub)
+    token3 = ExchangedTokenFactory(subject_sub=sub)  # Newest
+
+    # Token 1 (oldest) should be deleted
+    assert not ExchangedToken.objects.filter(id=token1.id).exists()
+
+    # Tokens 2 and 3 should exist
+    assert ExchangedToken.objects.filter(id=token2.id).exists()
+    assert ExchangedToken.objects.filter(id=token3.id).exists()
+
+
+def test_limit_isolated_between_users(settings):
+    """Test that token limit is enforced per user independently."""
+    settings.TOKEN_EXCHANGE_MAX_ACTIVE_TOKENS_PER_USER = 2
+    sub1 = "user-limit-4a"
+    sub2 = "user-limit-4b"
+
+    # Create 3 tokens for each user
+    for _i in range(3):
+        ExchangedTokenFactory(subject_sub=sub1)
+        ExchangedTokenFactory(subject_sub=sub2)
+
+    # Each user should have exactly 2 tokens
+    user_tokens = ExchangedToken.objects.filter(
+        subject_sub=sub1, expires_at__gt=timezone.now(), revoked_at__isnull=True
+    ).count()
+
+    other_user_tokens = ExchangedToken.objects.filter(
+        subject_sub=sub2, expires_at__gt=timezone.now(), revoked_at__isnull=True
+    ).count()
+
+    assert user_tokens == 2
+    assert other_user_tokens == 2
+
+
+def test_revoked_tokens_count_toward_limit(settings):
+    """Test that revoked tokens still count toward the active limit."""
+    settings.TOKEN_EXCHANGE_MAX_ACTIVE_TOKENS_PER_USER = 3
+    sub = "user-limit-5"
+
+    # Create 2 active tokens
+    token1 = ExchangedTokenFactory(subject_sub=sub)
+    token2 = ExchangedTokenFactory(subject_sub=sub)
+
+    # Revoke one
+    token1.revoke()
+
+    # Create 3 more tokens
+    for _i in range(3):
+        ExchangedTokenFactory(subject_sub=sub)
+
+    # Should have 3 active (non-revoked) tokens
+    active_count = ExchangedToken.objects.filter(
+        subject_sub=sub, expires_at__gt=timezone.now(), revoked_at__isnull=True
+    ).count()
+
+    assert active_count == 3
+
+
+def test_expired_tokens_excluded_from_limit_count(settings):
+    """Test that expired tokens are not counted in the active limit."""
+    settings.TOKEN_EXCHANGE_MAX_ACTIVE_TOKENS_PER_USER = 3
+    sub = "user-limit-6"
+
+    # Create 2 expired tokens
+    for _i in range(2):
+        ExpiredExchangedTokenFactory(subject_sub=sub)
+
+    # Create 3 active tokens
+    for _i in range(3):
+        ExchangedTokenFactory(subject_sub=sub)
+
+    # Should have exactly 3 active tokens (expired ones don't count)
+    active_count = ExchangedToken.objects.filter(
+        subject_sub=sub, expires_at__gt=timezone.now(), revoked_at__isnull=True
+    ).count()
+
+    assert active_count == 3
+
+    # Expired tokens should still exist in DB
+    expired_count = ExchangedToken.objects.filter(
+        subject_sub=sub, expires_at__lte=timezone.now()
+    ).count()
+
+    assert expired_count == 2
+
+
+def test_logs_token_deletion_when_limit_exceeded(settings, caplog):
+    """Test that token deletion due to limit is logged."""
+
+    caplog.set_level(logging.INFO)
+
+    settings.TOKEN_EXCHANGE_MAX_ACTIVE_TOKENS_PER_USER = 2
+    sub = "user-limit-7"
+
+    # Create 3 tokens to exceed limit
+    for _i in range(3):
+        ExchangedTokenFactory(subject_sub=sub)
+
+    # Check logging
+    assert "Enforced token limit" in caplog.text
+    assert sub in caplog.text

src/backend/token_exchange/tests/test_token_revocation.py

@@ -0,0 +1,236 @@
+"""Tests for token revocation endpoint (RFC 7009)."""
+
+import logging
+from base64 import b64encode
+
+from django.urls import reverse
+from django.utils import timezone
+
+import pytest
+from rest_framework import status
+
+from token_exchange.factories import (
+    ExchangedTokenFactory,
+    ServiceProviderCredentialsFactory,
+)
+
+pytestmark = pytest.mark.django_db
+
+
+def test_revoke_token_success(api_client):
+    """Test that authenticated user can successfully revoke a token."""
+    # Authenticate as a service provider
+    credentials = ServiceProviderCredentialsFactory()
+    credentials_b64 = b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    valid_token = ExchangedTokenFactory()
+
+    url = reverse("token-revoke")
+    data = {"token": valid_token.token}
+
+    response = api_client.post(
+        url, data, format="json", headers={"Authorization": f"Basic {credentials_b64}"}
+    )
+
+    # RFC 7009: Always return 200 OK
+    assert response.status_code == status.HTTP_200_OK
+
+    # Verify token was revoked
+    valid_token.refresh_from_db()
+    assert valid_token.is_revoked() is True
+    assert valid_token.revoked_at is not None
+    assert valid_token.is_valid() is False
+
+
+def test_revoke_sets_revoked_at_timestamp(api_client):
+    """Test that revocation sets revoked_at timestamp."""
+    credentials = ServiceProviderCredentialsFactory()
+    credentials_b64 = b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+    valid_token = ExchangedTokenFactory()
+
+    before_revoke = timezone.now()
+
+    url = reverse("token-revoke")
+    data = {"token": valid_token.token}
+
+    response = api_client.post(
+        url, data, format="json", headers={"Authorization": f"Basic {credentials_b64}"}
+    )
+
+    after_revoke = timezone.now()
+
+    assert response.status_code == status.HTTP_200_OK
+
+    valid_token.refresh_from_db()
+    assert valid_token.revoked_at is not None
+    assert before_revoke <= valid_token.revoked_at <= after_revoke
+
+
+def test_revoke_with_token_type_hint(api_client):
+    """Test revocation with optional token_type_hint parameter."""
+    credentials = ServiceProviderCredentialsFactory()
+    credentials_b64 = b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+    valid_token = ExchangedTokenFactory()
+
+    url = reverse("token-revoke")
+    data = {
+        "token": valid_token.token,
+        "token_type_hint": "access_token",
+    }
+
+    response = api_client.post(
+        url, data, format="json", headers={"Authorization": f"Basic {credentials_b64}"}
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+
+    valid_token.refresh_from_db()
+    assert valid_token.is_revoked() is True
+
+
+def test_revoked_token_becomes_invalid(api_client):
+    """Test that token becomes invalid after revocation."""
+    credentials = ServiceProviderCredentialsFactory()
+    credentials_b64 = b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+    valid_token = ExchangedTokenFactory()
+
+    # Token is valid before revocation
+    assert valid_token.is_valid() is True
+
+    url = reverse("token-revoke")
+    data = {"token": valid_token.token}
+
+    response = api_client.post(
+        url, data, format="json", headers={"Authorization": f"Basic {credentials_b64}"}
+    )
+    assert response.status_code == status.HTTP_200_OK
+
+    # Token is invalid after revocation
+    valid_token.refresh_from_db()
+    assert valid_token.is_valid() is False
+    assert valid_token.is_revoked() is True
+
+
+def test_introspection_returns_inactive_after_revocation(api_client):
+    """Test that introspection returns active=false after revocation."""
+    credentials = ServiceProviderCredentialsFactory()
+    credentials_b64 = b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    # First, revoke the token
+    valid_token = ExchangedTokenFactory()
+    revoke_url = reverse("token-revoke")
+    revoke_data = {"token": valid_token.token}
+    revoke_response = api_client.post(
+        revoke_url,
+        revoke_data,
+        format="json",
+        headers={"Authorization": f"Basic {credentials_b64}"},
+    )
+    assert revoke_response.status_code == status.HTTP_200_OK
+
+    # Now introspect the revoked token
+    introspect_url = reverse("token-introspect")
+    introspect_data = {"token": valid_token.token}
+
+    response = api_client.post(
+        introspect_url,
+        introspect_data,
+        format="json",
+        headers={"Authorization": f"Basic {credentials_b64}"},
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+    assert response.data["active"] is False
+
+
+def test_revoke_nonexistent_token_returns_200(api_client):
+    """Test that revoking non-existent token returns 200 (RFC 7009)."""
+    credentials = ServiceProviderCredentialsFactory()
+    credentials_b64 = b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    url = reverse("token-revoke")
+    data = {"token": "nonexistent-token-12345"}
+
+    response = api_client.post(
+        url, data, format="json", headers={"Authorization": f"Basic {credentials_b64}"}
+    )
+
+    # RFC 7009: Silent success
+    assert response.status_code == status.HTTP_200_OK
+
+
+def test_revoke_requires_authentication(api_client):
+    """Test that revocation requires authentication."""
+    valid_token = ExchangedTokenFactory()
+
+    # Don't authenticate
+    url = reverse("token-revoke")
+    data = {"token": valid_token.token}
+
+    response = api_client.post(url, data, format="json")
+
+    # Should fail without authentication (ResourceServerMixin requirement)
+    assert response.status_code in [
+        status.HTTP_401_UNAUTHORIZED,
+        status.HTTP_403_FORBIDDEN,
+    ]
+
+
+def test_revoke_logs_successful_revocation(api_client, caplog):
+    """Test that successful revocation is logged."""
+    credentials = ServiceProviderCredentialsFactory()
+    credentials_b64 = b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    caplog.set_level(logging.INFO)
+
+    valid_token = ExchangedTokenFactory()
+
+    url = reverse("token-revoke")
+    data = {"token": valid_token.token}
+
+    response = api_client.post(
+        url, data, format="json", headers={"Authorization": f"Basic {credentials_b64}"}
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+    assert "Token revoked" in caplog.text
+    # We removed user from log, used sub/email instead which are random/none in factory unless set
+    # check valid_token.subject_sub is randomized?
+    # Factory uses LazyFunction uuid4 for sub.
+    # It might NOT be logged if factory didn't set it (Factory has user field removed, and defaults set).
+    # We should verify "sub=" or "email=" is present.
+    assert "sub=" in caplog.text
+
+
+def test_revoke_logs_attempted_revocation(api_client, caplog):
+    """Test that attempted revocation of non-existent token is logged."""
+    credentials = ServiceProviderCredentialsFactory()
+    credentials_b64 = b64encode(
+        f"{credentials.client_id}:{credentials.client_secret}".encode()
+    ).decode()
+
+    caplog.set_level(logging.INFO)
+
+    url = reverse("token-revoke")
+    data = {"token": "nonexistent-token"}
+
+    response = api_client.post(
+        url, data, format="json", headers={"Authorization": f"Basic {credentials_b64}"}
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+    assert "Token revocation attempted" in caplog.text

src/backend/token_exchange/token_generator.py

@@ -0,0 +1,155 @@
+"""Token generator for RFC 8693 token exchange."""
+
+import secrets
+import uuid
+from datetime import datetime, timedelta
+
+from django.conf import settings
+from django.utils import timezone
+
+from joserfc import jwt
+from joserfc.jwk import RSAKey
+
+
+class TokenGenerator:
+    """Generator for exchanged tokens (opaque or JWT)."""
+
+    @staticmethod
+    def generate_opaque_token():
+        """
+        Generate a secure opaque token.
+
+        Returns:
+            str: A URL-safe random token
+        """
+        return secrets.token_urlsafe(32)
+
+    @staticmethod
+    def generate_jwt_token(  # noqa: PLR0913
+        sub,
+        email,
+        audiences,
+        scope,
+        expires_in,
+        actor_token=None,
+        may_act=None,
+        subject_token_jti=None,
+        kid=None,
+        grants=None,
+    ):
+        """
+        Generate a signed JWT token according to RFC 8693.
+
+        Args:
+            sub: Subject identifier
+            email: Subject email
+            audiences: List of audience strings
+            scope: Space-separated scopes or list of scopes
+            expires_in: Token lifetime in seconds
+            actor_token: Optional actor token for delegation
+            may_act: Optional may_act claim for delegation
+            subject_token_jti: JTI of the original SSO token
+            kid: Key ID for signing
+            grants: Optional list of grant dicts with throttle info
+
+        Returns:
+            str: A signed JWT token
+
+        Raises:
+            ValueError: If kid is not provided or not found in signing keys
+        """
+        if not kid:
+            raise ValueError("kid is required for JWT token generation")
+
+        signing_keys = settings.TOKEN_EXCHANGE_JWT_SIGNING_KEYS
+        if kid not in signing_keys:
+            raise ValueError(
+                f"Key ID '{kid}' not found in TOKEN_EXCHANGE_JWT_SIGNING_KEYS"
+            )
+
+        # Prepare scope as list if it's a string
+        if isinstance(scope, str):
+            scope_list = scope.split() if scope else []
+        else:
+            scope_list = scope or []
+
+        # Prepare claims
+        now = timezone.now()
+        claims = {
+            "sub": sub,
+            "aud": audiences,
+            "scope": scope_list,
+            "exp": int((now + timedelta(seconds=expires_in)).timestamp()),
+            "iat": int(now.timestamp()),
+            "jti": str(uuid.uuid4()),
+        }
+
+        # Add email if provided
+        if email:
+            claims["email"] = email
+
+        # Add grants if provided
+        if grants:
+            claims["grants"] = grants
+
+        # Add optional RFC 8693 claims
+        if actor_token:
+            # Extract subject from actor_token if possible, or use a placeholder
+            claims["act"] = {"sub": "actor"}  # Simplified for now
+
+        if may_act:
+            claims["may_act"] = may_act
+
+        # Prepare header with kid
+        header = {
+            "kid": kid,
+            "alg": settings.TOKEN_EXCHANGE_JWT_ALGORITHM,
+        }
+
+        # Load the private key
+        private_key_pem = signing_keys[kid]
+        key = RSAKey.import_key(private_key_pem)
+
+        # Sign the JWT
+        token = jwt.encode(header, claims, key)
+        return token.decode("utf-8") if isinstance(token, bytes) else token
+
+    @staticmethod
+    def verify_jwt_token(token):
+        """
+        Verify a JWT token signature supporting multiple keys (rotation).
+
+        Args:
+            token: The JWT token string to verify
+
+        Returns:
+            dict: The decoded claims if valid
+
+        Raises:
+            ValueError: If token is invalid or signature verification fails
+        """
+        # Decode header to get kid
+        try:
+            # Decode without verification first to get the kid
+            unverified = jwt.decode(token, None)
+            kid = unverified.header.get("kid")
+
+            if not kid:
+                raise ValueError("JWT token missing 'kid' in header")
+
+            signing_keys = settings.TOKEN_EXCHANGE_JWT_SIGNING_KEYS
+            if kid not in signing_keys:
+                raise ValueError(
+                    f"Key ID '{kid}' not found in TOKEN_EXCHANGE_JWT_SIGNING_KEYS"
+                )
+
+            # Load the public key and verify
+            private_key_pem = signing_keys[kid]
+            key = RSAKey.import_key(private_key_pem)
+
+            # Verify and decode
+            claims = jwt.decode(token, key)
+            return claims
+
+        except Exception as exc:
+            raise ValueError(f"Invalid JWT token: {exc}") from exc

src/backend/token_exchange/urls.py

@@ -0,0 +1,13 @@
+"""URL configuration for the token_exchange application."""
+
+from django.urls import path
+
+from .views import TokenExchangeView, TokenIntrospectionView, TokenRevocationView
+
+urlpatterns = [
+    path("token/exchange/", TokenExchangeView.as_view(), name="token-exchange"),
+    path(
+        "token/introspect/", TokenIntrospectionView.as_view(), name="token-introspect"
+    ),
+    path("token/revoke/", TokenRevocationView.as_view(), name="token-revoke"),
+]

src/backend/token_exchange/views.py

@@ -0,0 +1,567 @@
+"""Views for the token_exchange application."""
+
+import logging
+from datetime import timedelta
+
+from django.conf import settings
+from django.core.exceptions import SuspiciousOperation
+from django.db.models import F
+from django.utils import timezone
+from django.utils.module_loading import import_string
+
+from lasuite.oidc_resource_server.backend import ResourceServerBackend
+from requests import HTTPError
+from rest_framework import status
+from rest_framework.parsers import FormParser, JSONParser
+from rest_framework.response import Response
+from rest_framework.throttling import ScopedRateThrottle
+from rest_framework.views import APIView
+
+from .authentication import ServiceProviderBasicAuthentication
+from .models import (
+    ActionScopeGrant,
+    ExchangedToken,
+    ScopeGrant,
+    TokenExchangeActionPermission,
+    TokenExchangeRule,
+    TokenTypeChoices,
+)
+from .permissions import IsServiceProviderAuthenticated
+from .serializers import TokenExchangeSerializer, TokenRevocationSerializer
+from .token_generator import TokenGenerator
+
+logger = logging.getLogger(__name__)
+
+
+def get_resource_server_introspection_backend() -> type[ResourceServerBackend]:
+    """Return the resource server backend class based on the settings."""
+    return import_string(settings.OIDC_RS_BACKEND_CLASS)
+
+
+class TokenExchangeView(APIView):
+    """
+    RFC 8693 Token Exchange endpoint.
+
+    This endpoint allows exchanging an external SSO token
+    for a new token with different audiences, scopes,
+    or lifetime.
+
+    POST /auth/token/exchange/
+        Exchange a token according to RFC 8693
+    """
+
+    authentication_classes = [ServiceProviderBasicAuthentication]
+    permission_classes = [IsServiceProviderAuthenticated]
+    throttle_classes = [ScopedRateThrottle]
+    parser_classes = [FormParser, JSONParser]
+    throttle_scope = "token_exchange"
+
+    @staticmethod
+    def generate_invalid_target_response():
+        """Generate the RFC invalid_target error response."""
+        return Response(
+            {
+                "error": "invalid_target",
+                "error_description": "Invalid target audience",
+            },
+            status=status.HTTP_400_BAD_REQUEST,
+        )
+
+    def post(self, request):  # noqa: PLR0911, PLR0912, PLR0915
+        """
+        Handle token exchange requests.
+
+        Validates the request, checks scopes, generates a new token,
+        and returns an RFC 8693 compliant response.
+        """
+        # Check if feature is enabled
+        if not settings.TOKEN_EXCHANGE_ENABLED:
+            return Response(
+                {
+                    "error": "token_exchange_disabled",
+                    "error_description": "Token exchange feature is not enabled",
+                },
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+
+        # Retrieve authenticated service provider
+        service_provider = request.user
+
+        # Validate request
+        serializer = TokenExchangeSerializer(data=request.data)
+        if not serializer.is_valid():
+            return Response(
+                {
+                    "error": "invalid_request",
+                    "error_description": str(serializer.errors),
+                },
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+
+        validated_data = serializer.validated_data
+
+        # Retrieve targeted service providers, by default it's the requesting one
+        requested_audiences = validated_data.get(
+            "audience",
+            [service_provider.audience_id],  # compared to introspection result later
+        )
+
+        rules = TokenExchangeRule.objects.filter(
+            source_service=service_provider,
+            target_service__audience_id__in=requested_audiences,
+        ).select_related("target_service")
+
+        # Sanity check: we check each requested audience has a rule (and therefore exists)
+        __known_audiences = {rule.target_service.audience_id for rule in rules}
+
+        if not __known_audiences:
+            logger.error(
+                "Only unknown audiences requested: %s",
+                ", ".join(requested_audiences),
+            )
+            return self.generate_invalid_target_response()
+
+        if __unknown_services := (set(requested_audiences) - __known_audiences):
+            logger.warning(
+                "Unknown audiences requested: %s",
+                ", ".join(__unknown_services),
+            )
+            return self.generate_invalid_target_response()
+
+        if any(not rule.is_active for rule in rules):
+            logger.warning(
+                "Some rules are inactive: %s",
+                ", ".join(str(rule.pk) for rule in rules if not rule.is_active),
+            )
+            return self.generate_invalid_target_response()
+
+        # Retrieve the user from subject_token, the only token type
+        # accepted for now is the access token -> needs introspection
+        subject_token = validated_data["subject_token"]
+
+        # XXX : manage the user impersonation case ???? another view ?
+        try:
+            introspection_backend = get_resource_server_introspection_backend()()
+            introspection_backend._scopes = [  # noqa: SLF001
+                "openid"
+            ]  # Prevent backend from enforcing scopes: MUST improve code here
+            user_info = introspection_backend.get_user_info_with_introspection(
+                subject_token
+            )
+            if (
+                introspection_backend.token_origin_audience
+                != service_provider.audience_id
+            ):
+                logger.error(
+                    "Introspected token origin is different from requesting service: %s, %s",
+                    introspection_backend.token_origin_audience,
+                    service_provider.audience_id,
+                )
+                raise SuspiciousOperation()
+        except HTTPError as exc:
+            logger.exception("Failed to introspect subject token: %s", exc)
+            raise
+
+        # Check the user audience is the same as the requesting service
+        if (
+            user_info.get(settings.OIDC_RS_AUDIENCE_CLAIM)
+            != service_provider.audience_id
+        ):
+            logger.error(
+                "Introspected token audience is different from requesting service: %s, %s",
+                user_info.get(settings.OIDC_RS_AUDIENCE_CLAIM),
+                service_provider.audience_id,
+            )
+            raise SuspiciousOperation()
+
+        # Extract subject token scope
+        subject_token_scope = user_info.get("scope", "")
+        subject_token_jti = user_info.get("jti", "unknown")
+
+        # Extract identity (sub or email)
+        subject_sub = user_info.get("sub")
+        subject_email = user_info.get("email")
+
+        if not subject_sub and not subject_email:
+            # We need at least one identity
+            logger.warning("Introspection response missing both 'sub' and 'email'")
+            return Response(
+                {
+                    "error": "invalid_token",
+                    "error_description": "Subject token introspection failed to provide identity (sub or email)",
+                },
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+
+        # Parse subject token scopes
+        subject_scopes = (
+            set(subject_token_scope.split()) if subject_token_scope else set()
+        )
+
+        # Parse requested scopes
+        # If not scope specifically required, we switch to "best-effort" mode by returning
+        # the same scopes as the subject token and ignoring the ones not allowed.
+        _missing_requested_scope_fails = True
+        requested_scopes_and_actions = set(validated_data.get("scope", []))
+        if not requested_scopes_and_actions:
+            requested_actions = None
+            requested_scopes = subject_scopes
+            _missing_requested_scope_fails = False
+        else:
+            requested_actions = {
+                s for s in requested_scopes_and_actions if s.startswith("action:")
+            }
+            requested_scopes = requested_scopes_and_actions - requested_actions
+
+        # Dev note: requested_actions is a list with at most one element to prevent
+        # action with different throttling to interfere. The serializer is currently
+        # doing the job to limit this, yet we add another check here.
+        if requested_actions:
+            if len(requested_actions) > 1:
+                logger.warning("Multiple actions requested: %s", requested_actions)
+                return self.generate_invalid_target_response()
+            if requested_scopes:
+                # TODO: allow distinct scope (ie not contained in action) to be requested.
+                logger.warning(
+                    "Action requested along common scopes: %s, %s",
+                    requested_actions,
+                    requested_scopes,
+                )
+                return self.generate_invalid_target_response()
+
+        # Build the global request matrix
+        requested_accesses = set()
+        for audience_id in requested_audiences:
+            for requested_scope in requested_scopes:
+                requested_accesses.add(f"{audience_id}:{requested_scope}")
+
+        # Get the global access rules:
+        existing_access = {}
+        for scope_grant in ScopeGrant.objects.filter(
+            rule__in=rules, source_scope__in=subject_scopes
+        ).annotate(
+            audience_id=F("rule__target_service__audience_id"),
+        ):
+            existing_access[
+                f"{scope_grant.audience_id}:{scope_grant.granted_scope}"
+            ] = scope_grant
+
+        if _missing_requested_scope_fails and not requested_accesses.issubset(
+            set(existing_access.keys())
+        ):
+            return self.generate_invalid_target_response()
+
+        # Compute granted scopes and grants list
+        granted_scopes = set()
+        granted_actions = set()
+        grants_dict = {}
+
+        # Process scope grants
+        for requested_access in requested_accesses:
+            try:
+                scope_grant = existing_access[requested_access]
+            except KeyError:
+                if _missing_requested_scope_fails:
+                    raise
+                continue
+
+            if scope_grant.source_scope not in subject_scopes:
+                logger.info("Missing user scope: %s", scope_grant.source_scope)
+                if _missing_requested_scope_fails:
+                    return self.generate_invalid_target_response()
+            granted_scopes.add(scope_grant.granted_scope)
+            audience_id = scope_grant.audience_id
+            grants_dict.setdefault(audience_id, []).append(
+                {
+                    "scope": scope_grant.granted_scope,
+                    "throttle": (
+                        {"rate": scope_grant.throttle_rate}
+                        if scope_grant.throttle_rate
+                        else {}
+                    ),
+                }
+            )
+
+        # Process action scopes
+        action_permissions = TokenExchangeActionPermission.objects.filter(
+            rule__in=rules
+        ).select_related("action")
+
+        for action_perm in action_permissions:
+            # Check if user has required source scope for this action
+            if action_perm.required_source_scope:
+                if action_perm.required_source_scope not in subject_scopes:
+                    continue  # User doesn't have required scope, skip this action
+
+            # Check if action is requested (action scopes are prefixed with "action:")
+            if requested_scopes and action_perm.action.name not in requested_scopes:
+                continue  # Action not requested, skip
+
+            # Grant all scopes associated with this action on the target service
+            action_grants = ActionScopeGrant.objects.filter(
+                action=action_perm.action,
+                target_service__audience_id__in=requested_audiences,
+            )
+            if action_grants:
+                granted_actions.add(action_perm.action.name)
+            for ag in action_grants:
+                granted_scopes.add(ag.granted_scope)
+                audience_id = ag.target_service.audience_id
+                grants_dict.setdefault(audience_id, []).append(
+                    {
+                        "scope": ag.granted_scope,
+                        "throttle": (
+                            {"rate": ag.throttle_rate} if ag.throttle_rate else {}
+                        ),
+                    }
+                )
+
+        # If scopes were requested, validate they're all granted
+        if (
+            requested_scopes
+            and not requested_scopes.issubset(granted_scopes)
+            and _missing_requested_scope_fails
+        ):
+            missing_scopes = requested_scopes - granted_scopes
+            logger.error(
+                "Requested scopes cannot be granted. Missing scopes: %s",
+                ", ".join(missing_scopes),
+            )
+            return self.generate_invalid_target_response()
+
+        if requested_actions and not requested_actions.issubset(granted_actions):
+            missing_actions = requested_actions - granted_actions
+            logger.error(
+                "Requested actions cannot be granted. Missing actions: %s",
+                ", ".join(missing_actions),
+            )
+            return self.generate_invalid_target_response()
+
+        if not granted_scopes:
+            logger.warning("No scopes granted: %s", requested_scopes_and_actions)
+            return self.generate_invalid_target_response()
+
+        # Convert granted scopes to string
+        final_scope = " ".join(sorted(granted_scopes)) if granted_scopes else ""
+
+        # Determine token type and format
+        token_type = (
+            validated_data.get("requested_token_type") or TokenTypeChoices.ACCESS_TOKEN
+        )
+
+        # Determine expires_in from rule or default
+        expires_in = (
+            validated_data.get("expires_in")
+            # XXX or int(rule.exchanged_token_duration.total_seconds())
+            or settings.TOKEN_EXCHANGE_DEFAULT_EXPIRES_IN
+        )
+
+        # Determine token audiences
+        if settings.TOKEN_EXCHANGE_MULTI_AUDIENCES_ALLOWED:
+            audiences = requested_audiences
+        else:
+            audiences = [requested_audiences[0]]
+
+        # Generate token
+        if token_type == TokenTypeChoices.JWT:
+            kid = settings.TOKEN_EXCHANGE_JWT_CURRENT_KID
+            if not kid:
+                return Response(
+                    {
+                        "error": "server_error",
+                        "error_description": "JWT signing key not configured",
+                    },
+                    status=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                )
+            try:
+                token_value = TokenGenerator.generate_jwt_token(
+                    sub=subject_sub,
+                    email=subject_email,
+                    audiences=audiences,
+                    scope=final_scope,
+                    expires_in=expires_in,
+                    actor_token=validated_data.get("actor_token"),
+                    may_act=None,  # TODO: Parse from actor_token if needed
+                    subject_token_jti=subject_token_jti,
+                    kid=kid,
+                    grants=grants_dict if grants_dict else None,
+                )
+            except ValueError as exc:
+                return Response(
+                    {"error": "server_error", "error_description": str(exc)},
+                    status=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                )
+        else:
+            token_value = TokenGenerator.generate_opaque_token()
+            kid = None
+
+        # Create ExchangedToken record
+        expires_at = timezone.now() + timedelta(seconds=expires_in)
+        _exchanged_token = ExchangedToken.objects.create(
+            token=token_value,
+            token_type=token_type,
+            jwt_kid=kid,
+            subject_sub=subject_sub,
+            subject_email=subject_email,
+            audiences=audiences,
+            scope=final_scope,
+            grants=grants_dict if grants_dict else None,
+            expires_at=expires_at,
+            actor_token=validated_data.get("actor_token"),
+            may_act=None,  # TODO: Parse from actor_token if needed
+            subject_token_jti=subject_token_jti,
+            subject_token_scope=subject_token_scope,
+        )
+
+        # Log the exchange
+        logger.info(
+            "Token exchanged: sub=%s, email=%s, audiences=%s, token_type=%s, "
+            "expires_in=%s, subject_jti=%s, kid=%s, scopes_granted=%s, grants=%s",
+            subject_sub,
+            subject_email,
+            audiences,
+            token_type,
+            expires_in,
+            subject_token_jti,
+            kid,
+            final_scope,
+            len(grants_dict),
+        )
+
+        # Return RFC 8693 response
+        response_data = {
+            "access_token": token_value,
+            "issued_token_type": f"urn:ietf:params:oauth:token-type:{token_type}",
+            "token_type": "Bearer",
+            "expires_in": expires_in,
+        }
+
+        if final_scope:
+            response_data["scope"] = final_scope
+
+        if grants_dict:
+            response_data["grants"] = grants_dict
+
+        return Response(response_data, status=status.HTTP_200_OK)
+
+
+class TokenIntrospectionView(APIView):
+    """
+    RFC 7662 Token Introspection endpoint.
+
+    This endpoint allows validating exchanged tokens and retrieving
+    their metadata.
+
+    POST /auth/token/introspect/
+        Introspect a token according to RFC 7662
+    """
+
+    authentication_classes = [ServiceProviderBasicAuthentication]
+    permission_classes = [IsServiceProviderAuthenticated]
+
+    def post(self, request):
+        """
+        Handle token introspection requests.
+
+        Accepts a token and returns its validity and metadata.
+        """
+        # Get token from request (form-encoded or JSON)
+        token = request.data.get("token")
+        if not token:
+            return Response(
+                {"active": False},
+                status=status.HTTP_200_OK,
+            )
+
+        # Look up the token
+        try:
+            exchanged_token = ExchangedToken.objects.get(token=token)
+        except ExchangedToken.DoesNotExist:
+            logger.info("Token introspected: token not found, active=False")
+            return Response({"active": False}, status=status.HTTP_200_OK)
+
+        # Check if valid
+        if not exchanged_token.is_valid():
+            logger.info(
+                "Token introspected: token_jti=%s, active=False, format=%s",
+                exchanged_token.subject_token_jti,
+                exchanged_token.token_type,
+            )
+            return Response({"active": False}, status=status.HTTP_200_OK)
+
+        # For JWT tokens, verify signature
+        if exchanged_token.token_type == TokenTypeChoices.JWT:
+            try:
+                TokenGenerator.verify_jwt_token(token)
+            except ValueError as exc:
+                logger.warning(
+                    "Token introspected: JWT signature verification failed: %s",
+                    str(exc),
+                )
+                return Response({"active": False}, status=status.HTTP_200_OK)
+
+        # Return introspection response
+        response_data = exchanged_token.to_introspection_response()
+
+        logger.info(
+            "Token introspected: token_jti=%s, active=%s, format=%s, kid=%s",
+            exchanged_token.subject_token_jti,
+            response_data["active"],
+            exchanged_token.token_type,
+            exchanged_token.jwt_kid or "N/A",
+        )
+
+        return Response(response_data, status=status.HTTP_200_OK)
+
+
+class TokenRevocationView(APIView):
+    """
+    RFC 7009 Token Revocation endpoint.
+
+    This endpoint allows revoking exchanged tokens before their
+    natural expiration.
+
+    POST /auth/token/revoke/
+        Revoke a token according to RFC 7009
+    """
+
+    authentication_classes = [ServiceProviderBasicAuthentication]
+    permission_classes = [IsServiceProviderAuthenticated]
+
+    def post(self, request):
+        """
+        Handle token revocation requests.
+
+        Accepts a token and revokes it.
+        """
+        # Validate request
+        serializer = TokenRevocationSerializer(data=request.data)
+        if not serializer.is_valid():
+            # RFC 7009: Return 200 even for invalid requests
+            return Response(status=status.HTTP_200_OK)
+
+        token = serializer.validated_data["token"]
+
+        # Look up the token
+        try:
+            exchanged_token = ExchangedToken.objects.get(
+                token=token,
+            )
+        except ExchangedToken.DoesNotExist:
+            # RFC 7009: Silent success even if token doesn't exist
+            logger.info("Token revocation attempted: token not found")
+            return Response(status=status.HTTP_200_OK)
+
+        # Revoke the token
+        exchanged_token.revoke()
+
+        logger.info(
+            "Token revoked: token_jti=%s, sub=%s, email=%s, type=%s, audiences=%s",
+            exchanged_token.subject_token_jti,
+            exchanged_token.subject_sub,
+            exchanged_token.subject_email,
+            exchanged_token.token_type,
+            exchanged_token.audiences,
+        )
+
+        return Response(status=status.HTTP_200_OK)

src/backend/uv.lock

@@ -1074,6 +1074,7 @@ dev = [
     { name = "pytest-cov" },
     { name = "pytest-django" },
     { name = "pytest-icdiff" },
+    { name = "pytest-unordered" },
     { name = "pytest-xdist" },
     { name = "responses" },
     { name = "ruff" },
@@ -1127,6 +1128,7 @@ requires-dist = [
     { name = "pytest-cov", marker = "extra == 'dev'", specifier = "==7.0.0" },
     { name = "pytest-django", marker = "extra == 'dev'", specifier = "==4.11.1" },
     { name = "pytest-icdiff", marker = "extra == 'dev'", specifier = "==0.9" },
+    { name = "pytest-unordered", marker = "extra == 'dev'", specifier = "==0.7.0" },
     { name = "pytest-xdist", marker = "extra == 'dev'", specifier = "==3.8.0" },
     { name = "redis", specifier = "<=7.1.0" },
     { name = "requests", specifier = "==2.32.5" },
@@ -1418,6 +1420,18 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/e2/e1/cafe1edf7a30be6fa1bbbf43f7af12b34682eadcf19eb6e9f7352062c422/pytest_icdiff-0.9-py3-none-any.whl", hash = "sha256:efee0da3bd1b24ef2d923751c5c547fbb8df0a46795553fba08ef57c3ca03d82", size = 4994, upload-time = "2023-12-05T11:18:28.572Z" },
 ]
 
+[[package]]
+name = "pytest-unordered"
+version = "0.7.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "pytest" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/bd/3e/6ec9ec74551804c9e005d5b3cbe1fd663f03ed3bd4bdb1ce764c3d334d8e/pytest_unordered-0.7.0.tar.gz", hash = "sha256:0f953a438db00a9f6f99a0f4727f2d75e72dd93319b3d548a97ec9db4903a44f", size = 7930, upload-time = "2025-06-03T12:56:04.289Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/5c/95/ae2875e19472797e9672b65412858ab6639d8e55defd9859241e5ff80d02/pytest_unordered-0.7.0-py3-none-any.whl", hash = "sha256:486b26d24a2d3b879a275c3d16d14eda1bd9c32aafddbb17b98ac755daba7584", size = 6210, upload-time = "2025-06-03T12:36:06.66Z" },
+]
+
 [[package]]
 name = "pytest-xdist"
 version = "3.8.0"