Securing FastAPI Endpoints with JWT Authentication

1. What is JWT and Why is it Suited for Microservices?

In monolithic applications, like many traditional Django projects, authentication is often handled using server-side sessions. The server creates a session ID, stores it, and sends it to the client as a cookie. For every request, the server looks up the session to identify the user. This works well, but it requires a shared session store, which can become a bottleneck in a distributed system.

Microservices thrive on being stateless. A service shouldn't need to consult a central session store to know who a user is. This is where JSON Web Tokens (JWTs) come in.

A JWT is a compact, self-contained token that securely transmits user information. Think of it as a temporary ID card.

1. Authentication: You present your credentials (username/password) once at a login endpoint.
2. Issuance: The server validates you and issues a signed JWT.
3. Authorization: For every subsequent request to a protected resource, you present this JWT. The server can verify the signature on the token to trust its contents without needing to look up anything in a database.

A JWT consists of three parts separated by dots (.):

• Header: Metadata about the token, like the signing algorithm (e.g., HS256).
• Payload: The claims, or pieces of information about the user, such as their ID (sub), the token's expiration time (exp), and any custom data like roles.
• Signature: A cryptographic signature created using the header, the payload, and a secret key known only to the server. This ensures the token's integrity.

2. The Authentication Flow with an API Gateway

In our microservices architecture, the API Gateway is the perfect place to centralize authentication. It acts as the single point of entry and can enforce security policies before forwarding requests to internal services.

The flow works as follows:

1. Login Request: The client sends a POST request with the user's credentials to a /login or /token endpoint on the API Gateway.
2. Credential Validation: The Gateway forwards the credentials to the User service, which validates them against the database.
3. Token Issuance: If the credentials are correct, the Gateway generates a signed JWT and sends it back to the client.
4. Authenticated Request: The client stores the JWT and includes it in the Authorization header for all future requests to protected endpoints (e.g., Authorization: Bearer <your_jwt>).
5. Token Validation: The Gateway intercepts each request, extracts the JWT, and validates its signature and expiration. If valid, the request is passed on. If invalid, the Gateway immediately rejects the request with a 401 Unauthorized error.

3. Implementing JWT in a FastAPI Gateway

Now, let's get practical. We will implement this entire flow in FastAPI. The following video provides a clear, step-by-step guide to building a robust authentication system. We'll break it down into logical parts.

Step 1: Project Setup and Dependencies

First, we need to set up our project structure and install the necessary libraries for handling JWTs and hashing passwords. We'll use pyjwt for token management and bcrypt for secure password storage.

Step 2: Secure Configuration

A critical aspect of JWT security is the secret key, which is used to sign and verify tokens. This key must never be hardcoded in your application. We will load it from environment variables, a standard practice for production-grade applications. We'll also configure FastAPI's security scheme to recognize bearer tokens.

Step 3: Core Authentication Logic

This is where we handle user credentials. We need functions to:

1. Hash passwords when users register.
2. Verify a submitted password against its stored hash during login.
3. An authenticate_user function that encapsulates the logic of fetching a user and verifying their password.

Step 4: Token Generation and Validation

With user authentication in place, we can now create and validate JWTs.

• create_access_token: This function will take a user's identifier (like their email or ID), package it into a payload with an expiration time, and sign it to create the JWT.
• get_current_user: This will be our main FastAPI dependency. For any incoming request to a protected endpoint, this function will automatically run, extract the token from the header, decode it, validate its signature and expiration, and fetch the corresponding user from the database.

Step 5: The API Endpoints

Finally, we create two types of endpoints:

1. A public /token endpoint where users can log in.
2. One or more protected endpoints that require a valid JWT.

FastAPI's dependency injection system makes protecting an endpoint incredibly simple.

4. Hands-on Exercise: Secure an Endpoint

Now it's your turn to apply what you've learned. The code below contains a simplified FastAPI application with a public / route and an unprotected /profile/me route. Your task is to implement the necessary components to secure the /profile/me endpoint.

Instructions:

1. Make sure you have fastapi, uvicorn, and pyjwt installed (pip install "fastapi[all]" pyjwt).
2. Fill in the ... placeholders in the code.
3. Define the oauth2_scheme using OAuth2PasswordBearer, pointing it to the /token URL.
4. Implement the get_current_user dependency. For this exercise, you don't need a database. Simply decode the token and return the user's email from the sub claim. Handle potential JWTError if the token is invalid.
5. Add get_current_user as a dependency to the read_current_user_profile endpoint.
6. Run the app, use the /docs UI to log in at the /token endpoint, and then use the obtained token to authorize a request to /profile/me.

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import jwt, JWTError  # Using jose here for simplicity, pyjwt works similarly
from typing import Dict




