codekingpro/portable-devtools

1import secrets2 3from sqlalchemy import Column, String, Text, Integer4from authlib.common.encoding import json_loads, json_dumps5from authlib.oauth2.rfc6749 import ClientMixin6from authlib.oauth2.rfc6749 import scope_to_list, list_to_scope7 8 9class OAuth2ClientMixin(ClientMixin):10    client_id = Column(String(48), index=True)11    client_secret = Column(String(120))12    client_id_issued_at = Column(Integer, nullable=False, default=0)13    client_secret_expires_at = Column(Integer, nullable=False, default=0)14    _client_metadata = Column('client_metadata', Text)15 16    @property17    def client_info(self):18        """Implementation for Client Info in OAuth 2.0 Dynamic Client19        Registration Protocol via `Section 3.2.1`_.20 21        .. _`Section 3.2.1`: https://tools.ietf.org/html/rfc7591#section-3.2.122        """23        return dict(24            client_id=self.client_id,25            client_secret=self.client_secret,26            client_id_issued_at=self.client_id_issued_at,27            client_secret_expires_at=self.client_secret_expires_at,28        )29 30    @property31    def client_metadata(self):32        if 'client_metadata' in self.__dict__:33            return self.__dict__['client_metadata']34        if self._client_metadata:35            data = json_loads(self._client_metadata)36            self.__dict__['client_metadata'] = data37            return data38        return {}39 40    def set_client_metadata(self, value):41        self._client_metadata = json_dumps(value)42        if 'client_metadata' in self.__dict__:43            del self.__dict__['client_metadata']44 45    @property46    def redirect_uris(self):47        return self.client_metadata.get('redirect_uris', [])48 49    @property50    def token_endpoint_auth_method(self):51        return self.client_metadata.get(52            'token_endpoint_auth_method',53            'client_secret_basic'54        )55 56    @property57    def grant_types(self):58        return self.client_metadata.get('grant_types', [])59 60    @property61    def response_types(self):62        return self.client_metadata.get('response_types', [])63 64    @property65    def client_name(self):66        return self.client_metadata.get('client_name')67 68    @property69    def client_uri(self):70        return self.client_metadata.get('client_uri')71 72    @property73    def logo_uri(self):74        return self.client_metadata.get('logo_uri')75 76    @property77    def scope(self):78        return self.client_metadata.get('scope', '')79 80    @property81    def contacts(self):82        return self.client_metadata.get('contacts', [])83 84    @property85    def tos_uri(self):86        return self.client_metadata.get('tos_uri')87 88    @property89    def policy_uri(self):90        return self.client_metadata.get('policy_uri')91 92    @property93    def jwks_uri(self):94        return self.client_metadata.get('jwks_uri')95 96    @property97    def jwks(self):98        return self.client_metadata.get('jwks', [])99 100    @property101    def software_id(self):102        return self.client_metadata.get('software_id')103 104    @property105    def software_version(self):106        return self.client_metadata.get('software_version')107 108    def get_client_id(self):109        return self.client_id110 111    def get_default_redirect_uri(self):112        if self.redirect_uris:113            return self.redirect_uris[0]114 115    def get_allowed_scope(self, scope):116        if not scope:117            return ''118        allowed = set(self.scope.split())119        scopes = scope_to_list(scope)120        return list_to_scope([s for s in scopes if s in allowed])121 122    def check_redirect_uri(self, redirect_uri):123        return redirect_uri in self.redirect_uris124 125    def check_client_secret(self, client_secret):126        return secrets.compare_digest(self.client_secret, client_secret)127 128    def check_endpoint_auth_method(self, method, endpoint):129        if endpoint == 'token':130            return self.token_endpoint_auth_method == method131        # TODO132        return True133 134    def check_response_type(self, response_type):135        return response_type in self.response_types136 137    def check_grant_type(self, grant_type):138        return grant_type in self.grant_types139