My gRPC server application needed to extract a JWT token from each request being made by a Javascript client.
The JWT itself arrived at the application by means of the request metadata (basically as a request header), in the format:
Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzIiO...
I created a ServerInterceptor
to extract the JWT from the metadata:
import grpc from google.rpc import code_pb2 class JwtInterceptor(grpc.ServerInterceptor): def __init__(self): def _terminator(): def terminate(_, context): context.abort(code_pb2.PERMISSION_DENIED, 'Permission denied') return grpc.unary_unary_rpc_method_handler(terminate) self.terminator = _terminator() def intercept_service(self, continuation, handler_call_details): for key, value in handler_call_details.invocation_metadata: if key.lower() == 'authorization': # Split the header value into ['Bearer', 'token'] if self.authenticate(value.split()[1]): return continuation(handler_call_details) else: # Authentication failure return self.terminator # No authorization header found - terminate the request return self.terminator def authenticate(self, jwt): # Perform authentication
The authenticate()
method can be modified to decode the JWT and run some authentication logic on it. In my case I delegated that to a third party library.
I then wired the interceptor into the gRPC server at the point it’s created:
from concurrent import futures import grpc server = grpc.server(futures.ThreadPoolExecutor( max_workers=10, interceptors=[JwtInterceptor()] ) server.add_insecure_port('[::]:9802')