bifunctor Posts - Answer Overflow

bifunctor

•Created by bifunctor on 1/27/2025 in #💻┃support

How to implement the token based authentication in GRPC interceptor with Python SDK?

I want to implement token-based authentication in a gRPC interceptor using the Python SDK. In the interceptor of a gRPC server’s request flow, I aim to validate the bearer token. Below is the code I have implemented so far: 

class AuthInterceptor(grpc.ServerInterceptor):
    def __init__(self, identity_provider: IdentityProvider):
        self.__identity_provider = identity_provider

    def intercept_service(self, continuation, handler_call_details):
        metadata: dict[str, any] = dict(handler_call_details.invocation_metadata)

        if "authorization" not in metadata:
            return self.__abort(StatusCode.UNAUTHENTICATED, "Authorization token is missing")

        token: str = metadata["authorization"]
        if not token.startswith("Bearer "):
            return self.__abort(StatusCode.UNAUTHENTICATED, "Invalid token format")

        hashed_token: str = token[7:]
        if len(hashed_token) == 0:
            return self.__abort(StatusCode.UNAUTHENTICATED, "Invalid token")

        # >>>>> Insert code to evaluate the bearer token here <<<<<

        return continuation(handler_call_details)

    def __abort(self, code, details):
        def _end_call(ignored_request, context):
            context.abort(code, details)

        return grpc.unary_unary_rpc_method_handler(_end_call)

class AuthInterceptor(grpc.ServerInterceptor):
    def __init__(self, identity_provider: IdentityProvider):
        self.__identity_provider = identity_provider

    def intercept_service(self, continuation, handler_call_details):
        metadata: dict[str, any] = dict(handler_call_details.invocation_metadata)

        if "authorization" not in metadata:
            return self.__abort(StatusCode.UNAUTHENTICATED, "Authorization token is missing")

        token: str = metadata["authorization"]
        if not token.startswith("Bearer "):
            return self.__abort(StatusCode.UNAUTHENTICATED, "Invalid token format")

        hashed_token: str = token[7:]
        if len(hashed_token) == 0:
            return self.__abort(StatusCode.UNAUTHENTICATED, "Invalid token")

        # >>>>> Insert code to evaluate the bearer token here <<<<<

        return continuation(handler_call_details)

    def __abort(self, code, details):
        def _end_call(ignored_request, context):
            context.abort(code, details)

        return grpc.unary_unary_rpc_method_handler(_end_call)

Questions: 1. How can I evaluate the validity of the bearer token? Kinde provides https://docs.kinde.com/developer-tools/sdks/backend/python-sdk/. However, I haven't found the method JWTverifiy as it exists for https://docs.kinde.com/developer-tools/sdks/backend/express-sdk/#verify-jwt 2. How can I test the implementation with a session token? I created a test user in the user management section but I am uncertain about how to obtain a session token for the user to include in the request for authentication. It seems the Python SDK (https://github.com/kinde-oss/kinde-python-sdk) does not directly provide this functionality.

3 replies

Gaming

Programming