Part 4 - gRPC Interceptors

Introduction

Up until now in the gRPC Series we have built:

A simple gRPC service
Added a REST/HTTP interface exposing gRPC service RESTfully and showing a glimpse of the gRPC plugin universe
Introduced Buf.build to simplify plugin management

We are far from productionalizing our service. A production ready service would (at the very least) need several things:

Authentication/Authorization
Request Logging
Request Tracing
Caching
Rate Limiting
Load balancing
and more (security, multi-zone/multi-regional services, etc)

A common thread across all these aspects is that these apply in an (almost) uniform way to all requests to an operation (without the operation being aware of it). eg given a request handler function (more on this later):

Request Logging would printing out common metrics (like response times, error traces etc) after calling the handler.
Rate limiting can also be applied in a uniform way (by looking up a config of request specific limits) and only invoking the handler if within those limits.
Authentication can look for common request headers (if HTTP) before allowing continuing onto the request handler.

In this post we will describe interceptors - a power gRPC facility for (well) intercepting and modifying requests and responses (and streams) on a gRPC server.

Middleware

Before going int gRPC interceptors, let us look at their parallel in the HTTP world - the Middleware! In a typical HTTP endpoint (api or otherwise) middleware is used extensively to wrap/decorate/filter requests. The role of middleware is to

Intercept a request a handler,
Reject, Modify or Forward the request as is to the underlying handler
Intercept (the forwarded) request's response, and then modify/forward back to the caller.

Request handlers are typically functions that (req: HTTPRequest) => HTTPResponse (in your favorite language/platform (™)). Naturally middleware can also be thought of as "decorator" functions that return other handler functions, eg:

function mymiddleware(anotherHandler: HTTPHandler): HTTPHandler {
  newHandler = function(req: HTTPRequest): HTTPResponse {
    req = // do some preprocessing and get a modified request

    resp = anotherHandler(req)

    resp = // do some post processing and get a modified response

    return resp
  }

  return newHandler     // Return the new handler function
}

So now we could create a very simple rate-limiter (say for a 5 minute window) with:

function rateLimitingMiddleware(originalHandler: HTTPHandler): HTTPHandler {
  return function (req: HTTPRequest): HTTPResponse {
    method = req.method
    path = req.path

    rate_config = getRateLimitConfig(method, path)

    // (5 minutes in seconds)
    ourWindow = 5 * 60
    num_requests = getNumRequestsInWindow(method, path, ourWindow)

    if (num_requests > rate_config.limit) {
      return HTTPResponse(429, 'Too many requests')
    }

    return originalHandler(req)
  }
}

The advantage of Middleware is that they can be chained to apply separate concerns without the knowledge of the main request handling the business logic. A common pattern (say in a language like Python that supports decorators) thus would look like:

@ratelimiter_middleware
@authenticator_middleware
@logger_middleware
def main_handler(req: HTTPRequest) -> HTTPResponse:
    return 200, "Hello World"

Without any syntactical decorator support this could be achieved with:


func createHttpserver() {
  ...
  ...

  widgetHandler := func(w http.ResponseWriter, r *http.Request) {
    // return a widget listing
  }

  mux := http.NewServeMux()
  mux.Handle("/api/widgets/",
                  authMiddleware(
                      loggerMiddleware(
                          rateLimitingMiddlewarea(
                              widgetHandler))))

	http.ListenAndServe("localhost:8080", mux)
  ...
  ...
}

There are fancier things one can do like apply middleware en-masse to an entire collection of routes. Such framework specific aesthetics are outside the scope of this post. If you are interested to learn more check out the amazing Gin Web Framework!

Interceptors

Now that we have seen their HTTP equivalent, interceptors (in gRPC) are very intuitive. Interceptors are also a way to decorate requests and responses in gRPC. However they come in two standardized specialized flavors:

Unary interceptors: These are "oneshot" interceptors. They either intercept a request or a response - once in the request/response's lifecycle.
Stream interceptors: These are "continuous". ie They intercept every message in a streaming request (client -> server) or a streaming response (server -> client).

Since interceptors can apply to both the client and server, we have four total favors:

