Server-side RFC9207 implementation

README.md

@@ -51,6 +51,7 @@ Generic, spec-compliant implementation to build clients and providers:
   - [RFC8414: OAuth 2.0 Authorization Server Metadata](https://docs.authlib.org/en/latest/specs/rfc8414.html)
   - [RFC8628: OAuth 2.0 Device Authorization Grant](https://docs.authlib.org/en/latest/specs/rfc8628.html)
   - [RFC9068: JSON Web Token (JWT) Profile for OAuth 2.0 Access Tokens](https://docs.authlib.org/en/latest/specs/rfc9068.html)
+  - [RFC9207: OAuth 2.0 Authorization Server Issuer Identification](https://docs.authlib.org/en/latest/specs/rfc9207.html)
 - [Javascript Object Signing and Encryption](https://docs.authlib.org/en/latest/jose/index.html)
   - [RFC7515: JSON Web Signature](https://docs.authlib.org/en/latest/jose/jws.html)
   - [RFC7516: JSON Web Encryption](https://docs.authlib.org/en/latest/jose/jwe.html)

authlib/oauth2/rfc6749/authorization_server.py

@@ -268,9 +268,12 @@ def create_authorization_response(self, request=None, grant_user=None):
         try:
             redirect_uri = grant.validate_authorization_request()
             args = grant.create_authorization_response(redirect_uri, grant_user)
-            return self.handle_response(*args)
+            response = self.handle_response(*args)
         except OAuth2Error as error:
-            return self.handle_error_response(request, error)
+            response = self.handle_error_response(request, error)
+
+        grant.execute_hook('after_authorization_response', response)
+        return response
 
     def create_token_response(self, request=None):
         """Validate token request and create token response.

authlib/oauth2/rfc6749/grants/base.py

@@ -24,6 +24,7 @@ def __init__(self, request: OAuth2Request, server):
         self.server = server
         self._hooks = {
             'after_validate_authorization_request': set(),
+            'after_authorization_response': set(),
             'after_validate_consent_request': set(),
             'after_validate_token_request': set(),
             'process_token': set(),

authlib/oauth2/rfc9207/__init__.py

@@ -0,0 +1,3 @@
+from .parameter import IssuerParameter
+
+__all__ = ["IssuerParameter"]

authlib/oauth2/rfc9207/parameter.py

@@ -0,0 +1,29 @@
+from authlib.common.urls import add_params_to_uri
+from typing import Optional
+
+
+class IssuerParameter:
+    def __call__(self, grant):
+        grant.register_hook(
+            'after_authorization_response',
+            self.add_issuer_parameter,
+        )
+
+    def add_issuer_parameter(self, hook_type : str, response):
+        if self.get_issuer():
+            # RFC9207 §2
+            # In authorization responses to the client, including error responses,
+            # an authorization server supporting this specification MUST indicate
+            # its identity by including the iss parameter in the response.
+
+            new_location = add_params_to_uri(response.location, {"iss": self.get_issuer()})
+            response.location += new_location
+
+    def get_issuer(self) -> Optional[str]:
+        """Return the issuer URL.
+        Developers MAY implement this method if they want to support :rfc:`RFC9207 <9207>`::
+
+            def get_issuer(self) -> str:
+                return "https://auth.example.org"
+        """
+        return None

docs/changelog.rst

@@ -11,6 +11,7 @@ Version 1.x.x
 
 **Unreleased**
 
+- Implement server-side :rfc:`RFC9207 <9207>`. :issue:`700`
 - ``generate_id_token`` can take a ``kid`` parmaeter. :pr:`702`
 
 Version 1.4.1

docs/specs/index.rst

@@ -26,5 +26,6 @@ works.
     rfc8037
     rfc8414
     rfc8628
+    rfc9207
     rfc9068
     oidc

docs/specs/rfc9207.rst

@@ -0,0 +1,30 @@
+.. _specs/rfc9207:
+
+RFC9207: OAuth 2.0 Authorization Server Issuer Identification
+=============================================================
+
+This section contains the generic implementation of :rfc:`RFC9207 <9207>`.
+
+In summary, RFC9207 advise to return an ``iss`` parameter in authorization code responses.
+This can simply be done by implementing the :meth:`~authlib.oauth2.rfc9207.parameter.IssuerParameter.get_issuer` method in the :class:`~authlib.oauth2.rfc9207.parameter.IssuerParameter` class,
+and pass it as a :class:`~authlib.oauth2.rfc6749.grants.AuthorizationCodeGrant` extension::
+
+    from authlib.oauth2.rfc6749.parameter import IssuerParameter as _IssuerParameter
+
+    class IssuerParameter(_IssuerParameter):
+        def get_issuer(self) -> str:
+            return "https://auth.example.org"
+
+    ...
+
+    authorization_server.register_grant(AuthorizationCodeGrant, [IssuerParameter()])
+
+API Reference
+-------------
+
+.. module:: authlib.oauth2.rfc9207
+
+.. autoclass:: IssuerParameter
+    :member-order: bysource
+    :members:
+

tests/flask/test_oauth2/test_authorization_code_iss_parameter.py

@@ -0,0 +1,98 @@
+from authlib.oauth2.rfc6749.grants import (
+    AuthorizationCodeGrant as _AuthorizationCodeGrant,
+)
+from .models import db, User, Client
+from .models import CodeGrantMixin, save_authorization_code
+from .oauth2_server import TestCase
+from .oauth2_server import create_authorization_server
+from authlib.oauth2.rfc9207 import IssuerParameter as _IssuerParameter
+
+
+class AuthorizationCodeGrant(CodeGrantMixin, _AuthorizationCodeGrant):
+    TOKEN_ENDPOINT_AUTH_METHODS = ['client_secret_basic', 'client_secret_post', 'none']
+
+    def save_authorization_code(self, code, request):
+        return save_authorization_code(code, request)
+
+
+class IssuerParameter(_IssuerParameter):
+    def get_issuer(self) -> str:
+        return "https://auth.test"
+
+
+class RFC9207AuthorizationCodeTest(TestCase):
+    LAZY_INIT = False
+
+    def prepare_data(
+            self, is_confidential=True,
+            response_type='code', grant_type='authorization_code',
+            token_endpoint_auth_method='client_secret_basic', rfc9207=True):
+        server = create_authorization_server(self.app, self.LAZY_INIT)
+        extensions = [IssuerParameter()] if rfc9207 else []
+        server.register_grant(AuthorizationCodeGrant, extensions=extensions)
+        self.server = server
+
+        user = User(username='foo')
+        db.session.add(user)
+        db.session.commit()
+
+        if is_confidential:
+            client_secret = 'code-secret'
+        else:
+            client_secret = ''
+        client = Client(
+            user_id=user.id,
+            client_id='code-client',
+            client_secret=client_secret,
+        )
+        client.set_client_metadata({
+            'redirect_uris': ['https://a.b'],
+            'scope': 'profile address',
+            'token_endpoint_auth_method': token_endpoint_auth_method,
+            'response_types': [response_type],
+            'grant_types': grant_type.splitlines(),
+        })
+        self.authorize_url = (
+            '/oauth/authorize?response_type=code'
+            '&client_id=code-client'
+        )
+        db.session.add(client)
+        db.session.commit()
+
+    def test_rfc9207_enabled_success(self):
+        """Check that when RFC9207 is implemented,
+        the authorization response has an ``iss`` parameter."""
+
+        self.prepare_data(rfc9207=True)
+        url = self.authorize_url + '&state=bar'
+        rv = self.client.post(url, data={'user_id': '1'})
+        self.assertIn('iss=https%3A%2F%2Fauth.test', rv.location)
+
+    def test_rfc9207_disabled_success_no_iss(self):
+        """Check that when RFC9207 is not implemented,
+        the authorization response contains no ``iss`` parameter."""
+
+        self.prepare_data(rfc9207=False)
+        url = self.authorize_url + '&state=bar'
+        rv = self.client.post(url, data={'user_id': '1'})
+        self.assertNotIn('iss=', rv.location)
+
+    def test_rfc9207_enabled_error(self):
+        """Check that when RFC9207 is implemented,
+        the authorization response has an ``iss`` parameter,
+        even when an error is returned."""
+
+        self.prepare_data(rfc9207=True)
+        rv = self.client.post(self.authorize_url)
+        self.assertIn('error=access_denied', rv.location)
+        self.assertIn('iss=https%3A%2F%2Fauth.test', rv.location)
+
+    def test_rfc9207_disbled_error_no_iss(self):
+        """Check that when RFC9207 is not implemented,
+        the authorization response contains no ``iss`` parameter,
+        even when an error is returned."""
+
+        self.prepare_data(rfc9207=False)
+        rv = self.client.post(self.authorize_url)
+        self.assertIn('error=access_denied', rv.location)
+        self.assertNotIn('iss=', rv.location)