codekingpro/portable-devtools

1from authlib.consts import default_json_headers2from ..rfc6749 import (3    TokenEndpoint,4    InvalidRequestError,5    UnsupportedTokenTypeError,6)7 8 9class IntrospectionEndpoint(TokenEndpoint):10    """Implementation of introspection endpoint which is described in11    `RFC7662`_.12 13    .. _RFC7662: https://tools.ietf.org/html/rfc766214    """15    #: Endpoint name to be registered16    ENDPOINT_NAME = 'introspection'17 18    def authenticate_token(self, request, client):19        """The protected resource calls the introspection endpoint using an HTTP20        ``POST`` request with parameters sent as21        "application/x-www-form-urlencoded" data. The protected resource sends a22        parameter representing the token along with optional parameters23        representing additional context that is known by the protected resource24        to aid the authorization server in its response.25 26        token27            **REQUIRED**  The string value of the token. For access tokens, this28            is the ``access_token`` value returned from the token endpoint29            defined in OAuth 2.0. For refresh tokens, this is the30            ``refresh_token`` value returned from the token endpoint as defined31            in OAuth 2.0.32 33        token_type_hint34            **OPTIONAL**  A hint about the type of the token submitted for35            introspection.36        """37 38        self.check_params(request, client)39        token = self.query_token(request.form['token'], request.form.get('token_type_hint'))40        if token and self.check_permission(token, client, request):41            return token42 43    def check_params(self, request, client):44        params = request.form45        if 'token' not in params:46            raise InvalidRequestError()47 48        hint = params.get('token_type_hint')49        if hint and hint not in self.SUPPORTED_TOKEN_TYPES:50            raise UnsupportedTokenTypeError()51 52    def create_endpoint_response(self, request):53        """Validate introspection request and create the response.54 55        :returns: (status_code, body, headers)56        """57        # The authorization server first validates the client credentials58        client = self.authenticate_endpoint_client(request)59 60        # then verifies whether the token was issued to the client making61        # the revocation request62        token = self.authenticate_token(request, client)63 64        # the authorization server invalidates the token65        body = self.create_introspection_payload(token)66        return 200, body, default_json_headers67 68    def create_introspection_payload(self, token):69        # the token is not active, does not exist on this server, or the70        # protected resource is not allowed to introspect this particular71        # token, then the authorization server MUST return an introspection72        # response with the "active" field set to "false"73        if not token:74            return {'active': False}75        if token.is_expired() or token.is_revoked():76            return {'active': False}77        payload = self.introspect_token(token)78        if 'active' not in payload:79            payload['active'] = True80        return payload81 82    def check_permission(self, token, client, request):83        """Check if the request has permission to introspect the token. Developers84        MUST implement this method::85 86            def check_permission(self, token, client, request):87                # only allow a special client to introspect the token88                return client.client_id == 'introspection_client'89 90        :return: bool91        """92        raise NotImplementedError()93 94    def query_token(self, token_string, token_type_hint):95        """Get the token from database/storage by the given token string.96        Developers should implement this method::97 98            def query_token(self, token_string, token_type_hint):99                if token_type_hint == 'access_token':100                    tok = Token.query_by_access_token(token_string)101                elif token_type_hint == 'refresh_token':102                    tok = Token.query_by_refresh_token(token_string)103                else:104                    tok = Token.query_by_access_token(token_string)105                    if not tok:106                        tok = Token.query_by_refresh_token(token_string)107                return tok108        """109        raise NotImplementedError()110 111    def introspect_token(self, token):112        """Read given token and return its introspection metadata as a113        dictionary following `Section 2.2`_::114 115            def introspect_token(self, token):116                return {117                    'active': True,118                    'client_id': token.client_id,119                    'token_type': token.token_type,120                    'username': get_token_username(token),121                    'scope': token.get_scope(),122                    'sub': get_token_user_sub(token),123                    'aud': token.client_id,124                    'iss': 'https://server.example.com/',125                    'exp': token.expires_at,126                    'iat': token.issued_at,127                }128 129        .. _`Section 2.2`: https://tools.ietf.org/html/rfc7662#section-2.2130        """131        raise NotImplementedError()132