Client Unary Interceptor - For intercepting a request just as it leaves the client but before it is sent to the server. A typical use case for these could be for a client that may look up a local (on-client) cache for queries instead of forwarding to a server. Another example is for client side routing - where the client may decide which server-shard to forward a request to based on the entity's ID. Other cases could be to log/monitor client-side latencies of requests, etc.
Server Unary Interceptor - These intercept a request that is received by a server (but before forwarding to the request handler). Server side interceptors are great for common validation of auth tokens or logging/monitoring server side latencies and errors and more.
Client Stream Interceptor - These - similar to their Unary counterpart - intercept and process/transform each message being streamed from the client to the server. A great use case for this could be an interceptor/agent that may collect multiple messages and collect them in a window before forwarding to the server (eg logs or metrics).
Server Stream Interceptor - Similar to their Unary counterpart - these intercept messages in a single connection when received at the server.

Interceptors provide more benefits than plain HTTP middleware:

HTTP middleware are very language/framework specific so each framework has their own conventions for creating/enforcing this.
HTTP middleware has no standard ways to decorate streams (eg Websocket packets). Since gRPC offers framing in streaming messages, stream interceptors can intercept individual messages in a stream. In HTTP (or Websockets) lack of a "typed message" stream means applications would have to implement their own framing of messages and decorator "schemas" to process these messages in arbitrary ways.

Implementing Interceptors

Our example does not (yet) have any streaming rpcs. So will add unary interceptors for now and add stream interceptors when we look at Websockets and Streaming.

First we will add a Client Unary Interceptor to our service clients (invoked by the gRPC gateway) to ensure that only requests that contain the auth header (with username+password) are forwarded to the server. Otherwise the call to the server is not even made (and a 403 is returned).

Then we will add a Server Unary Interceptor to our service to accept and validate these credentials (after all the server cannot just accept what ever the client sends at face value).

Support basic http auth in the gRPC gateway so that caller of our API can pass in a username/password to authenticate a user.
The gRPC gateway (http server) extracts the username/password (from http headers) and forwards it to the service (via gRPC metadata - see below).
The Server Unary Interceptor validates this username/password - against a static list of users/passwords. 3a. If the credentials are invalid then the interceptor returns an error to the gRPC gateway (without invoking the gRPC handler). 3b. If the credentials are valid the underlying service's handler is invoked

Clearly this auth scheme is very simplistic and we will look at more full-fledged and complex example in the article on Authentication.

Now let us look at the implementation of each of these.

Step 1 - Extract Username/Password from HTTP Request Headers

In Part 2, our startGatewayServer method simply starts a http server forwarding requests to the underlying gRPC service. Here we also introduced the NewServeMux method in the grpc-gateway/v2/runtime module as a better replacement over the standard library's NewServeMux method due its close understanding of the gRPC environment.

The first step thus for us is to extract the Auth related http headers from the incoming http request and add it to the Metadata that will be sent to the gRPC service. You can think of the Metadata as the headers equivalent in the gRPC environment. These are simply key/value pairs.

This is done below (in cmd/main.go):


import (
  ...
  ...
  // Add Imports
  "strings"
  "google.golang.org/grpc/codes"
  "google.golang.org/grpc/metadata"
  "google.golang.org/grpc/status"
  ...
  ...
)

...
...

func startGatewayServer(grpc_addr string, gw_addr string) {

    ctx := context.Background()

    // 
    // Step 1 - Add extra options to NewServeMux
    //
    mux := runtime.NewServeMux(
      runtime.WithMetadata(func (ctx context.Context, request *http.Request) metadata.MD {

        //
        // Step 2 - Extend the context
        //
        ctx = metadata.AppendToOutgoingContext(ctx)

        //
        // Step 3 - get the basic auth params
        //
        if username, password, ok := request.BasicAuth(); ok {
          md := metadata.Pairs()
          md.Append("OneHubUsername", username)
          md.Append("OneHubPassword", password)
          return md
        } else {
          return nil
        }
      }))

    opts := []grpc.DialOption{grpc.WithInsecure()}
    ...
    ...
    ...
}

The aditions are pretty minimal:

We modify the NewServeMux to include our first ServeMuxOption function (middleware)
This ServeMuxOption function extracts username/password basic auth params from the headers.
If the basic auth params are found they are wrapped as 2 metadata pairs and returned (to be passed to the service).

Step 2 - Ensure Auth Params in Client originating Metadata

Here is our first Client Unary Interceptor that before even forwarding a request to the gRPC service will ensure that the OneHubUsername and OneHubPassword metadata pairs are set. Why even send an unauthenticated request to the service to begin with!

