Emit RFC 9207 iss authorization-response parameter

authserver/authserver/settings.py

@@ -165,6 +165,15 @@
     # This is here because django-oauth-toolkit checks for it in oauth2_provider.models, but we provide the keys
     # from the Domain object associated with the JWT
     'OIDC_RSA_PRIVATE_KEY': '<unused>',
+    # RFC 9207: we emit the `iss` authorization-response parameter ourselves
+    # (see mailauth/oauth2_backends.py) because upstream's issuer derivation
+    # cannot produce our per-host 'https://<host>/o2' issuers. Pinned False so
+    # the django-oauth-toolkit 4.0 default flip cannot turn on the upstream
+    # emission (which would produce a mismatching 'https://<host>' issuer and
+    # break RFC 9207-validating clients). Do not remove without reading
+    # .agent-docs/dcr-cimd-authz-iss/implementation-plan.rst section 5.
+    'COMPLIANT_BCP_RFC9700_AUTHZ_RESPONSE_ISS': False,
+    'OAUTH2_BACKEND_CLASS': 'mailauth.oauth2_backends.MNOAuthLibCore',
 }
 
 # we use our own modular crypt format sha256 hasher for maximum compatibility

authserver/authserver/urls.py

@@ -3,7 +3,6 @@
 from django.contrib.auth import views as auth_views
 
 from oauth2_provider import views as oauth2_views
-from oauth2_provider.views import oidc as oidc_views
 from authserver import base_views
 from authserver import selfservice_views
 from mailauth import views as mail_views
@@ -18,7 +17,7 @@
     re_path(r'^revoke_token/$', oauth2_views.RevokeTokenView.as_view(), name='revoke-token'),
     re_path(r'^fake-userinfo/$', mail_views.FakeUserInfoView.as_view(), name='fake-user-info'),
     re_path(r'^userinfo/$', oauth2_views.UserInfoView.as_view(), name='user-info'),
-    re_path(r'^\.well-known/openid-configuration/?$', oidc_views.ConnectDiscoveryInfoView.as_view(),
+    re_path(r'^\.well-known/openid-configuration/?$', mail_views.MNConnectDiscoveryInfoView.as_view(),
             name='oidc-connect-discovery-info'),
     re_path(r"^\.well-known/jwks.json$", mail_views.JwksInfoView.as_view(), name="jwks-info")
 ], 'oauth2_provider')
@@ -34,9 +33,9 @@
     # /.well-known/oauth-authorization-server/o2 for it. The plain form (issuer "https://<host>")
     # is served as well for clients that don't implement RFC 8414 issuer path handling.
     re_path(r'^\.well-known/oauth-authorization-server/?$',
-            oauth2_views.OAuthServerMetadataView.as_view(), name='oauth-server-metadata'),
+            mail_views.MNOAuthServerMetadataView.as_view(), name='oauth-server-metadata'),
     path('.well-known/oauth-authorization-server/<path:issuer_path>',
-         oauth2_views.OAuthServerMetadataView.as_view(), name='oauth-server-metadata-issuer'),
+         mail_views.MNOAuthServerMetadataView.as_view(), name='oauth-server-metadata-issuer'),
     re_path(r'^$', selfservice_views.HomeView.as_view(), name='selfservice-home'),
     re_path(r'^dashboard/$', selfservice_views.DashboardView.as_view(), name='selfservice-dashboard'),
     re_path(r'^action/login/$', selfservice_views.SelfServiceLoginView.as_view(), name='authserver-login'),

authserver/mailauth/oauth2_backends.py

@@ -0,0 +1,23 @@
+"""Custom OAuthLib request/response glue for authserver."""
+from oauth2_provider.oauth2_backends import OAuthLibCore, _add_iss_to_redirect
+from oauth2_provider.settings import oauth2_settings
+
+
+class MNOAuthLibCore(OAuthLibCore):
+    """
+    Adds the RFC 9207 `iss` parameter to successful authorization
+    responses using the same per-request issuer that the OIDC discovery
+    document advertises (https://<host>/o2). Upstream's implementation
+    (gated by COMPLIANT_BCP_RFC9700_AUTHZ_RESPONSE_ISS, pinned False in
+    settings) cannot derive path-component multi-tenant issuers.
+    """
+
+    def create_authorization_response(self, request, scopes, credentials, allow):
+        uri, headers, body, status = super().create_authorization_response(
+            request, scopes, credentials, allow
+        )
+        if uri is not None:
+            issuer = oauth2_settings.oidc_issuer(request)
+            uri = _add_iss_to_redirect(uri, issuer)
+            headers["Location"] = uri
+        return uri, headers, body, status

