Modern Microservices, Part 4: Decorator Pattern With gRPC Interceptors

So far in our series on modern microservices, 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 (service) requests (without the operation being aware of it); e.g., given a request handler function (more on this later):

• Request logging would be 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.

If you want to jump right into the code, you can find it here.

Middleware

Before going into 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 from 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, e.g.:

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 the following:

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 it can be chained to apply separate concerns without the knowledge of the main request handling the business logic. Thus, a common pattern (say, in a language like Python that supports decorators) 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 "one-shot" interceptors.  They either intercept a request or a response, once in the request/response's lifecycle.
• Stream interceptors: These are "continuous." 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:

1. Client Unary Interceptor - Client Unary Interceptors are 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.
2. 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.
3. Client Stream Interceptor - Similar to their Unary counterpart, these 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 them to the server (e.g., logs or metrics).
4. 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:

1. HTTP middleware is very language/framework specific so each framework has its own conventions for creating/enforcing this.
2. HTTP middleware has no standard ways to decorate streams (e.g., WebSocket packets). Since gRPC offers framing in streaming messages, stream interceptors can intercept individual messages in a stream. In HTTP (or WebSockets), the 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. We will only add unary interceptors for now, and add stream interceptors when we look at a future post on 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 whatever the client sends at face value):

1. Support basic HTTP auth in the gRPC Gateway so that the caller of our API can pass in a username/password to authenticate a user.
2. The gRPC Gateway (HTTP server) extracts the username/password (from HTTP headers) and forwards it to the service (via gRPC metadata - see below).
3. The Server Unary Interceptor validates this username/password against a static list of users/passwords. 
   1. If the credentials are invalid, then the interceptor returns an error to the gRPC gateway (without invoking the gRPC handler). 
   2. 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 examples in a future post on authentication.

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

Step 1: Extract Username/Password From HTTP Request Headers

Our startGatewayServer method simply starts an 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 for the standard library's NewServeMux method due to its close understanding of the gRPC environment.

Thus, the first step for us is to extract the auth-related HTTP headers from the incoming HTTP request and add them 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 additions are pretty minimal:

1. We modify NewServeMux to include our first ServeMuxOption function (middleware).
2. This ServeMuxOption function extracts username/password basic auth params from the headers.
3. 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, which, before 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 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 a username/password are present; otherwise, it 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) and 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":""}

Lo and behold: 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 meanings of incoming and outgoing are reversed on the server side as the request is incoming and the response is outgoing.

Step 3: Add Server-Side Authentication

While it is commendable that the client ensured the presence of BasicAuth credentials, it is up to 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 - 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 won't 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 won't cover them here just yet.  Wait for it, though!

Conclusion

By using interceptors a service can be wrapped/decorated with a lot of common/cross-cutting capabilities in a way transparent to the underlying service (and method handlers). This 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 are 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 and environment for easy development, portability, and packaging.  This will also pave the way for keeping development/startup simple as we add more services for different extensions on our canonical chat service!