Going back to our startGatewayServer method - once we are past the ServeMux, it is time to configure our DialOptions. gprc.DialOption simply configures how a connection is to be made to the service. In our example so far we just specified that we would like to configure our connection over an Insecure transport (in a secure environment the clients would also be issued certificates etc for authentication).

A client interceptor can be added as an additional DialOption! That is it. A unary client interceptor is just a function with the following signature:

type UnaryClientInterceptor func(ctx context.Context,
      method string,        // Method to be invoked on the service (eg GetTopics)
      req,                  // Request payload  (eg GetTopicsRequest)
      reply interface{},    // Response payload (eg GetTopicsResponse)
      cc *ClientConn,       // the underlying connection to the service
      invoker UnaryInvoker, // The next handler
      opts ...CallOption) error

The signature is hopefully self explanatory. The key parameter is the invoker which is the "next" handler that must be called by the interceptor if the chain is to be continued. The interceptor can choose to not call the invoker and instead return an error or a custom response (or error).

Our client interceptor is simple. It will call the invoker if username/password are present otherwise will throw an error:

func EnsureAuthExists(ctx context.Context,
	method string, // Method to be invoked on the service (eg GetTopics)
	req, // Request payload  (eg GetTopicsRequest)
	reply interface{}, // Response payload (eg GetTopicsResponse)
	cc *grpc.ClientConn, // the underlying connection to the service
	invoker grpc.UnaryInvoker, // The next handler
	opts ...grpc.CallOption) error {

	md, ok := metadata.FromOutgoingContext(ctx)
	if ok {
		usernames := md.Get("OneHubUsername")
		passwords := md.Get("OneHubPassword")
		if len(usernames) > 0 && len(passwords) > 0 {
			username := strings.TrimSpace(usernames[0])
			password := strings.TrimSpace(passwords[0])
			if len(username) > 0 && len(password) > 0 {
				// All fine - just call the invoker
				return invoker(ctx, method, req, reply, cc, opts...)
			}
		}
	}
	return status.Error(codes.NotFound, "BasicAuth params not found")
}

Note that metadata entries are really key/value-list pairs (much like headers or query-params in http). Now all that is left is to add our Interceptor to our DialOptions in the client:

func startGatewayServer(grpc_addr string, gw_addr string) {
    mux := ....

    opts := []grpc.DialOption{
        grpc.WithInsecure(),
        // Add our interceptor as a DialOption
        grpc.WithUnaryInterceptor(EnsureAuthExists),
    }
    ...
    ...
    ...
}

grpc.WithUnaryInterceptor takes a Unary Client Interceptor function and turns it into a DialOption. That's it!

Now start the server again (go cmd/server.go) let us test calls to our chat service and see how this works.

First let us try an unauthenticated call:

$ curl localhost:8080/v1/topics

{"code":5,"message":"BasicAuth params not found","details":[]}

As expected the call without Basic Auth headers was intercepted and rejected.

Now let us try with a username/password:

$ curl localhost:8080/v1/topics -u login:password

{"topics":[], "nextPageKey":""}

And lo behond - our request from the client was served by the server - though the request was not authenticated by the server.

One thing to observe in the above examples is how the Metadata object is created.

It is created from the Context
Specifically it is created from the "outgoing" context. There are 2 contexts associated - the incoming and outgoing context for responses and requests respectively.
The meaning of incoming and outgoing are reversed on the server side as the request is incoming and response is outgoing.

Step 3 - Add server side authentication

While it is commendable that the client ensured the presence of BasicAuth credentials, it is upto the server to validate them.

To do this we will add (as you guessed) a UnaryServerInterceptor which is a function with the signature:

type UnaryServerInterceptor func(
        ctx context.Context,
        req interface{},
        info *UnaryServerInfo,
        handler UnaryHandler
) (resp interface{}, err error)

This looks very similar to a UnaryClientInterceptor. The important parameters here are:

info - which contains RPC related information the interceptor can use and operate on.
handler - A wrapper over the service method implementation that is to be called by the interceptor (if the chain is to be continued).

For our server side auth - we shall add a basic interceptor:

