K
Kinde•2mo ago
bifunctor

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.
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.
2 Replies
Ages
Ages•2mo ago
Hi @bifunctor Thanks for reach out. To implement token-based authentication in a gRPC server using Python, you can create a custom interceptor to validate bearer tokens. Here's a concise guide: - Define a class that inherits from grpc.ServerInterceptor and override the intercept_service method to handle authentication. - In the intercept_service method, extract the Authorization header from the request metadata. Ensure the token starts with "Bearer " and is followed by a valid JWT. - Use a JWT library to decode and verify the token's signature and claims. Ensure the token is not expired and contains the necessary claims. - If the token is missing, invalid, or expired, abort the request with a UNAUTHENTICATED status. - To test interceptor, you can use a gRPC client that includes a valid JWT in the Authorization header. Ensure the token is signed with the same secret key used in the interceptor. Note: - Replace 'HS256' and self.secret_key with the appropriate algorithm and secret key used in your application. - Ensure the jwt library is installed (pip install pyjwt). For more detailed information on implementing token-based authentication in gRPC with Python, refer to the gRPC Python examples and the Kinde Python SDK documentation. Let me know if you need help with this process or if there’s anything else I can assist with.
GitHub
grpc/examples/python/auth/token_based_auth_client.py at master · gr...
The C based gRPC (C++, Python, Ruby, Objective-C, PHP, C#) - grpc/grpc
Ages
Ages•2mo ago
Hi @bifunctor , Just following up—are you still facing this issue? Have you had a chance to try the suggested approach? If you have any questions or need further assistance, let us know. Otherwise, we’ll close this query for now.

Did you find this page helpful?