OAuth2 Password Flow with JWT

Understanding the Password Flow Mechanism

The OAuth2 Password Flow is essentially a credentials-exchange protocol where a client sends a username and password directly to your token endpoint. FastAPI simplifies this by providing OAuth2PasswordRequestForm, a utility that expects form-encoded data—not JSON. When the user provides credentials, your server validates them against your database. If they are correct, you generate a JWT—a signed, base64-encoded string containing user claims—and return it to the client. The client then stores this token locally and includes it in the 'Authorization: Bearer <token>' header for subsequent requests. By using JWTs, your server remains stateless; you do not need to look up a session ID in a database for every incoming request. Instead, you verify the token's signature using a secret key. This decoupling allows your API to scale horizontally easily, as any server instance can validate the token independently without shared storage.

from fastapi import FastAPI, Depends
from fastapi.security import OAuth2PasswordRequestForm

app = FastAPI()

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    # In a real scenario, you would fetch the user from a DB here
    # and verify the password hash using a library like passlib.
    return {"access_token": "fake-jwt-token", "token_type": "bearer"}

JWT Generation and Signing

To generate a JWT, you generally use a library like PyJWT. The JWT consists of three parts: a header, a payload, and a cryptographic signature. The payload contains claims, which are statements about the user—typically their unique identifier or username—along with an expiration timestamp. The signature is created by hashing the header and payload with a secret key known only to your server. Why is this critical? If a user attempted to modify their username in the payload, the recalculated signature would not match the original signature, and your server would reject the token as forged. By setting an 'exp' (expiration) claim, you limit the window of opportunity for an attacker to use a leaked token. This process converts your authentication logic from a lookup problem into a verification problem, which is inherently faster and more resilient to high-traffic spikes in production environments.

import jwt
from datetime import datetime, timedelta

SECRET_KEY = "super-secret-key"
ALGORITHM = "HS256"

def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=30)
    to_encode.update({"exp": expire})
    # Encode the payload with the secret key and algorithm
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

Securing Endpoints with Dependencies

FastAPI leverages dependency injection to make securing your API incredibly clean. You define an 'OAuth2PasswordBearer' instance, which instructs the framework to look for the token in the 'Authorization' header of every request. By injecting this as a dependency, your protected endpoints will only execute if the request provides a valid token. The dependency first extracts the token string from the header and passes it to your validation logic. Within that logic, you decode the JWT using your secret key. If the token is expired or the signature is invalid, an exception is raised, and the request is automatically blocked with a 401 Unauthorized status code. This allows you to protect your business logic by simply adding a parameter to your route handler. It enforces a strict separation of concerns where authentication logic is centralized and reused across all sensitive operations.

from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.get("/users/me")
async def read_users_me(token: str = Depends(oauth2_scheme)):
    # Logic to decode and verify the token would go here
    return {"token": token}

Decoding and Verifying Claims

Once the token is extracted, you must decode it to retrieve the user's data. Decoding is not just about parsing the JSON; it is about verifying the signature and the expiration time. PyJWT does this automatically when you call 'jwt.decode'. If the token has expired, the library raises an exception, which you should catch to return an appropriate error message to the client. Beyond checking existence, you should verify the specific claims you stored, such as user roles or permissions, to enforce fine-grained access control. This verification process is the moment where you 'trust' the token. If an attacker manages to steal a token, they have the identity of that user until the token expires. This is why you must never expose the secret key in source control and should use environment variables for all sensitive configuration settings, ensuring the integrity of the signing process remains uncompromised.

import jwt

def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        # Decode will raise an error if the token is invalid or expired
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: str = payload.get("sub")
        return {"user_id": user_id}
    except jwt.PyJWTError:
        raise HTTPException(status_code=401, detail="Could not validate credentials")

Handling Token Expiration and Security

A crucial aspect of production-ready authentication is handling expiration gracefully. Because JWTs are stateless, you cannot 'log out' a user by deleting a record on the server; the token remains valid until it expires. Consequently, you must keep access tokens short-lived—for instance, 15 to 30 minutes. To maintain user experience without forcing frequent re-authentication, you might implement a refresh token strategy, where the client exchanges a long-lived refresh token for a new short-lived access token. Always force the use of HTTPS; without transport layer security, the JWT can be intercepted in transit, allowing an attacker to impersonate the user completely. By combining short-lived tokens, secure storage practices, and robust validation routines, you create a hardened authentication system that effectively protects your resources while maintaining the performance benefits of a stateless API architecture.

from fastapi import HTTPException

# Secure endpoint requiring authentication
@app.get("/data")
def get_secret_data(current_user = Depends(get_current_user)):
    if not current_user:
        raise HTTPException(status_code=403, detail="Access denied")
    return {"message": "Sensitive info revealed", "user": current_user}

Interview questions

All FastAPI interview questions →

Check yourself

Take the full FastAPI quiz →