Securing gRPC with JWTs

Welcome back! In the last lesson, you built a crucial piece of our security infrastructure: a JWT-based authentication layer in the FastAPI API Gateway. This secures the "front door" to our system, ensuring that only authenticated users can make requests. However, our internal service-to-service communication remains unprotected. Any service on our private network can call another without restriction.

This lesson tackles that vulnerability. We will extend our security model to the gRPC layer, implementing a zero-trust architecture where every service call is authenticated. You will learn how to propagate the JWT received by the API Gateway to downstream gRPC services and how those services can validate the token to authorize the request. This is a fundamental pattern for building secure, production-grade microservices.

Our main tool for this will be gRPC Interceptors, which function like middleware in a web framework, allowing us to inspect and modify RPCs before they are executed.

1. The Zero-Trust Principle in Microservices

In a monolithic application, once a user is authenticated, components generally trust each other. In a distributed system, this assumption is dangerous. A "zero-trust" security model dictates that we should never trust a request just because it originates from within our network. Every service must validate the caller's identity and permissions for every single request.

This is how we'll adapt our architecture:

Client (End-User) -> API Gateway: The user logs in and gets a JWT. They include this JWT in the Authorization header of their request to the Gateway.
API Gateway (as a gRPC Client): The Gateway validates the JWT. Instead of discarding it, it propagates the token by attaching it to the metadata of its outgoing gRPC call to a downstream service (e.g., the User Service).
Downstream gRPC Service (Server): The gRPC service receives the call, extracts the JWT from the call's metadata, and re-validates it to ensure the request is legitimate before processing the RPC.

This flow ensures that the original user's identity is securely carried throughout the call chain.

Microservices Security Architecture with JWT and mTLS — This diagram shows the security flow. A user authenticates via the API Gateway, which obtains a JWT. The Gateway then propagates this JWT to internal services like the Order Service, which in turn might use mTLS for its own calls. Our focus is on the JWT propagation from the Gateway to the first gRPC service.

2. Introducing gRPC Interceptors

To implement this flow elegantly, we use gRPC Interceptors. An interceptor is a function that "intercepts" an RPC on either the client or server side, allowing you to run logic before the call is sent or before the handler is executed. They are perfect for cross-cutting concerns like:

Authentication & Authorization (our focus)
Logging
Metrics & Tracing
Request/Response Transformation

gRPC Interceptors for JWT Authentication Flow — This diagram illustrates how client and server interceptors sit in the request path for both unary and streaming calls, providing hooks for logic like JWT validation.

We will implement two interceptors: one on the server to validate tokens and one on the client to attach them.