authserver/mailauth/tests/test_authz_response_iss.py

@@ -0,0 +1,146 @@
+import base64
+import hashlib
+import json
+import secrets
+from urllib.parse import parse_qs, urlsplit
+
+from django.test import TestCase
+from django.urls import reverse
+from oauth2_provider.settings import oauth2_settings
+
+from mailauth import models
+from mailauth.utils import generate_rsa_key
+
+
+CLIENT_SECRET = "sekrit-test-client-secret"
+REDIRECT_URI = "https://client.example.com/callback"
+REDIRECT_URI_WITH_ISS = "https://client.example.com/callback?iss=https://evil.example"
+
+
+class RFC9207AuthzResponseIssTests(TestCase):
+    """
+    RFC 9207 `iss` authorization-response parameter tests. Reuses the
+    domain/user/app setup pattern from test_oauth2_oidc.py.
+    """
+
+    @classmethod
+    def setUpTestData(cls) -> None:
+        cls.key = generate_rsa_key()
+        cls.domain = models.Domain.objects.create(name="example.com", jwtkey=cls.key.private_key)
+
+        cls.user = models.MNUser.objects.create_user(
+            identifier="testuser", fullname="Test User", password="secret"
+        )
+        cls.alias = models.EmailAlias.objects.create(
+            user=cls.user, domain=cls.domain, mailprefix="testuser"
+        )
+        cls.user.delivery_mailbox = cls.alias
+        cls.user.save()
+
+        cls.app = models.MNApplication.objects.create(
+            name="testclient",
+            client_type=models.MNApplication.CLIENT_CONFIDENTIAL,
+            authorization_grant_type=models.MNApplication.GRANT_AUTHORIZATION_CODE,
+            redirect_uris=REDIRECT_URI,
+            skip_authorization=True,
+            domain=cls.domain,
+            client_secret=CLIENT_SECRET,
+            hash_client_secret=False,
+        )
+
+        cls.app_with_iss_redirect = models.MNApplication.objects.create(
+            name="testclient-iss-redirect",
+            client_type=models.MNApplication.CLIENT_CONFIDENTIAL,
+            authorization_grant_type=models.MNApplication.GRANT_AUTHORIZATION_CODE,
+            redirect_uris=REDIRECT_URI_WITH_ISS,
+            skip_authorization=True,
+            domain=cls.domain,
+            client_secret=CLIENT_SECRET,
+            hash_client_secret=False,
+        )
+
+    def _authorize(self, app, redirect_uri, code_verifier: str, state: "str | None" = None):
+        code_challenge = (
+            base64.urlsafe_b64encode(hashlib.sha256(code_verifier.encode("ascii")).digest())
+            .rstrip(b"=")
+            .decode("ascii")
+        )
+        self.client.force_login(self.user)
+        params = {
+            "response_type": "code",
+            "client_id": app.client_id,
+            "redirect_uri": redirect_uri,
+            "scope": "openid profile email",
+            "code_challenge": code_challenge,
+            "code_challenge_method": "S256",
+        }
+        if state is not None:
+            params["state"] = state
+        return self.client.get(reverse("oauth2_provider:authorize"), params, secure=True)
+
+    def _redeem_code(self, app, code: str, redirect_uri: str, code_verifier: str) -> dict:
+        response = self.client.post(
+            reverse("oauth2_provider:token"),
+            {
+                "grant_type": "authorization_code",
+                "code": code,
+                "redirect_uri": redirect_uri,
+                "client_id": app.client_id,
+                "client_secret": CLIENT_SECRET,
+                "code_verifier": code_verifier,
+            },
+            secure=True,
+        )
+        self.assertEqual(200, response.status_code, response.content)
+        return json.loads(response.content)
+
+    def test_authorize_redirect_contains_iss_query_parameter_once(self) -> None:
+        code_verifier = secrets.token_urlsafe(48)
+        response = self._authorize(self.app, REDIRECT_URI, code_verifier)
+
+        self.assertEqual(302, response.status_code)
+        self.assertTrue(response.url.startswith(REDIRECT_URI))
+        query = parse_qs(urlsplit(response.url).query)
+        self.assertEqual(["https://testserver/o2"], query["iss"])
+        self.assertIn("code", query)
+
+    def test_full_code_flow_completes_with_iss_present(self) -> None:
+        code_verifier = secrets.token_urlsafe(48)
+        state = "opaque-state-value"
+        response = self._authorize(self.app, REDIRECT_URI, code_verifier, state=state)
+
+        self.assertEqual(302, response.status_code)
+        query = parse_qs(urlsplit(response.url).query)
+        self.assertEqual(["https://testserver/o2"], query["iss"])
+        self.assertEqual([state], query["state"])
+        code = query["code"][0]
+
+        token_data = self._redeem_code(self.app, code, REDIRECT_URI, code_verifier)
+        self.assertIn("access_token", token_data)
+        self.assertIn("id_token", token_data)
+
+    def test_redirect_uri_with_preexisting_iss_param_is_overridden(self) -> None:
+        code_verifier = secrets.token_urlsafe(48)
+        response = self._authorize(self.app_with_iss_redirect, REDIRECT_URI_WITH_ISS, code_verifier)
+
+        self.assertEqual(302, response.status_code)
+        query = parse_qs(urlsplit(response.url).query)
+        # only the server's value survives, exactly once
+        self.assertEqual(["https://testserver/o2"], query["iss"])
+        self.assertIn("code", query)
+
+    def test_discovery_documents_advertise_iss_parameter_support(self) -> None:
+        oidc_response = self.client.get("/o2/.well-known/openid-configuration", secure=True)
+        self.assertEqual(200, oidc_response.status_code)
+        oidc_data = json.loads(oidc_response.content)
+        self.assertIs(True, oidc_data["authorization_response_iss_parameter_supported"])
+
+        rfc8414_response = self.client.get(
+            "/.well-known/oauth-authorization-server/o2", secure=True
+        )
+        self.assertEqual(200, rfc8414_response.status_code)
+        rfc8414_data = json.loads(rfc8414_response.content)
+        self.assertIs(True, rfc8414_data["authorization_response_iss_parameter_supported"])
+
+    def test_upstream_iss_emission_gate_is_pinned_off(self) -> None:
+        self.assertIs(False, oauth2_settings.COMPLIANT_BCP_RFC9700_AUTHZ_RESPONSE_ISS)