# --- Configuration ---
SECRET_KEY = "your-super-secret-key"  # In production, use a secure, generated key from env vars
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

app = FastAPI()




# TODO 1: Define the oauth2_scheme using OAuth2PasswordBearer
# It should point to the "/token" endpoint.
oauth2_scheme = ...





# --- Mock Database and Utilities ---
# In a real app, this would be a real database.
fake_users_db = {
    "sagarmatha@example.com": {
        "username": "sagarmatha",
        "email": "sagarmatha@example.com",
        "hashed_password": "fakehashedpassword", # In a real app, use bcrypt
    }
}

def get_user(db, username: str):
    if username in db:
        return db[username]




# --- Security Functions ---




# TODO 2: Implement the get_current_user dependency
async def get_current_user(token: str = Depends(oauth2_scheme)) -> Dict:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:



        # Decode the token
        payload = ...  # Use jwt.decode with SECRET_KEY and ALGORITHM
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        



        # In a real app, you'd fetch the user from the DB
        user = get_user(fake_users_db, username)
        if user is None:
            raise credentials_exception
        return user
    except JWTError:



        # If decoding fails, raise the exception
        ...





# --- API Endpoints ---
@app.post("/token", response_model=Dict)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):



    # Simplified authentication
    user = get_user(fake_users_db, form_data.username)
    if not user or user["hashed_password"] != "fakehashedpassword": # Simplified check
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
        )



    # Create token
    to_encode = {"sub": user["email"]}
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return {"access_token": encoded_jwt, "token_type": "bearer"}

@app.get("/")
async def read_root():
    return {"message": "Welcome to the Secure API"}




# TODO 3: Protect this endpoint by adding the get_current_user dependency
@app.get("/profile/me")
async def read_current_user_profile():



    # The current_user object will be injected by FastAPI
    return {"user": "This should be the current user's data"}

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




# --- Configuration ---
SECRET_KEY = "your-super-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

app = FastAPI()




# 1. Define the oauth2_scheme
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")




# --- Mock Database and Utilities ---
fake_users_db = {
    "sagarmatha@example.com": {
        "username": "sagarmatha",
        "email": "sagarmatha@example.com",
        "hashed_password": "fakehashedpassword",
    }
}

def get_user(db, username: str):
    if username in db:
        return db[username]




# --- Security Functions ---




# 2. Implement the get_current_user dependency
async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]) -> Dict:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        
        user = get_user(fake_users_db, username)
        if user is None:
            raise credentials_exception
        return user
    except JWTError:
        raise credentials_exception




# --- API Endpoints ---
@app.post("/token", response_model=Dict)
async def login_for_access_token(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
    user = get_user(fake_users_db, form_data.username)
    if not user or user["hashed_password"] != "fakehashedpassword":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
        )
    to_encode = {"sub": user["email"]}
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return {"access_token": encoded_jwt, "token_type": "bearer"}

@app.get("/")
async def read_root():
    return {"message": "Welcome to the Secure API"}




# 3. Protect this endpoint by adding the get_current_user dependency
@app.get("/profile/me")
async def read_current_user_profile(current_user: Annotated[Dict, Depends(get_current_user)]):
    return {"user": current_user}

Conclusion

Congratulations! You have now secured your API Gateway, transforming it from an open door into a guarded entrance. This is a fundamental skill for any backend developer working with microservices.

Key Takeaways:

• JWTs provide a stateless, scalable authentication mechanism ideal for microservices.
• The API Gateway is the logical place to centralize user authentication, validating credentials and issuing tokens.
• The standard flow is: client logs in with credentials, receives a JWT, and includes it as a Bearer token in the Authorization header for subsequent requests.
• FastAPI's OAuth2PasswordBearer and dependency injection system (Depends) make implementing this flow straightforward and elegant.
• Protecting an endpoint is as simple as adding a dependency that validates the JWT and fetches the current user.

Preview of the Next Lesson

We have secured the "front door" of our system. But what about the communication happening inside? Right now, any service on our internal network can call any other service. In the next lesson, we will address this by learning how to secure service-to-service gRPC calls. We will propagate the JWT from the gateway to downstream gRPC services and validate it there, creating a truly zero-trust, end-to-end secure architecture.