codekingpro/portable-devtools

1import time2 3 4def create_query_client_func(session, client_model):5    """Create an ``query_client`` function that can be used in authorization6    server.7 8    :param session: SQLAlchemy session9    :param client_model: Client model class10    """11    def query_client(client_id):12        q = session.query(client_model)13        return q.filter_by(client_id=client_id).first()14    return query_client15 16 17def create_save_token_func(session, token_model):18    """Create an ``save_token`` function that can be used in authorization19    server.20 21    :param session: SQLAlchemy session22    :param token_model: Token model class23    """24    def save_token(token, request):25        if request.user:26            user_id = request.user.get_user_id()27        else:28            user_id = None29        client = request.client30        item = token_model(31            client_id=client.client_id,32            user_id=user_id,33            **token34        )35        session.add(item)36        session.commit()37    return save_token38 39 40def create_query_token_func(session, token_model):41    """Create an ``query_token`` function for revocation, introspection42    token endpoints.43 44    :param session: SQLAlchemy session45    :param token_model: Token model class46    """47    def query_token(token, token_type_hint):48        q = session.query(token_model)49        if token_type_hint == 'access_token':50            return q.filter_by(access_token=token).first()51        elif token_type_hint == 'refresh_token':52            return q.filter_by(refresh_token=token).first()53        # without token_type_hint54        item = q.filter_by(access_token=token).first()55        if item:56            return item57        return q.filter_by(refresh_token=token).first()58    return query_token59 60 61def create_revocation_endpoint(session, token_model):62    """Create a revocation endpoint class with SQLAlchemy session63    and token model.64 65    :param session: SQLAlchemy session66    :param token_model: Token model class67    """68    from authlib.oauth2.rfc7009 import RevocationEndpoint69    query_token = create_query_token_func(session, token_model)70 71    class _RevocationEndpoint(RevocationEndpoint):72        def query_token(self, token, token_type_hint):73            return query_token(token, token_type_hint)74 75        def revoke_token(self, token, request):76            now = int(time.time())77            hint = request.form.get('token_type_hint')78            token.access_token_revoked_at = now79            if hint != 'access_token':80                token.refresh_token_revoked_at = now81            session.add(token)82            session.commit()83 84    return _RevocationEndpoint85 86 87def create_bearer_token_validator(session, token_model):88    """Create an bearer token validator class with SQLAlchemy session89    and token model.90 91    :param session: SQLAlchemy session92    :param token_model: Token model class93    """94    from authlib.oauth2.rfc6750 import BearerTokenValidator95 96    class _BearerTokenValidator(BearerTokenValidator):97        def authenticate_token(self, token_string):98            q = session.query(token_model)99            return q.filter_by(access_token=token_string).first()100 101    return _BearerTokenValidator102