Page cover

🔐Appendix C

JWT Authentication in a Microservice Architecture

Introduction

In modern applications, especially those built on a microservices architecture, services need to communicate with each other securely. A common challenge is handling user authentication and authorization across these distributed services. Sharing a single secret key (as used with the HS256 algorithm) among all services is a security risk; if one service is compromised, the entire system's secret is exposed.

A more secure and scalable solution is to use an asymmetric key pair (RS256). In this pattern, a dedicated Authentication Service signs JWTs with a private key, while multiple Resource Services verify those tokens using a corresponding public key.

This appendix provides a step-by-step guide to implementing this pattern using two FastAPI applications.

  • auth_service: The central authority that validates user credentials and issues JWTs. It is the only service that holds the private key.

  • course_service: A resource service that protects its endpoints and provides data only to requests with a valid JWT. It only needs the public key to verify tokens.

Core Concept: Asymmetric Keys (RS256)

Unlike symmetric algorithms (like HS256) that use one secret key for both signing and verifying, asymmetric algorithms use a pair of keys:

  • Private Key (.pem) 🔑: Kept secret and known only to the auth_service. It is used to sign (create) the JWT. Think of it as the unique, personal signature of the issuer.

  • Public Key (.pub) 🔓: Can be shared freely with any number of resource services. It is used to verify the signature of a JWT. Anyone with the public key can confirm that the token was signed by the holder of the private key, but they cannot create new valid tokens themselves.

This separation is the key to the pattern's security and scalability.

Implementation Walkthrough

Step 1: Generate the RSA Key Pair

First, you need to generate the private and public keys. You can do this using the openssl command-line tool.

Generate a 2048-bit RSA Private Key:

openssl genpkey -algorithm RSA -out private_key.pem -pkeyopt rsa_keygen_bits:2048

Extract the Public Key from the Private Key:

openssl rsa -pubout -in private_key.pem -out public_key.pem

You will now have two files in your directory: private_key.pem and public_key.pem.

Step 2: Build the Authentication Service

Create a file named auth_service.py. This service will issue tokens signed with the private key.

# auth_service.py

from datetime import datetime, timedelta, timezone
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from jose import jwt
from pwdlib import PasswordHash
from pydantic import BaseModel
from typing import Annotated

# --- Configuration & Setup ---
ALGORITHM = "RS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# Load the keys from files
with open("private_key.pem", "rb") as f:
    PRIVATE_KEY = f.read()

app = FastAPI(title="Authentication Service")
pwd_context = PasswordHash.recommended()

# --- Mock Database ---
fake_users_db = {
    "student1": {
        "username": "student1",
        "hashed_password": pwd_context.hash("password123"),
    }
}

# --- Pydantic Models ---
class Token(BaseModel):
    access_token: str
    token_type: str

# --- Core Logic ---
def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    # Use the private key to sign the JWT
    encoded_jwt = jwt.encode(to_encode, PRIVATE_KEY, algorithm=ALGORITHM)
    return encoded_jwt

# --- API Endpoint ---
@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
    user = fake_users_db.get(form_data.username)
    if not user or not verify_password(form_data.password, user["hashed_password"]):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(data={"sub": user["username"]})
    return {"access_token": access_token, "token_type": "bearer"}

Step 3: Build the Resource Service

Create a second file named course_service.py. This service will use the public key to validate tokens and protect its data.

# course_service.py

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from pydantic import BaseModel
from typing import Annotated

# --- Configuration & Setup ---
ALGORITHM = "RS256"

# Load the public key
with open("public_key.pem", "rb") as f:
    PUBLIC_KEY = f.read()

app = FastAPI(title="Course Service")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") # The client gets the token from the auth_service

# --- Pydantic Models ---
class TokenData(BaseModel):
    username: str | None = None

# --- Dependency for Verifying Token ---
async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        # Use the public key to verify the JWT
        payload = jwt.decode(token, PUBLIC_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception
    return token_data

# --- API Endpoints ---
@app.get("/courses")
async def get_courses(current_user: Annotated[TokenData, Depends(get_current_user)]):
    """
    Protected endpoint that returns a list of courses.
    Requires a valid JWT verified with the public key.
    """
    return [
        {"id": "CS101", "name": "Introduction to AI"},
        {"id": "SE305", "name": "Microservice Architecture Patterns"}
    ]

@app.get("/health")
async def health_check():
    return {"status": "ok"}

Step 4: Run and Test the Services

Open two separate terminal windows in your project directory.

1/ Start the Authentication Service on port 8081:

uvicorn auth_service:app --port 8081

2/ Start the Course Service on port 8082:

uvicorn course_service:app --port 8082 --reload

3/ Test the Flow:

  • Navigate to the auth_service docs at http://127.0.0.1:8081/docs.

  • Use the /token endpoint with username: "student1" and password: "password123" to get an access token.

  • Copy the access_token string.

  • Navigate to the course_service docs at http://127.0.0.1:8082/docs.

  • Click the "Authorize" button, paste the token into the value field, and authorize.

  • Now, execute the /courses endpoint. You should successfully receive the list of courses.

Architectural Benefits of This Pattern

  • No Shared Secrets: The highly sensitive private key is isolated within the auth_service. Resource services don't need it, minimizing the attack surface.

  • Decentralized Verification: Any number of microservices can be given the public key to verify tokens independently. This allows them to protect their endpoints without needing to call the auth_service for every request, which is highly efficient and scalable.

  • Improved Security Posture: If a course_service instance is compromised, attackers cannot issue new, valid tokens because they do not have the private key. The scope of the breach is limited.

PreviousAppendix B

Last updated