gRPC interceptor for extracting a JWT token

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')