authserver/mailauth/views.py

@@ -18,6 +18,8 @@
 from oauth2_provider.forms import AllowForm
 from oauth2_provider.models import get_application_model
 from oauth2_provider.views import ProtectedResourceView
+from oauth2_provider import views as oauth2_views
+from oauth2_provider.views import oidc as oidc_views
 from oauth2_provider.views.base import AuthorizationView
 from oauth2_provider.settings import oauth2_settings
 from django_ratelimit.decorators import ratelimit
@@ -461,3 +463,31 @@ def get(self, request: HttpRequest) -> HttpResponse:
         )
         response["Access-Control-Allow-Origin"] = "*"
         return response
+
+
+class IssAdvertisingDiscoveryMixin:
+    """
+    Advertises RFC 9207 support in discovery documents. Upstream only sets
+    ``authorization_response_iss_parameter_supported`` when
+    ``COMPLIANT_BCP_RFC9700_AUTHZ_RESPONSE_ISS`` is True; authserver pins that
+    gate False and emits the `iss` parameter itself (mailauth.oauth2_backends
+    .MNOAuthLibCore), so the flag must be advertised independently here.
+    """
+
+    def get(self, request: HttpRequest, *args: Any, **kwargs: Any) -> HttpResponse:
+        response = super().get(request, *args, **kwargs)  # type: ignore[misc]
+        if response.status_code == 200:
+            data = json.loads(response.content)
+            data["authorization_response_iss_parameter_supported"] = True
+            new_response = JsonResponse(data)
+            new_response["Access-Control-Allow-Origin"] = "*"
+            return new_response
+        return response
+
+
+class MNConnectDiscoveryInfoView(IssAdvertisingDiscoveryMixin, oidc_views.ConnectDiscoveryInfoView):
+    pass
+
+
+class MNOAuthServerMetadataView(IssAdvertisingDiscoveryMixin, oauth2_views.OAuthServerMetadataView):
+    pass