backend / protocols / grpc / 06_interceptors_metadata_auth.md

gRPC Interceptors, Metadata, and Auth

6 interview angles 6 min read source

gRPC Interceptors, Metadata, and Auth

Interceptors are gRPC’s middleware. Metadata is the headers-equivalent — key-value pairs that ride along with every RPC. Together they’re how you do auth, logging, tracing, retry, rate limiting.

Metadata

Key-value pairs attached to RPCs. Like HTTP headers, but the gRPC name.

# Client — send metadata
metadata = [
    ("authorization", "Bearer eyJ..."),
    ("x-request-id", "abc-123"),
    ("x-tenant-id", "acme"),
]
response = stub.GetUser(request, metadata=metadata)

# Server — read metadata
def GetUser(self, request, context):
    md = dict(context.invocation_metadata())
    auth = md.get("authorization", "")
    request_id = md.get("x-request-id", "")
    ...

Rules:

  • Keys are lowercase, ASCII.
  • Values are strings, OR bytes for keys ending in -bin (binary metadata).
  • : prefix is reserved (used internally for HTTP/2 pseudo-headers like :method).
  • Don’t put PII unencrypted in metadata — gets logged by intermediaries.

Auth via metadata

The standard pattern:

# Client puts auth in metadata
def get_auth_metadata():
    token = get_current_token()
    return [("authorization", f"Bearer {token}")]

response = stub.GetUser(request, metadata=get_auth_metadata())

Server validates in an interceptor (see below) or in each method.

Auth via channel credentials

For automatic per-call auth (token attached to every RPC):

# Custom call credentials
class TokenCredentials(grpc.AuthMetadataPlugin):
    def __init__(self, token):
        self._token = token
    def __call__(self, context, callback):
        callback([("authorization", f"Bearer {self._token}")], None)

call_creds = grpc.metadata_call_credentials(TokenCredentials("eyJ..."))
ssl_creds = grpc.ssl_channel_credentials()
composite = grpc.composite_channel_credentials(ssl_creds, call_creds)

channel = grpc.secure_channel("server:50051", composite)

Now every RPC on this channel carries the token. Useful when the token doesn’t change per request.

For token refresh: AuthMetadataPlugin.__call__ can fetch a fresh token before each RPC.

Server-side interceptor

The middleware equivalent. Runs before each RPC.

import grpc

class AuthInterceptor(grpc.ServerInterceptor):
    def intercept_service(self, continuation, handler_call_details):
        # handler_call_details.invocation_metadata is the metadata
        md = dict(handler_call_details.invocation_metadata)
        auth = md.get("authorization", "")

        if not auth.startswith("Bearer "):
            return self._unary_unary_terminator(grpc.StatusCode.UNAUTHENTICATED, "Missing token")

        try:
            user = validate_token(auth[7:])
        except InvalidTokenError:
            return self._unary_unary_terminator(grpc.StatusCode.UNAUTHENTICATED, "Invalid token")

        # store on context (custom propagation)
        # gRPC doesn't have a clean per-RPC context object;
        # use ContextVar or pass via header re-reading in the handler
        return continuation(handler_call_details)

    def _unary_unary_terminator(self, code, message):
        def terminate(request, context):
            context.abort(code, message)
        return grpc.unary_unary_rpc_method_handler(terminate)

server = grpc.server(executor, interceptors=[AuthInterceptor()])

Interceptors are chained — multiple in the list, applied in order. Common ones: auth, logging, tracing, rate limiting.

Async interceptor

class AsyncAuthInterceptor(grpc.aio.ServerInterceptor):
    async def intercept_service(self, continuation, handler_call_details):
        md = dict(handler_call_details.invocation_metadata)
        if not validate(md.get("authorization", "")):
            ...
        return await continuation(handler_call_details)

server = grpc.aio.server(interceptors=[AsyncAuthInterceptor()])

Same API, async flavor.

Logging interceptor — the always-useful one

import time
import logging

class LoggingInterceptor(grpc.ServerInterceptor):
    def intercept_service(self, continuation, handler_call_details):
        method = handler_call_details.method
        start = time.perf_counter()

        handler = continuation(handler_call_details)

        def wrapped_handler(request, context):
            try:
                response = handler.unary_unary(request, context)
                elapsed = time.perf_counter() - start
                logging.info(f"{method} OK in {elapsed*1000:.1f}ms")
                return response
            except Exception as e:
                elapsed = time.perf_counter() - start
                logging.error(f"{method} FAILED in {elapsed*1000:.1f}ms: {e}")
                raise

        return grpc.unary_unary_rpc_method_handler(wrapped_handler)

Production servers should always have this (or OpenTelemetry, which generates spans).

Client-side interceptor

For adding metadata uniformly, logging client RPCs, retries:

class ClientLoggingInterceptor(grpc.UnaryUnaryClientInterceptor):
    def intercept_unary_unary(self, continuation, client_call_details, request):
        start = time.perf_counter()
        response = continuation(client_call_details, request)
        elapsed = time.perf_counter() - start
        logging.info(f"called {client_call_details.method} in {elapsed*1000:.1f}ms")
        return response

channel = grpc.intercept_channel(channel, ClientLoggingInterceptor())

For each RPC type (unary-unary, unary-stream, stream-unary, stream-stream), implement the corresponding interceptor interface.

Per-method auth — when interceptors are too coarse

def GetUser(self, request, context):
    md = dict(context.invocation_metadata())
    user = validate_token(md.get("authorization", "").removeprefix("Bearer "))
    if user.id != request.user_id and not user.is_admin:
        context.abort(grpc.StatusCode.PERMISSION_DENIED, "Cannot view other users")
    return ...

