gRPC in Python. Part 4: Interceptors

Code

The code for this article is available at: https://github.com/theundeadmonk/python-grpc-demo/

This is part of a series of gRPC in python. We will cover the following

1. [Implementing a server](https://adityamattos.com/grpc-in-python-part-1-building-a-grpc-server)

2. Implementing a client

3. gRPC Streaming

4. Advanced features like interceptors, etc.

5. Serving frontends using gRPC gateway

What are gRPC interceptors?

Those of us who have built Web applications will be familiar with the concept of middleware. Middleware allows developers to add custom functionality to the request/response processing pipeline. HTTP middleware intercepts incoming HTTP requests and outgoing HTTP responses and can perform various operations such as logging, authentication, caching, and more. Interceptors in gRPC can be thought of as middleware components that sit between the client and server, providing a way to inspect and modify gRPC messages as they pass through the system. They are essentially hooks that allow you to intercept and modify gRPC requests and responses as they pass through the system. Similar to regular HTTP middleware, Interceptors can be used to perform a wide range of tasks, such as logging, authentication, rate limiting, and error handling.

Implementing gRPC interceptors.

Let's start by implementing an interceptor to perform authentication on our server. The interceptor will check to see if incoming requests have an Authorization header with a value of 42 and fail the request if they don't.

In our server, let's create a folder called interceptors and within it, a file called authentication_interceptor.py

The first thing we need to do is create a terminator function. This is a function that will terminate the gRPC request if the header is not present or is invalid. Let's create the function.

import grpc

def _unary_unary_rpc_terminator(code, details):

    def terminate(ignored_request, context):
        context.abort(code, details)

    return grpc.unary_unary_rpc_method_handler(terminate)

When called, this function will cause the gRPC request to terminate with the error code and details passed in.

Now we can create the actual interceptor class.

class AuthenticationInterceptor:
    def __init__(self, header, value, code, details):
        self._header = header
        self._value = value
        self._terminator = _unary_unary_rpc_terminator(code, details)

    def intercept_service(self, continuation, handler_call_details):
        if (self._header,
                self._value) in handler_call_details.invocation_metadata:
            return continuation(handler_call_details)
        else:
             return self._terminator

Over here, we define the interceptor. It has an init method where we pass it the header name, the value and the error details for a failed request. The intercept_service method is where the magic happens. The first line, if (self._header,self._value) in handler_call_details.invocation_metadata: does the actual authentication. handler_call_details.invocation_metadata contains the data passed into the gRPC request. if the authentication is successful, we return a continuation, which is a fancy way of saying the execution of the program is allowed to continue. Otherwise, we terminate the execution of the program with the terminator.

Finally, we add the interceptor to our gRPC serve in server.py

    authenticator = AuthenticationInterceptor(
        'authorization',
        '42',
        grpc.StatusCode.UNAUTHENTICATED,
        'Access denied!'
    )
    server = grpc.server(
        futures.ThreadPoolExecutor(max_workers=10),
        interceptors=(authenticator,)
        )

If we now try and make a request with our client, we get the following error.

grpc._channel._InactiveRpcError: <_InactiveRpcError of RPC that terminated with:
        status = StatusCode.UNAUTHENTICATED
        details = "Access denied!"
        debug_error_string = "UNKNOWN:Error received from peer ipv4:127.0.0.1:50051 {created_time:"2023-03-30T17:05:51.7178673-07:00", grpc_status:16, grpc_message:"Access denied!"}"

Let's now add an interceptor to the client to add the relevant authentication data to the request.

Let's create a file called authentication interceptor in our client and add the following code

import collections

import grpc

class _ClientCallDetails(
        collections.namedtuple(
            '_ClientCallDetails',
            ('method', 'timeout', 'metadata', 'credentials')),
        grpc.ClientCallDetails):
    pass

class AuthenticationInterceptor(grpc.UnaryUnaryClientInterceptor):
    def __init__(self, interceptor_function):
        self._fn = interceptor_function

    def intercept_unary_unary(self, continuation, client_call_details, request):
        new_details, new_request_iterator, postprocess = self._fn(
            client_call_details, iter((request,)), False, False)
        response = continuation(new_details, next(new_request_iterator))
        return postprocess(response) if postprocess else response

def add_authentication(header, value):
    def intercept_call(client_call_details, request_iterator, request_streaming,
                       response_streaming):
        metadata = []
        if client_call_details.metadata is not None:
            metadata = list(client_call_details.metadata)
        metadata.append((
            header,
            value,
        ))
        client_call_details = _ClientCallDetails(
            client_call_details.method, client_call_details.timeout, metadata,
            client_call_details.credentials)
        return client_call_details, request_iterator, None

    return AuthenticationInterceptor(intercept_call)

We can now add the interceptor to our client in the following manner.

in client.py

...
from python_grpc_demo.client.interceptors.authentication_interceptor import add_authentication

def run():
    authentication = add_authentication('authorization', '42')
    name = input('What is your name?\n')

    with grpc.insecure_channel('localhost:50051') as channel:
        intercept_channel = grpc.intercept_channel(channel, authentication)  
...

Now that we've set up the interceptors correctly, the client and server should work correctly.

poetry run run-grpc-client
What is your name?
a
Hello a!

Thats it! We've seen how to get a simple gRPC interceptor running in Python.