feat: implement rfc9068 JWT Access Tokens

README.md

@@ -52,6 +52,7 @@ Generic, spec-compliant implementation to build clients and providers:
   - [RFC7662: OAuth 2.0 Token Introspection](https://docs.authlib.org/en/latest/specs/rfc7662.html)
   - [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)
 - [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/integrations/django_oauth2/resource_protector.py

@@ -15,26 +15,32 @@
 
 
 class ResourceProtector(_ResourceProtector):
-    def acquire_token(self, request, scopes=None):
+    def acquire_token(self, request, scopes=None, **kwargs):
         """A method to acquire current valid token with the given scope.
 
         :param request: Django HTTP request instance
         :param scopes: a list of scope values
         :return: token object
         """
         req = DjangoJsonRequest(request)
-        if isinstance(scopes, str):
-            scopes = [scopes]
-        token = self.validate_request(scopes, req)
+        # backward compatibility
+        kwargs['scopes'] = scopes
+        for claim in kwargs:
+            if isinstance(kwargs[claim], str):
+                kwargs[claim] = [kwargs[claim]]
+        token = self.validate_request(request=req, **kwargs)
         token_authenticated.send(sender=self.__class__, token=token)
         return token
 
-    def __call__(self, scopes=None, optional=False):
+    def __call__(self, scopes=None, optional=False, **kwargs):
+        claims = kwargs
+        # backward compatibility
+        claims['scopes'] = scopes
         def wrapper(f):
             @functools.wraps(f)
             def decorated(request, *args, **kwargs):
                 try:
-                    token = self.acquire_token(request, scopes)
+                    token = self.acquire_token(request, **claims)
                     request.oauth_token = token
                 except MissingAuthorizationError as error:
                     if optional:

authlib/integrations/flask_oauth2/resource_protector.py

@@ -54,17 +54,19 @@ def raise_error_response(self, error):
         headers = error.get_headers()
         raise_http_exception(status, body, headers)
 
-    def acquire_token(self, scopes=None):
+    def acquire_token(self, scopes=None, **kwargs):
         """A method to acquire current valid token with the given scope.
 
         :param scopes: a list of scope values
         :return: token object
         """
         request = FlaskJsonRequest(_req)
-        # backward compatible
-        if isinstance(scopes, str):
-            scopes = [scopes]
-        token = self.validate_request(scopes, request)
+        # backward compatibility
+        kwargs['scopes'] = scopes
+        for claim in kwargs:
+            if isinstance(kwargs[claim], str):
+                kwargs[claim] = [kwargs[claim]]
+        token = self.validate_request(request=request, **kwargs)
         token_authenticated.send(self, token=token)
         g.authlib_server_oauth2_token = token
         return token
@@ -85,12 +87,15 @@ def user_api():
         except OAuth2Error as error:
             self.raise_error_response(error)
 
-    def __call__(self, scopes=None, optional=False):
+    def __call__(self, scopes=None, optional=False, **kwargs):
+        claims = kwargs
+        # backward compatibility
+        claims['scopes'] = scopes
         def wrapper(f):
             @functools.wraps(f)
             def decorated(*args, **kwargs):
                 try:
-                    self.acquire_token(scopes)
+                    self.acquire_token(**claims)
                 except MissingAuthorizationError as error:
                     if optional:
                         return f(*args, **kwargs)

authlib/jose/errors.py

@@ -82,6 +82,7 @@ class InvalidClaimError(JoseError):
     error = 'invalid_claim'
 
     def __init__(self, claim):
+        self.claim_name = claim
         description = f'Invalid claim "{claim}"'
         super().__init__(description=description)
 

authlib/jose/rfc7519/jwt.py

@@ -50,7 +50,7 @@ def encode(self, header, payload, key, check=True):
         :param check: check if sensitive data in payload
         :return: bytes
         """
-        header['typ'] = 'JWT'
+        header.setdefault('typ', 'JWT')
 
         for k in ['exp', 'iat', 'nbf']:
             # convert datetime into timestamp

authlib/oauth2/rfc6749/authorization_server.py

@@ -179,16 +179,21 @@ def authenticate_user(self, credential):
         if hasattr(grant_cls, 'check_token_endpoint'):
             self._token_grants.append((grant_cls, extensions))
 
-    def register_endpoint(self, endpoint_cls):
+    def register_endpoint(self, endpoint):
         """Add extra endpoint to authorization server. e.g.
         RevocationEndpoint::
 
             authorization_server.register_endpoint(RevocationEndpoint)
 
-        :param endpoint_cls: A endpoint class
+        :param endpoint_cls: A endpoint class or instance.
         """
-        endpoints = self._endpoints.setdefault(endpoint_cls.ENDPOINT_NAME, [])
-        endpoints.append(endpoint_cls(self))
+        if isinstance(endpoint, type):
+            endpoint = endpoint(self)
+        else:
+            endpoint.server = self
+
+        endpoints = self._endpoints.setdefault(endpoint.ENDPOINT_NAME, [])
+        endpoints.append(endpoint)
 
     def get_authorization_grant(self, request):
         """Find the authorization grant for current request.

authlib/oauth2/rfc6749/resource_protector.py

@@ -131,10 +131,10 @@ def parse_request_authorization(self, request):
         validator = self.get_token_validator(token_type)
         return validator, token_string
 
-    def validate_request(self, scopes, request):
+    def validate_request(self, scopes, request, **kwargs):
         """Validate the request and return a token."""
         validator, token_string = self.parse_request_authorization(request)
         validator.validate_request(request)
         token = validator.authenticate_token(token_string)
-        validator.validate_token(token, scopes, request)
+        validator.validate_token(token, scopes, request, **kwargs)
         return token

authlib/oauth2/rfc7009/revocation.py

@@ -27,17 +27,19 @@ def authenticate_token(self, request, client):
             OPTIONAL.  A hint about the type of the token submitted for
             revocation.
         """
+        self.check_params(request, client)
+        token = self.query_token(request.form['token'], request.form.get('token_type_hint'))
+        if token and token.check_client(client):
+            return token
+
+    def check_params(self, request, client):
         if 'token' not in request.form:
             raise InvalidRequestError()
 
         hint = request.form.get('token_type_hint')
         if hint and hint not in self.SUPPORTED_TOKEN_TYPES:
             raise UnsupportedTokenTypeError()
 
-        token = self.query_token(request.form['token'], hint)
-        if token and token.check_client(client):
-            return token
-
     def create_endpoint_response(self, request):
         """Validate revocation request and create the response for revocation.
         For example, a client may request the revocation of a refresh token

authlib/oauth2/rfc7662/introspection.py

@@ -34,6 +34,13 @@ def authenticate_token(self, request, client):
             **OPTIONAL**  A hint about the type of the token submitted for
             introspection.
         """
+
+        self.check_params(request, client)
+        token = self.query_token(request.form['token'], request.form.get('token_type_hint'))
+        if token and self.check_permission(token, client, request):
+            return token
+
+    def check_params(self, request, client):
         params = request.form
         if 'token' not in params:
             raise InvalidRequestError()
@@ -42,10 +49,6 @@ def authenticate_token(self, request, client):
         if hint and hint not in self.SUPPORTED_TOKEN_TYPES:
             raise UnsupportedTokenTypeError()
 
-        token = self.query_token(params['token'], hint)
-        if token and self.check_permission(token, client, request):
-            return token
-
     def create_endpoint_response(self, request):
         """Validate introspection request and create the response.
 

authlib/oauth2/rfc9068/__init__.py

@@ -0,0 +1,11 @@
+from .introspection import JWTIntrospectionEndpoint
+from .revocation import JWTRevocationEndpoint
+from .token import JWTBearerTokenGenerator
+from .token_validator import JWTBearerTokenValidator
+
+__all__ = [
+    'JWTBearerTokenGenerator',
+    'JWTBearerTokenValidator',
+    'JWTIntrospectionEndpoint',
+    'JWTRevocationEndpoint',
+]

authlib/oauth2/rfc9068/claims.py

@@ -0,0 +1,62 @@
+from authlib.jose.errors import InvalidClaimError
+from authlib.jose.rfc7519 import JWTClaims
+
+
+class JWTAccessTokenClaims(JWTClaims):
+    REGISTERED_CLAIMS = JWTClaims.REGISTERED_CLAIMS + [
+        'client_id',
+        'auth_time',
+        'acr',
+        'amr',
+        'scope',
+        'groups',
+        'roles',
+        'entitlements',
+    ]
+
+    def validate(self, **kwargs):
+        self.validate_typ()
+
+        super().validate(**kwargs)
+        self.validate_client_id()
+        self.validate_auth_time()
+        self.validate_acr()
+        self.validate_amr()
+        self.validate_scope()
+        self.validate_groups()
+        self.validate_roles()
+        self.validate_entitlements()
+
+    def validate_typ(self):
+        # The resource server MUST verify that the 'typ' header value is 'at+jwt'
+        # or 'application/at+jwt' and reject tokens carrying any other value.
+        if self.header['typ'].lower() not in ('at+jwt', 'application/at+jwt'):
+            raise InvalidClaimError('typ')
+
+    def validate_client_id(self):
+        return self._validate_claim_value('client_id')
+
+    def validate_auth_time(self):
+        auth_time = self.get('auth_time')
+        if auth_time and not isinstance(auth_time, (int, float)):
+            raise InvalidClaimError('auth_time')
+
+    def validate_acr(self):
+        return self._validate_claim_value('acr')
+
+    def validate_amr(self):
+        amr = self.get('amr')
+        if amr and not isinstance(self['amr'], list):
+            raise InvalidClaimError('amr')
+
+    def validate_scope(self):
+        return self._validate_claim_value('scope')
+
+    def validate_groups(self):
+        return self._validate_claim_value('groups')
+
+    def validate_roles(self):
+        return self._validate_claim_value('roles')
+
+    def validate_entitlements(self):
+        return self._validate_claim_value('entitlements')

authlib/oauth2/rfc9068/introspection.py

@@ -0,0 +1,126 @@
+from ..rfc7662 import IntrospectionEndpoint
+from authlib.common.errors import ContinueIteration
+from authlib.consts import default_json_headers
+from authlib.jose.errors import ExpiredTokenError
+from authlib.jose.errors import InvalidClaimError
+from authlib.oauth2.rfc6750.errors import InvalidTokenError
+from authlib.oauth2.rfc9068.token_validator import JWTBearerTokenValidator
+
+
+class JWTIntrospectionEndpoint(IntrospectionEndpoint):
+    '''
+    JWTIntrospectionEndpoint inherits from :ref:`specs/rfc7662`
+    :class:`~authlib.oauth2.rfc7662.IntrospectionEndpoint` and implements the machinery
+    to automatically process the JWT access tokens.
+
+    :param issuer: The issuer identifier for which tokens will be introspected.
+
+    :param \\*\\*kwargs: Other parameters are inherited from
+        :class:`~authlib.oauth2.rfc7662.introspection.IntrospectionEndpoint`.
+
+    ::
+
+        class MyJWTAccessTokenIntrospectionEndpoint(JWTRevocationEndpoint):
+            def get_jwks(self):
+                ...
+
+            def get_username(self, user_id):
+                ...
+
+        authorization_server.register_endpoint(
+            MyJWTAccessTokenIntrospectionEndpoint(
+                issuer="https://authorization-server.example.org",
+            )
+        )
+        authorization_server.register_endpoint(MyRefreshTokenIntrospectionEndpoint)
+
+    '''
+
+    #: Endpoint name to be registered
+    ENDPOINT_NAME = 'introspection'
+
+    def __init__(self, issuer, server=None, *args, **kwargs):
+        super().__init__(*args, server=server, **kwargs)
+        self.issuer = issuer
+
+    def create_endpoint_response(self, request):
+        ''''''
+        # The authorization server first validates the client credentials
+        client = self.authenticate_endpoint_client(request)
+
+        # then verifies whether the token was issued to the client making
+        # the revocation request
+        token = self.authenticate_token(request, client)
+
+        # the authorization server invalidates the token
+        body = self.create_introspection_payload(token)
+        return 200, body, default_json_headers
+
+    def authenticate_token(self, request, client):
+        ''''''
+        self.check_params(request, client)
+
+        # do not attempt to decode refresh_tokens
+        if request.form.get('token_type_hint') not in ('access_token', None):
+            raise ContinueIteration()
+
+        validator = JWTBearerTokenValidator(issuer=self.issuer, resource_server=None)
+        validator.get_jwks = self.get_jwks
+        try:
+            token = validator.authenticate_token(request.form['token'])
+
+        # if the token is not a JWT, fall back to the regular flow
+        except InvalidTokenError:
+            raise ContinueIteration()
+
+        if token and self.check_permission(token, client, request):
+            return token
+
+    def create_introspection_payload(self, token):
+        if not token:
+            return {'active': False}
+
+        try:
+            token.validate()
+        except ExpiredTokenError:
+            return {'active': False}
+        except InvalidClaimError as exc:
+            if exc.claim_name == 'iss':
+                raise ContinueIteration()
+            raise InvalidTokenError()
+
+
+        payload = {
+            'active': True,
+            'token_type': 'Bearer',
+            'client_id': token['client_id'],
+            'scope': token['scope'],
+            'sub': token['sub'],
+            'aud': token['aud'],
+            'iss': token['iss'],
+            'exp': token['exp'],
+            'iat': token['iat'],
+        }
+
+        if username := self.get_username(token['sub']):
+            payload['username'] = username
+
+        return payload
+
+    def get_jwks(self):
+        '''Return the JWKs that will be used to check the JWT access token signature.
+        Developers MUST re-implement this method::
+
+            def get_jwks(self):
+                return load_jwks("jwks.json")
+        '''
+        raise NotImplementedError()
+
+    def get_username(self, user_id: str) -> str:
+        '''Returns an username from a user ID.
+        Developers MAY re-implement this method::
+
+            def get_username(self, user_id):
+                return User.get(id=user_id).username
+        '''
+        return None