codekingpro/portable-devtools

1from authlib.consts import default_json_headers2from authlib.jose import JoseError3from ..rfc7591.claims import ClientMetadataClaims4from ..rfc6749 import scope_to_list5from ..rfc6749 import AccessDeniedError6from ..rfc6749 import InvalidClientError7from ..rfc6749 import InvalidRequestError8from ..rfc6749 import UnauthorizedClientError9from ..rfc7591 import InvalidClientMetadataError10 11 12class ClientConfigurationEndpoint:13    ENDPOINT_NAME = 'client_configuration'14 15    #: The claims validation class16    claims_class = ClientMetadataClaims17 18    def __init__(self, server):19        self.server = server20 21    def __call__(self, request):22        return self.create_configuration_response(request)23 24    def create_configuration_response(self, request):25        # This request is authenticated by the registration access token issued26        # to the client.27        token = self.authenticate_token(request)28        if not token:29            raise AccessDeniedError()30 31        request.credential = token32 33        client = self.authenticate_client(request)34        if not client:35            # If the client does not exist on this server, the server MUST respond36            # with HTTP 401 Unauthorized and the registration access token used to37            # make this request SHOULD be immediately revoked.38            self.revoke_access_token(request, token)39            raise InvalidClientError(status_code=401)40 41        if not self.check_permission(client, request):42            # If the client does not have permission to read its record, the server43            # MUST return an HTTP 403 Forbidden.44            raise UnauthorizedClientError(status_code=403)45 46        request.client = client47 48        if request.method == 'GET':49            return self.create_read_client_response(client, request)50        elif request.method == 'DELETE':51            return self.create_delete_client_response(client, request)52        elif request.method == 'PUT':53            return self.create_update_client_response(client, request)54 55    def create_endpoint_request(self, request):56        return self.server.create_json_request(request)57 58    def create_read_client_response(self, client, request):59        body = self.introspect_client(client)60        body.update(self.generate_client_registration_info(client, request))61        return 200, body, default_json_headers62 63    def create_delete_client_response(self, client, request):64        self.delete_client(client, request)65        headers = [66            ('Cache-Control', 'no-store'),67            ('Pragma', 'no-cache'),68        ]69        return 204, '', headers70 71    def create_update_client_response(self, client, request):72        # The updated client metadata fields request MUST NOT include the73        # 'registration_access_token', 'registration_client_uri',74        # 'client_secret_expires_at', or 'client_id_issued_at' fields75        must_not_include = (76            'registration_access_token',77            'registration_client_uri',78            'client_secret_expires_at',79            'client_id_issued_at',80        )81        for k in must_not_include:82            if k in request.data:83                raise InvalidRequestError()84 85        # The client MUST include its 'client_id' field in the request86        client_id = request.data.get('client_id')87        if not client_id:88            raise InvalidRequestError()89        if client_id != client.get_client_id():90            raise InvalidRequestError()91 92        # If the client includes the 'client_secret' field in the request,93        # the value of this field MUST match the currently issued client94        # secret for that client.95        if 'client_secret' in request.data:96            if not client.check_client_secret(request.data['client_secret']):97                raise InvalidRequestError()98 99        client_metadata = self.extract_client_metadata(request)100        client = self.update_client(client, client_metadata, request)101        return self.create_read_client_response(client, request)102 103    def extract_client_metadata(self, request):104        json_data = request.data.copy()105        options = self.get_claims_options()106        claims = self.claims_class(json_data, {}, options, self.get_server_metadata())107 108        try:109            claims.validate()110        except JoseError as error:111            raise InvalidClientMetadataError(error.description)112        return claims.get_registered_claims()113 114    def get_claims_options(self):115        metadata = self.get_server_metadata()116        if not metadata:117            return {}118 119        scopes_supported = metadata.get('scopes_supported')120        response_types_supported = metadata.get('response_types_supported')121        grant_types_supported = metadata.get('grant_types_supported')122        auth_methods_supported = metadata.get('token_endpoint_auth_methods_supported')123        options = {}124        if scopes_supported is not None:125            scopes_supported = set(scopes_supported)126 127            def _validate_scope(claims, value):128                if not value:129                    return True130                scopes = set(scope_to_list(value))131                return scopes_supported.issuperset(scopes)132 133            options['scope'] = {'validate': _validate_scope}134 135        if response_types_supported is not None:136            response_types_supported = set(response_types_supported)137 138            def _validate_response_types(claims, value):139                return response_types_supported.issuperset(set(value))140 141            options['response_types'] = {'validate': _validate_response_types}142 143        if grant_types_supported is not None:144            grant_types_supported = set(grant_types_supported)145 146            def _validate_grant_types(claims, value):147                return grant_types_supported.issuperset(set(value))148 149            options['grant_types'] = {'validate': _validate_grant_types}150 151        if auth_methods_supported is not None:152            options['token_endpoint_auth_method'] = {'values': auth_methods_supported}153 154        return options155 156    def introspect_client(self, client):157        return {**client.client_info, **client.client_metadata}158 159    def generate_client_registration_info(self, client, request):160        """Generate ```registration_client_uri`` and ``registration_access_token``161        for RFC7592. By default this method returns the values sent in the current162        request. Developers MUST rewrite this method to return different registration163        information.::164 165            def generate_client_registration_info(self, client, request):{166                access_token = request.headers['Authorization'].split(' ')[1]167                return {168                    'registration_client_uri': request.uri,169                    'registration_access_token': access_token,170                }171 172        :param client: the instance of OAuth client173        :param request: formatted request instance174        """175        raise NotImplementedError()176 177    def authenticate_token(self, request):178        """Authenticate current credential who is requesting to register a client.179        Developers MUST implement this method in subclass::180 181            def authenticate_token(self, request):182                auth = request.headers.get('Authorization')183                return get_token_by_auth(auth)184 185        :return: token instance186        """187        raise NotImplementedError()188 189    def authenticate_client(self, request):190        """Read a client from the request payload.191        Developers MUST implement this method in subclass::192 193            def authenticate_client(self, request):194                client_id = request.data.get('client_id')195                return Client.get(client_id=client_id)196 197        :return: client instance198        """199        raise NotImplementedError()200 201    def revoke_access_token(self, token, request):202        """Revoke a token access in case an invalid client has been requested.203        Developers MUST implement this method in subclass::204 205            def revoke_access_token(self, token, request):206                token.revoked = True207                token.save()208 209        """210        raise NotImplementedError()211 212    def check_permission(self, client, request):213        """Checks wether the current client is allowed to be accessed, edited214        or deleted. Developers MUST implement it in subclass, e.g.::215 216            def check_permission(self, client, request):217                return client.editable218 219        :return: boolean220        """221        raise NotImplementedError()222 223    def delete_client(self, client, request):224        """Delete authorization code from database or cache. Developers MUST225        implement it in subclass, e.g.::226 227            def delete_client(self, client, request):228                client.delete()229 230        :param client: the instance of OAuth client231        :param request: formatted request instance232        """233        raise NotImplementedError()234 235    def update_client(self, client, client_metadata, request):236        """Update the client in the database. Developers MUST implement this method237        in subclass::238 239            def update_client(self, client, client_metadata, request):240                client.set_client_metadata({**client.client_metadata, **client_metadata})241                client.save()242                return client243 244        :param client: the instance of OAuth client245        :param client_metadata: a dict of the client claims to update246        :param request: formatted request instance247        :return: client instance248        """249 250        raise NotImplementedError()251 252    def get_server_metadata(self):253        """Return server metadata which includes supported grant types,254        response types and etc.255        """256        raise NotImplementedError()257