[gRPC #15] Implement gRPC interceptor + JWT authentication in Go

To get a clear, high-level understanding of interceptors, watch this introductory segment from the "Implement gRPC interceptor" video by TECH SCHOOL. Although the code is in Go, the concepts are identical for Python.

Watch the first minute on what gRPC interceptors are. It clearly explains the role of both client-side and server-side interceptors and their common use cases.

3. Server-Side: The Validator Interceptor

Let's start with the gRPC service that needs to be protected. We'll create a server-side interceptor that checks for a valid JWT in the incoming call's metadata. If the token is missing or invalid, the interceptor will reject the call immediately.

The grpcio library provides the grpc.aio.ServerInterceptor base class. We need to inherit from it and implement the intercept_service method.

The logic within intercept_service is as follows:

Check if the RPC method being called is public. If so, let it pass through.
If it's a protected method, access the call's metadata via handler_call_details.invocation_metadata.
Extract the authorization header value. It should be in the format Bearer <token>.
Use our JWTManager (from the previous lesson) to validate the token.
If validation succeeds, proceed with the RPC by calling await continuation(handler_call_details).
If validation fails, abort the RPC with a grpc.StatusCode.UNAUTHENTICATED status.

The article "How to Implement JWT Authentication in gRPC" provides an excellent Python implementation of this pattern.

How to Implement JWT Authentication in gRPC

This resource from OneUptime provides a complete Python implementation for JWT authentication in gRPC, including both server and client components.

Focus on the Python Implementation section. Study the JWTManager and AuthInterceptor classes. Notice how the intercept_service method retrieves the token from the invocation_metadata and uses the JWT manager to validate it. Don't worry about the JWTCallCredentials class yet; we'll cover that next.

The grpc.aio documentation officially defines the ServerInterceptor class and the context object used to get metadata.

grpc.aio.ServerInterceptor: The base class for our validator. You can review its definition in the official gRPC Python Docs (LINK, section 5).
servicer_context.invocation_metadata(): This method, available inside an RPC handler, is how you access incoming headers. Our interceptor gets this information from the handler_call_details object. The definition is in the docs here (LINK, section 3).

Once the interceptor is defined, you simply pass an instance of it when creating your gRPC server:




# In your gRPC server file
auth_interceptor = AuthInterceptor(jwt_manager, public_methods=['/my.service.MyService/PublicMethod'])
server = grpc.aio.server(interceptors=[auth_interceptor])

4. Client-Side: Propagating the JWT

Now for the client side—our FastAPI Gateway. When it makes a gRPC call, it must attach the user's JWT to the request metadata. The standard key for this is authorization, with the value Bearer <token>.

While you can pass metadata manually on every call, a much cleaner and more reusable approach is to use a Call Credentials Plugin. This is a class that automatically attaches metadata for you. The resource you just read, "How to Implement JWT Authentication in gRPC", demonstrates this with the JWTCallCredentials class.

This class inherits from grpc.AuthMetadataPlugin and implements a __call__ method. This method receives a callback that it must invoke with the metadata to be attached.

import grpc

class JWTCallCredentials(grpc.AuthMetadataPlugin):
    """Client-side JWT credentials plugin."""
    def __init__(self, token: str):
        self.token = token

    def __call__(self, context, callback):



        # Attach the authorization header to the gRPC metadata
        metadata = (('authorization', f'Bearer {self.token}'),)
        callback(metadata, None)

To use it, you create an instance of this class with the user's token and combine it with the channel credentials. If you're using an insecure channel for development, it looks like this:




# In your FastAPI endpoint or gRPC client code
token = "the-jwt-from-the-http-request"




# Create call credentials from our plugin
call_credentials = grpc.metadata_call_credentials(JWTCallCredentials(token))




# For an insecure channel, you can use them directly
channel = grpc.aio.insecure_channel(
    'localhost:50051',
    credentials=call_credentials
)



# For a secure channel, you combine them with channel credentials (e.g., SSL)
# channel_creds = grpc.ssl_channel_credentials()
# composite_creds = grpc.composite_channel_credentials(channel_creds, call_credentials)
# channel = grpc.aio.secure_channel('localhost:50051', composite_creds)

stub = YourServiceStub(channel)
await stub.ProtectedMethod(request) # The token is now attached automatically

This pattern is extremely powerful because it decouples the authentication mechanism from your application logic. Your service code doesn't need to know how the token is attached; it just uses an authenticated channel.

5. Hands-on Exercise: Secure a gRPC Service

Let's put this all together. You will be provided with two files:

user_service.py: A simple gRPC service with a public and a protected method.
gateway.py: A FastAPI application that acts as a client to the user_service.

Your mission is to secure the communication between them.

Instructions:

Setup: You'll need grpcio, grpcio-tools, fastapi, uvicorn, and pyjwt. Generate the gRPC stubs by running python -m grpc_tools.protoc -I. --python_out=. --pyi_out=. --grpc_python_out=. user.proto.
Part 1 (Server): In user_service.py, implement the AuthInterceptor class. Its intercept_service method should validate the JWT for the GetSecretData RPC. Use the provided JWTManager. Add the interceptor to the grpc.aio.server.
Part 2 (Client): In gateway.py, implement the JWTCallCredentials class. In the /secret-data endpoint, extract the token, create the call_credentials, and use them when creating the channel to the user service.
Test:
- Run user_service.py.
- Run gateway.py using uvicorn.
- Go to http://127.0.0.1:8000/docs.
- First, try calling /secret-data without a token. It should fail.
- Call /token to get a JWT.
- Use the "Authorize" button in Swagger UI to set the Bearer token.
- Now call /secret-data again. It should succeed and return the secret message.

Files for the exercise:

user.proto

syntax = "proto3";

package user;

service UserService {
    rpc GetPublicData (Empty) returns (DataResponse);
    rpc GetSecretData (Empty) returns (DataResponse);
}

message Empty {}

message DataResponse {
    string message = 1;
}

user_service.py (Your workspace)

import asyncio
import grpc
import jwt
from concurrent import futures
from typing import Optional, Dict




# Assume user_pb2 and user_pb2_grpc are generated
import user_pb2
import user_pb2_grpc




# --- JWT Manager (from previous lesson) ---
class JWTManager:
    def __init__(self, secret_key: str, algorithm: str = 'HS256'):
        self.secret_key = secret_key
        self.algorithm = algorithm

    def validate_token(self, token: str) -> Optional[Dict]:
        try:
            payload = jwt.decode(token, self.secret_key, algorithms=[self.algorithm])
            return payload
        except (jwt.ExpiredSignatureError, jwt.InvalidTokenError):
            return None

jwt_manager = JWTManager(secret_key="your-super-secret-key")




# TODO 1: Implement the server-side authentication interceptor
class AuthInterceptor(grpc.aio.ServerInterceptor):
    def __init__(self, jwt_manager: JWTManager, public_methods: set):
        self._jwt_manager = jwt_manager
        self._public_methods = public_methods

    async def intercept_service(self, continuation, handler_call_details):
        method_name = handler_call_details.method
        if method_name in self._public_methods:
            return await continuation(handler_call_details)

        metadata = dict(handler_call_details.invocation_metadata)
        auth_header = metadata.get('authorization', '')

        if not auth_header.startswith('Bearer '):



            # Define a function to abort the call
            def abort_unauthenticated(request, context):
                context.abort(grpc.StatusCode.UNAUTHENTICATED, "Auth token is missing or invalid format")



            # Return a handler that immediately aborts
            return grpc.unary_unary_rpc_method_handler(abort_unauthenticated)

        token = auth_header[7:]
        payload = self._jwt_manager.validate_token(token)

        if payload is None:
            def abort_invalid(request, context):
                context.abort(grpc.StatusCode.UNAUTHENTICATED, "Token is invalid or expired")
            return grpc.unary_unary_rpc_method_handler(abort_invalid)
        



        # If token is valid, proceed
        # You could add the payload to the context here for downstream use
        # using contextvars if needed.
        return await continuation(handler_call_details)

class UserService(user_pb2_grpc.UserServiceServicer):
    async def GetPublicData(self, request, context):
        return user_pb2.DataResponse(message="This is public data. Anyone can see this.")

    async def GetSecretData(self, request, context):
        return user_pb2.DataResponse(message="This is secret data! You must be authenticated to see this.")

async def serve():
    public_methods = {'/user.UserService/GetPublicData'}



    # TODO 2: Instantiate your interceptor
    auth_interceptor = ...
    



    # TODO 3: Add the interceptor to the server
    server = grpc.aio.server(interceptors=...) # <-- Pass a list with your interceptor instance
    
    user_pb2_grpc.add_UserServiceServicer_to_server(UserService(), server)
    server.add_insecure_port('[::]:50051')
    await server.start()
    print("User service started on port 50051")
    await server.wait_for_termination()

if __name__ == '__main__':
    asyncio.run(serve())

gateway.py (Your workspace)

import grpc
import jwt
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from typing import Annotated

import user_pb2
import user_pb2_grpc




# --- Configuration ---
SECRET_KEY = "your-super-secret-key"
ALGORITHM = "HS256"
USER_SERVICE_ADDR = "localhost:50051"

app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")




# TODO 1: Implement the client-side JWT credentials plugin
class JWTCallCredentials(grpc.AuthMetadataPlugin):
    def __init__(self, token: str):
        self.token = token

    def __call__(self, context, callback):
        metadata = ... # Should be a tuple of key-value pairs
        callback(metadata, None)




# --- API Endpoints ---
@app.post("/token")
async def login():



    # In a real app, you'd verify username/password
    to_encode = {"sub": "testuser@example.com", "scope": "read write"}
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return {"access_token": encoded_jwt, "token_type": "bearer"}

@app.get("/public-data")
async def get_public_data():
    async with grpc.aio.insecure_channel(USER_SERVICE_ADDR) as channel:
        stub = user_pb2_grpc.UserServiceStub(channel)
        response = await stub.GetPublicData(user_pb2.Empty())
        return {"response": response.message}

@app.get("/secret-data")
async def get_secret_data(token: Annotated[str, Depends(oauth2_scheme)]):



    # TODO 2: Create an instance of your JWTCallCredentials
    call_credentials = ...




    # TODO 3: Create a channel with the call credentials
    # For an insecure channel, they are passed to the `credentials` argument.
    async with grpc.aio.insecure_channel(
        USER_SERVICE_ADDR,
        credentials=...
    ) as channel:
        stub = user_pb2_grpc.UserServiceStub(channel)
        try:
            response = await stub.GetSecretData(user_pb2.Empty())
            return {"response": response.message}
        except grpc.aio.AioRpcError as e:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=str(e.details()))

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Solution