Resource-level auth (this user can see this specific resource) usually happens in the method, not the interceptor.

Context propagation — request ID, tracing

Distributed tracing typically rides on metadata:

traceparent: 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01

W3C Trace Context standard. OpenTelemetry-instrumented gRPC clients/servers handle this automatically.

For app-level context (current user, tenant) across function calls within a request, Python’s contextvars:

from contextvars import ContextVar

current_user: ContextVar[User | None] = ContextVar("current_user", default=None)

class AuthInterceptor(grpc.ServerInterceptor):
    def intercept_service(self, continuation, handler_call_details):
        token = get_token(handler_call_details.invocation_metadata)
        user = validate_token(token)
        current_user.set(user)         # context-local; visible to handler
        return continuation(handler_call_details)

def GetUser(self, request, context):
    user = current_user.get()
    if user is None:
        context.abort(grpc.StatusCode.UNAUTHENTICATED)
    ...

ContextVars work with asyncio and threadpool servers.

OpenTelemetry instrumentation

pip install opentelemetry-instrumentation-grpc
from opentelemetry.instrumentation.grpc import GrpcInstrumentorServer, GrpcInstrumentorClient

GrpcInstrumentorServer().instrument()
GrpcInstrumentorClient().instrument()

Automatic spans for every RPC (in and out), with propagated trace IDs. Plug into Jaeger / Zipkin / DataDog / etc.

Rate limiting interceptor

class RateLimitInterceptor(grpc.ServerInterceptor):
    def __init__(self):
        self.bucket = TokenBucket(rate=100, capacity=200)

    def intercept_service(self, continuation, handler_call_details):
        if not self.bucket.consume(1):
            def reject(req, ctx):
                ctx.abort(grpc.StatusCode.RESOURCE_EXHAUSTED, "Rate limit exceeded")
            return grpc.unary_unary_rpc_method_handler(reject)
        return continuation(handler_call_details)

For per-client rate limiting, key the bucket by client identity from metadata.

mTLS for service-to-service auth

The standard pattern in service meshes (Istio, Linkerd) — every service has a cert, both sides authenticate. See 05_python_grpc.md for setup.

The cert identifies the calling service; you read it via:

def GetUser(self, request, context):
    auth_context = context.auth_context()
    peer_cn = auth_context.get("x509_common_name", [b""])[0].decode()
    # peer_cn = "users-service.acme.svc.cluster.local"

Combined with policy: “service A can call methods X, Y; service B can call X only.”

JWT as alternative to mTLS

Skip cert distribution; use signed tokens:

class JWTAuthInterceptor(grpc.ServerInterceptor):
    def intercept_service(self, continuation, handler_call_details):
        md = dict(handler_call_details.invocation_metadata)
        token = md.get("authorization", "").removeprefix("Bearer ")
        try:
            claims = jwt.decode(token, public_key, algorithms=["RS256"], audience="my-service")
        except jwt.InvalidTokenError:
            ...
        current_user.set(User(id=claims["sub"], role=claims["role"]))
        return continuation(handler_call_details)

Cheaper to operate (no cert rotation per service); less cryptographically strong than mTLS (token can be stolen and replayed).

Common pitfalls

  • Trusting metadata without validation — clients can send any metadata. Validate auth on the server.
  • Using mTLS but not checking the peer cert in app code — the TLS layer authenticates; the app must authorize.
  • Auth interceptor that returns the wrong status code — return UNAUTHENTICATED for missing/invalid creds, PERMISSION_DENIED for valid but disallowed.
  • Putting tokens in plaintext metadata over grpc.insecure_channel — anyone on the network can grab them. Use TLS.
  • Forgetting to install for the right RPC pattern — UnaryUnaryClientInterceptor doesn’t cover streaming. Implement all four flavors if you have all four.

Common interview confusions

  • “Interceptors run on the client only / server only.” — both have their own. Symmetric concept.
  • “Metadata is encrypted in TLS.” — yes in transit; not at the application layer. Don’t log it.
  • “UNAUTHENTICATED and PERMISSION_DENIED are the same.” — UNAUTHENTICATED = no/invalid credentials. PERMISSION_DENIED = authenticated but not allowed. Same as HTTP 401 vs 403.

Interview angle

  • “What’s gRPC metadata?” — key-value pairs attached to each RPC, equivalent to HTTP headers. Used for auth tokens, request IDs, trace context, tenant IDs. Set on the client, read on the server.
  • “How would you implement auth in a gRPC service?” — client sends Bearer token in metadata; server-side interceptor validates the token and attaches user info to a ContextVar (or per-RPC context). Resource-level checks happen in the method handler.
  • “What’s a gRPC interceptor?” — middleware: a function called before/around each RPC. Common uses: auth, logging, tracing, rate limiting. Server-side and client-side both exist.
  • “mTLS vs JWT for service-to-service auth?” — mTLS authenticates the calling service via its cert (strong, infrastructure-level). JWT is application-level signed tokens — simpler to manage but token theft is a risk. Service meshes do mTLS automatically; greenfield services often use JWT.
  • “How do you propagate request-scoped state in Python gRPC?” — contextvars.ContextVar. Set in an interceptor; readable across the call chain (works with asyncio and threadpool).
  • “How does OpenTelemetry integrate with gRPC?” — opentelemetry-instrumentation-grpc automatically adds spans on every RPC, propagates trace IDs via metadata (traceparent header per W3C Trace Context).
Previous
Python gRPC