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)