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)

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.

Kinde docs

Python SDK

Our developer tools provide everything you need to get started with Kinde.

GitHub

GitHub - kinde-oss/kinde-python-sdk: Kinde SDK for Python

Kinde SDK for Python. Contribute to kinde-oss/kinde-python-sdk development by creating an account on GitHub.

Kinde docs

Express.js SDK

Our developer tools provide everything you need to get started with Kinde.

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

Similar Threads

Similar Threads

Similar Threads