user_service.py solution:

import asyncio
import grpc
import jwt
from concurrent import futures
from typing import Optional, Dict

import user_pb2
import user_pb2_grpc




# --- JWT Manager (from previous lesson) ---
class JWTManager:
    def __init__(self, secret_key: str, algorithm: str = 'HS256'):
        self.secret_key = secret_key
        self.algorithm = algorithm

    def validate_token(self, token: str) -> Optional[Dict]:
        try:
            payload = jwt.decode(token, self.secret_key, algorithms=[self.algorithm])
            return payload
        except (jwt.ExpiredSignatureError, jwt.InvalidTokenError):
            return None

jwt_manager = JWTManager(secret_key="your-super-secret-key")




# 1. Server-side authentication interceptor implemented
class AuthInterceptor(grpc.aio.ServerInterceptor):
    def __init__(self, jwt_manager: JWTManager, public_methods: set):
        self._jwt_manager = jwt_manager
        self._public_methods = public_methods

    async def intercept_service(self, continuation, handler_call_details):
        method_name = handler_call_details.method
        if method_name in self._public_methods:
            return await continuation(handler_call_details)

        metadata = dict(handler_call_details.invocation_metadata)
        auth_header = metadata.get('authorization', '')

        if not auth_header.startswith('Bearer '):
            def abort_unauthenticated(request, context):
                context.abort(grpc.StatusCode.UNAUTHENTICATED, "Auth token is missing or invalid format")
            return grpc.unary_unary_rpc_method_handler(abort_unauthenticated)

        token = auth_header[7:]
        payload = self._jwt_manager.validate_token(token)

        if payload is None:
            def abort_invalid(request, context):
                context.abort(grpc.StatusCode.UNAUTHENTICATED, "Token is invalid or expired")
            return grpc.unary_unary_rpc_method_handler(abort_invalid)
        
        return await continuation(handler_call_details)

