codekingpro/portable-devtools

1import time2from typing import List3from typing import Optional4from typing import Union5 6from authlib.common.security import generate_token7from authlib.jose import jwt8from authlib.oauth2.rfc6750.token import BearerTokenGenerator9 10 11class JWTBearerTokenGenerator(BearerTokenGenerator):12    '''A JWT formatted access token generator.13 14    :param issuer: The issuer identifier. Will appear in the JWT ``iss`` claim.15 16    :param \\*\\*kwargs: Other parameters are inherited from17        :class:`~authlib.oauth2.rfc6750.token.BearerTokenGenerator`.18 19    This token generator can be registered into the authorization server::20 21        class MyJWTBearerTokenGenerator(JWTBearerTokenGenerator):22            def get_jwks(self):23                ...24 25            def get_extra_claims(self, client, grant_type, user, scope):26                ...27 28        authorization_server.register_token_generator(29            'default',30            MyJWTBearerTokenGenerator(issuer='https://authorization-server.example.org'),31        )32    '''33 34    def __init__(35        self,36        issuer,37        alg='RS256',38        refresh_token_generator=None,39        expires_generator=None,40    ):41        super().__init__(42            self.access_token_generator, refresh_token_generator, expires_generator43        )44        self.issuer = issuer45        self.alg = alg46 47    def get_jwks(self):48        '''Return the JWKs that will be used to sign the JWT access token.49        Developers MUST re-implement this method::50 51            def get_jwks(self):52                return load_jwks("jwks.json")53        '''54        raise NotImplementedError()55 56    def get_extra_claims(self, client, grant_type, user, scope):57        '''Return extra claims to add in the JWT access token. Developers MAY58        re-implement this method to add identity claims like the ones in59        :ref:`specs/oidc` ID Token, or any other arbitrary claims::60 61            def get_extra_claims(self, client, grant_type, user, scope):62                return generate_user_info(user, scope)63        '''64        return {}65 66    def get_audiences(self, client, user, scope) -> Union[str, List[str]]:67        '''Return the audience for the token. By default this simply returns68        the client ID. Developpers MAY re-implement this method to add extra69        audiences::70 71            def get_audiences(self, client, user, scope):72                return [73                    client.get_client_id(),74                    resource_server.get_id(),75                ]76        '''77        return client.get_client_id()78 79    def get_acr(self, user) -> Optional[str]:80        '''Authentication Context Class Reference.81        Returns a user-defined case sensitive string indicating the class of82        authentication the used performed. Token audience may refuse to give access to83        some resources if some ACR criterias are not met.84        :ref:`specs/oidc` defines one special value: ``0`` means that the user85        authentication did not respect `ISO29115`_ level 1, and will be refused monetary86        operations. Developers MAY re-implement this method::87 88            def get_acr(self, user):89                if user.insecure_session():90                    return '0'91                return 'urn:mace:incommon:iap:silver'92 93        .. _ISO29115: https://www.iso.org/standard/45138.html94        '''95        return None96 97    def get_auth_time(self, user) -> Optional[int]:98        '''User authentication time.99        Time when the End-User authentication occurred. Its value is a JSON number100        representing the number of seconds from 1970-01-01T0:0:0Z as measured in UTC101        until the date/time. Developers MAY re-implement this method::102 103            def get_auth_time(self, user):104                return datetime.timestamp(user.get_auth_time())105        '''106        return None107 108    def get_amr(self, user) -> Optional[List[str]]:109        '''Authentication Methods References.110        Defined by :ref:`specs/oidc` as an option list of user-defined case-sensitive111        strings indication which authentication methods have been used to authenticate112        the user. Developers MAY re-implement this method::113 114            def get_amr(self, user):115                return ['2FA'] if user.has_2fa_enabled() else []116        '''117        return None118 119    def get_jti(self, client, grant_type, user, scope) -> str:120        '''JWT ID.121        Create an unique identifier for the token. Developers MAY re-implement122        this method::123 124            def get_jti(self, client, grant_type, user scope):125                return generate_random_string(16)126        '''127        return generate_token(16)128 129    def access_token_generator(self, client, grant_type, user, scope):130        now = int(time.time())131        expires_in = now + self._get_expires_in(client, grant_type)132 133        token_data = {134            'iss': self.issuer,135            'exp': expires_in,136            'client_id': client.get_client_id(),137            'iat': now,138            'jti': self.get_jti(client, grant_type, user, scope),139            'scope': scope,140        }141 142        # In cases of access tokens obtained through grants where a resource owner is143        # involved, such as the authorization code grant, the value of 'sub' SHOULD144        # correspond to the subject identifier of the resource owner.145 146        if user:147            token_data['sub'] = user.get_user_id()148 149        # In cases of access tokens obtained through grants where no resource owner is150        # involved, such as the client credentials grant, the value of 'sub' SHOULD151        # correspond to an identifier the authorization server uses to indicate the152        # client application.153 154        else:155            token_data['sub'] = client.get_client_id()156 157        # If the request includes a 'resource' parameter (as defined in [RFC8707]), the158        # resulting JWT access token 'aud' claim SHOULD have the same value as the159        # 'resource' parameter in the request.160 161        # TODO: Implement this with RFC8707162        if False:  # pragma: no cover163            ...164 165        # If the request does not include a 'resource' parameter, the authorization166        # server MUST use a default resource indicator in the 'aud' claim. If a 'scope'167        # parameter is present in the request, the authorization server SHOULD use it to168        # infer the value of the default resource indicator to be used in the 'aud'169        # claim. The mechanism through which scopes are associated with default resource170        # indicator values is outside the scope of this specification.171 172        else:173            token_data['aud'] = self.get_audiences(client, user, scope)174 175        # If the values in the 'scope' parameter refer to different default resource176        # indicator values, the authorization server SHOULD reject the request with177        # 'invalid_scope' as described in Section 4.1.2.1 of [RFC6749].178        # TODO: Implement this with RFC8707179 180        if auth_time := self.get_auth_time(user):181            token_data['auth_time'] = auth_time182 183        # The meaning and processing of acr Claim Values is out of scope for this184        # specification.185 186        if acr := self.get_acr(user):187            token_data['acr'] = acr188 189        # The definition of particular values to be used in the amr Claim is beyond the190        # scope of this specification.191 192        if amr := self.get_amr(user):193            token_data['amr'] = amr194 195        # Authorization servers MAY return arbitrary attributes not defined in any196        # existing specification, as long as the corresponding claim names are collision197        # resistant or the access tokens are meant to be used only within a private198        # subsystem. Please refer to Sections 4.2 and 4.3 of [RFC7519] for details.199 200        token_data.update(self.get_extra_claims(client, grant_type, user, scope))201 202        # This specification registers the 'application/at+jwt' media type, which can203        # be used to indicate that the content is a JWT access token. JWT access tokens204        # MUST include this media type in the 'typ' header parameter to explicitly205        # declare that the JWT represents an access token complying with this profile.206        # Per the definition of 'typ' in Section 4.1.9 of [RFC7515], it is RECOMMENDED207        # that the 'application/' prefix be omitted. Therefore, the 'typ' value used208        # SHOULD be 'at+jwt'.209 210        header = {'alg': self.alg, 'typ': 'at+jwt'}211 212        access_token = jwt.encode(213            header,214            token_data,215            key=self.get_jwks(),216            check=False,217        )218        return access_token.decode()219