codekingpro/portable-devtools

1import logging2from authlib.jose import jwt, JoseError3from ..rfc6749 import BaseGrant, TokenEndpointMixin4from ..rfc6749 import (5    UnauthorizedClientError,6    InvalidRequestError,7    InvalidGrantError,8    InvalidClientError,9)10from .assertion import sign_jwt_bearer_assertion11 12log = logging.getLogger(__name__)13JWT_BEARER_GRANT_TYPE = 'urn:ietf:params:oauth:grant-type:jwt-bearer'14 15 16class JWTBearerGrant(BaseGrant, TokenEndpointMixin):17    GRANT_TYPE = JWT_BEARER_GRANT_TYPE18 19    #: Options for verifying JWT payload claims. Developers MAY20    #: overwrite this constant to create a more strict options.21    CLAIMS_OPTIONS = {22        'iss': {'essential': True},23        'aud': {'essential': True},24        'exp': {'essential': True},25    }26 27    @staticmethod28    def sign(key, issuer, audience, subject=None,29             issued_at=None, expires_at=None, claims=None, **kwargs):30        return sign_jwt_bearer_assertion(31            key, issuer, audience, subject, issued_at,32            expires_at, claims, **kwargs)33 34    def process_assertion_claims(self, assertion):35        """Extract JWT payload claims from request "assertion", per36        `Section 3.1`_.37 38        :param assertion: assertion string value in the request39        :return: JWTClaims40        :raise: InvalidGrantError41 42        .. _`Section 3.1`: https://tools.ietf.org/html/rfc7523#section-3.143        """44        try:45            claims = jwt.decode(46                assertion, self.resolve_public_key,47                claims_options=self.CLAIMS_OPTIONS)48            claims.validate()49        except JoseError as e:50            log.debug('Assertion Error: %r', e)51            raise InvalidGrantError(description=e.description)52        return claims53 54    def resolve_public_key(self, headers, payload):55        client = self.resolve_issuer_client(payload['iss'])56        return self.resolve_client_key(client, headers, payload)57 58    def validate_token_request(self):59        """The client makes a request to the token endpoint by sending the60        following parameters using the "application/x-www-form-urlencoded"61        format per `Section 2.1`_:62 63        grant_type64             REQUIRED.  Value MUST be set to65             "urn:ietf:params:oauth:grant-type:jwt-bearer".66 67        assertion68             REQUIRED.  Value MUST contain a single JWT.69 70        scope71            OPTIONAL.72 73        The following example demonstrates an access token request with a JWT74        as an authorization grant:75 76        .. code-block:: http77 78            POST /token.oauth2 HTTP/1.179            Host: as.example.com80            Content-Type: application/x-www-form-urlencoded81 82            grant_type=urn%3Aietf%3Aparams%3Aoauth%3Agrant-type%3Ajwt-bearer83            &assertion=eyJhbGciOiJFUzI1NiIsImtpZCI6IjE2In0.84            eyJpc3Mi[...omitted for brevity...].85            J9l-ZhwP[...omitted for brevity...]86 87        .. _`Section 2.1`: https://tools.ietf.org/html/rfc7523#section-2.188        """89        assertion = self.request.form.get('assertion')90        if not assertion:91            raise InvalidRequestError('Missing "assertion" in request')92 93        claims = self.process_assertion_claims(assertion)94        client = self.resolve_issuer_client(claims['iss'])95        log.debug('Validate token request of %s', client)96 97        if not client.check_grant_type(self.GRANT_TYPE):98            raise UnauthorizedClientError()99 100        self.request.client = client101        self.validate_requested_scope()102 103        subject = claims.get('sub')104        if subject:105            user = self.authenticate_user(subject)106            if not user:107                raise InvalidGrantError(description='Invalid "sub" value in assertion')108 109            log.debug('Check client(%s) permission to User(%s)', client, user)110            if not self.has_granted_permission(client, user):111                raise InvalidClientError(112                    description='Client has no permission to access user data')113            self.request.user = user114 115    def create_token_response(self):116        """If valid and authorized, the authorization server issues an access117        token.118        """119        token = self.generate_token(120            scope=self.request.scope,121            user=self.request.user,122            include_refresh_token=False,123        )124        log.debug('Issue token %r to %r', token, self.request.client)125        self.save_token(token)126        return 200, token, self.TOKEN_RESPONSE_HEADER127 128    def resolve_issuer_client(self, issuer):129        """Fetch client via "iss" in assertion claims. Developers MUST130        implement this method in subclass, e.g.::131 132            def resolve_issuer_client(self, issuer):133                return Client.query_by_iss(issuer)134 135        :param issuer: "iss" value in assertion136        :return: Client instance137        """138        raise NotImplementedError()139 140    def resolve_client_key(self, client, headers, payload):141        """Resolve client key to decode assertion data. Developers MUST142        implement this method in subclass. For instance, there is a143        "jwks" column on client table, e.g.::144 145            def resolve_client_key(self, client, headers, payload):146                # from authlib.jose import JsonWebKey147 148                key_set = JsonWebKey.import_key_set(client.jwks)149                return key_set.find_by_kid(headers['kid'])150 151        :param client: instance of OAuth client model152        :param headers: headers part of the JWT153        :param payload: payload part of the JWT154        :return: ``authlib.jose.Key`` instance155        """156        raise NotImplementedError()157 158    def authenticate_user(self, subject):159        """Authenticate user with the given assertion claims. Developers MUST160        implement it in subclass, e.g.::161 162            def authenticate_user(self, subject):163                return User.get_by_sub(subject)164 165        :param subject: "sub" value in claims166        :return: User instance167        """168        raise NotImplementedError()169 170    def has_granted_permission(self, client, user):171        """Check if the client has permission to access the given user's resource.172        Developers MUST implement it in subclass, e.g.::173 174            def has_granted_permission(self, client, user):175                permission = ClientUserGrant.query(client=client, user=user)176                return permission.granted177 178        :param client: instance of OAuth client model179        :param user: instance of User model180        :return: bool181        """182        raise NotImplementedError()183