func EnsureAuthIsValid(ctx context.Context,
	req interface{},
	info *grpc.UnaryServerInfo,
	handler grpc.UnaryHandler) (resp interface{}, err error) {
	md, ok := metadata.FromIncomingContext(ctx)
	if ok {
		usernames := md.Get("OneHubUsername")
		passwords := md.Get("OneHubPassword")
		if len(usernames) > 0 && len(passwords) > 0 {
			username := strings.TrimSpace(usernames[0])
			password := strings.TrimSpace(passwords[0])

			// Make sure you use better passwords than this!
			if len(username) > 0 && password == fmt.Sprintf("%s123", username) {
				// All fine - just call the invoker
				return handler(ctx, req)
			}
		}
	}
	return nil, status.Error(codes.NotFound, "Invalid username/password")
}

This is very similar to our Client interceptor.

Get the metadata from the Incoming context (recall that on the client side this was from the Outgoing context).
Ensure password == username + 123 (needless to say we could do better here)
If passwords match continue on otherwise return an error

We have one final step left - activating it. This is very similar to activating our client interceptor. The client interceptor was activated by passing our interceptor as a DialOption. The server interceptor will be passed a ServerOption to the NewServer method in the startGRPCEndpoints function:

func startGRPCServer(addr string) {
	  // create new gRPC server
	  server := grpc.NewServer(
		    grpc.UnaryInterceptor(EnsureAuthIsValid),
	  )
    ...
    ...

Let us test it again now. Passing the last login:password combo while valid from the client should now get rejected by the server (note the different error messages):

$ curl localhost:8080/v1/topics -u login:password

{"code":5, "message":"Invalid username/password", "details":[]}

Passing the right password fixes this:

$ curl localhost:8080/v1/topics -u login:login123

{"topics":[], "nextPageKey":""}

Step 4 - Use metadata

Up until now our service methods have been shielded so that they wont even be called if an auth param was not passed or was invalid (albeit with a simple check for a "123" prefix). Sometimes it is necessary for the service methods to obtain and use this information. For example when an entity is created the service may want to enforce that the "creator" is set to the logged-in/authenticated user instead of an arbitrary value passed by the caller.

This is quite simple. Let us take the CreateTopic method:

func (s *TopicService) CreateTopic(ctx context.Context, req *protos.CreateTopicRequest) (resp *protos.CreateTopicResponse, err error) {
	resp = &protos.CreateTopicResponse{}
	resp.Topic = s.EntityStore.Create(req.Topic)
	return
}

It can now use the auth info passed in via the interceptors:

func (s *TopicService) CreateTopic(ctx context.Context, req *protos.CreateTopicRequest) (resp *protos.CreateTopicResponse, err error) {
	resp = &protos.CreateTopicResponse{}
  req.Topic.CreatorId = GetAuthedUser(ctx)
  if req.Topic.CreatorId == "" {
    return nil, status.Error(codes.PermissionDenied, "User is not authenticated to create a topic")
  }
	resp.Topic = s.EntityStore.Create(req.Topic)
	return
}

If we try to create a topic any custom will be overwritten by the ID of the logged in user:

curl -X POST localhost:8080/v1/topics  \
     -u auser:auser123       \
     -H 'Content-Type: application/json' \
     -d '{"topic": {"name": "First Topic", "creator_id": "user1"}}' | json_pp

yielding:

{
   "topic" : {
      "createdAt" : "2023-08-04T08:52:52.861406Z",
      "creatorId" : "auser",
      "id" : "1",
      "name" : "First Topic",
      "updatedAt" : "2023-08-04T08:52:52.861407Z",
      "users" : []
   }
}

That's it. That's all there is to interceptors. Stream interceptors are very similar but we wont cover them here just yet. Wait for it though!

Conclusion

By using interceptors a service can be wrapped/decorated a lot of common/cross-cutting capabilities in a way transparent to the underlying service (and method handlers). This clean allows separation of concerns as well the ability to plug/play/replace these common behaviors with other providers in the future. Some of the interesting things that can be done with interceptors is to enable logging, request tracing, authentication, rate-limiting, load balancing and much more.

To summarize, In this article:

We contrasted HTTP middleware and gRPC interceptors
Touched upon the versatility of interceptors in providing a wide variety of functionality
Implemented Unary interceptors to decorate requests both on the client as well as the server side to provide a simple authentication mechanism.

In the next post we will finally start persisting our data in a real database. We will also containerize our whole setup for easy development, portability and startup. This will also pave the way for keeping development/startup simple as we add more services for different extensions on our canonical Chat service!