class UserService(user_pb2_grpc.UserServiceServicer):
    async def GetPublicData(self, request, context):
        return user_pb2.DataResponse(message="This is public data. Anyone can see this.")

    async def GetSecretData(self, request, context):
        return user_pb2.DataResponse(message="This is secret data! You must be authenticated to see this.")

async def serve():
    public_methods = {'/user.UserService/GetPublicData'}



    # 2. Instantiate the interceptor
    auth_interceptor = AuthInterceptor(jwt_manager, public_methods)
    



    # 3. Add the interceptor to the server
    server = grpc.aio.server(interceptors=[auth_interceptor])
    
    user_pb2_grpc.add_UserServiceServicer_to_server(UserService(), server)
    server.add_insecure_port('[::]:50051')
    await server.start()
    print("User service started on port 50051")
    await server.wait_for_termination()

if __name__ == '__main__':
    asyncio.run(serve())

gateway.py solution:

import grpc
import jwt
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from typing import Annotated

import user_pb2
import user_pb2_grpc




# --- Configuration ---
SECRET_KEY = "your-super-secret-key"
ALGORITHM = "HS256"
USER_SERVICE_ADDR = "localhost:50051"

app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")




# 1. Client-side JWT credentials plugin implemented
class JWTCallCredentials(grpc.AuthMetadataPlugin):
    def __init__(self, token: str):
        self.token = token

    def __call__(self, context, callback):
        metadata = (('authorization', f'Bearer {self.token}'),)
        callback(metadata, None)




# --- API Endpoints ---
@app.post("/token")
async def login():
    to_encode = {"sub": "testuser@example.com", "scope": "read write"}
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return {"access_token": encoded_jwt, "token_type": "bearer"}

@app.get("/public-data")
async def get_public_data():
    async with grpc.aio.insecure_channel(USER_SERVICE_ADDR) as channel:
        stub = user_pb2_grpc.UserServiceStub(channel)
        response = await stub.GetPublicData(user_pb2.Empty())
        return {"response": response.message}

@app.get("/secret-data")
async def get_secret_data(token: Annotated[str, Depends(oauth2_scheme)]):



    # 2. Create an instance of JWTCallCredentials
    jwt_creds = JWTCallCredentials(token)
    call_credentials = grpc.metadata_call_credentials(jwt_creds)




    # 3. Create a channel with the call credentials
    async with grpc.aio.insecure_channel(
        USER_SERVICE_ADDR,
        credentials=call_credentials
    ) as channel:
        stub = user_pb2_grpc.UserServiceStub(channel)
        try:
            response = await stub.GetSecretData(user_pb2.Empty())
            return {"response": response.message}
        except grpc.aio.AioRpcError as e:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=str(e.details()))

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Conclusion

Excellent work. You have now implemented end-to-end authentication in our microservices system. By propagating and re-validating JWTs at each service boundary, you've established a zero-trust architecture that is robust and secure.

Key Takeaways:

Zero-Trust: Never trust internal network calls by default. Every service is responsible for its own authorization.
gRPC Interceptors: They are the standard, non-invasive way to implement cross-cutting concerns like authentication for gRPC services.
Server-Side Validation: A ServerInterceptor can inspect incoming call metadata, validate credentials, and abort the RPC if unauthorized.
Client-Side Propagation: AuthMetadataPlugin and metadata_call_credentials provide a clean, reusable pattern for attaching authentication tokens to outgoing gRPC calls.

Preview of the Next Lesson

We have spent the last two lessons focused on security, building a resilient and authenticated system. Now, we will shift our focus to another critical aspect of microservices: performance and load reduction. In the next lesson, you will learn how to implement a cache-aside strategy using Redis to reduce database load on read-heavy services, making our system faster and more scalable.

Recommended resources