Before users can chat, we need to know who they are. This chapter focuses on implementing a basic user authentication system using JSON Web Tokens (JWT) in FastAPI. JWTs are a common, secure way to transmit information between parties as a JSON object, ideal for stateless authentication in APIs.

Purpose of this Chapter

By the end of this chapter, you will:

  • Understand what JWTs are and why they are used for authentication.
  • Set up libraries for password hashing and JWT generation.
  • Implement user creation and login endpoints.
  • Create a dependency to protect FastAPI routes with JWT.

Concepts Explained: JWT and Hashing

JSON Web Tokens (JWT)

A JWT is a compact, URL-safe means of representing claims to be transferred between two parties. The claims in a JWT are encoded as a JSON object and are digitally signed using a secret (with HMAC algorithm) or a public/private key pair (with RSA or ECDSA).

Why JWT?

  • Statelessness: The server doesn’t need to store session information. Each request contains the token, which the server can validate.
  • Scalability: Easier to scale horizontally as no session state needs to be shared between servers.
  • Security: Signed tokens prevent tampering.

A typical JWT looks like header.payload.signature. The header and payload are Base64Url encoded JSON objects.

Password Hashing

Never store plain text passwords! If your database is breached, all user accounts are compromised. Instead, we store a hash of the password. Hashing is a one-way function: you can create a hash from a password, but you cannot easily revert a hash back to the original password.

We’ll use passlib with the bcrypt hashing scheme, which is generally recommended as it’s designed to be slow and resistant to brute-force attacks.

Step-by-Step Tasks

1. Install Authentication Libraries

First, stop your Uvicorn server (Ctrl+C). We need to install python-jose for JWT handling and passlib for password hashing, along with bcrypt as its backend.

pipenv install python-jose[cryptography] passlib[bcrypt]

2. Create app/auth.py for Authentication Logic

Let’s create a new file app/auth.py to encapsulate our authentication-related functions and models.

# app/auth.py

from datetime import datetime, timedelta
from typing import Optional

from jose import JWTError, jwt
from passlib.context import CryptContext

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from pydantic import BaseModel

# --- Configuration (usually from environment variables) ---
# In a real app, use environment variables for these secrets!
SECRET_KEY = "your-super-secret-key"  # Change this in production!
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# --- Password Hashing ---
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

class Hasher:
    @staticmethod
    def verify_password(plain_password: str, hashed_password: str) -> bool:
        return pwd_context.verify(plain_password, hashed_password)

    @staticmethod
    def get_password_hash(password: str) -> str:
        return pwd_context.hash(password)

# --- JWT Token Models ---
class Token(BaseModel):
    access_token: str
    token_type: str = "bearer"

class TokenData(BaseModel):
    username: Optional[str] = None

# --- JWT Token Functions ---
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

# --- OAuth2PasswordBearer for dependency injection ---
# This is how FastAPI knows to look for a token in the Authorization header
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") # "token" is our login endpoint

async def get_current_user(token: str = Depends(oauth2_scheme)):
    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
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception
    # In a real app, you would fetch the user from a database here
    # For now, we'll just return the username from the token
    return token_data.username

Code Explanation (app/auth.py):

  • Configuration: SECRET_KEY, ALGORITHM, ACCESS_TOKEN_EXPIRE_MINUTES. Crucially, SECRET_KEY MUST be a strong, randomly generated string and ideally loaded from environment variables in production.
  • Hasher Class: Provides static methods to verify_password against a hash and get_password_hash for new passwords.
  • Token and TokenData: Pydantic models for the structure of our JWT tokens and their payload.
  • create_access_token: Function to generate a new JWT with an expiration time.
  • OAuth2PasswordBearer: FastAPI’s way to define that security scheme. It expects tokens in the Authorization: Bearer <token> header. tokenUrl specifies where the client can obtain the token.
  • get_current_user: This is an asynchronous dependency function. It extracts the token, decodes it, validates it, and retrieves the username (subject, “sub”). If anything fails, it raises an HTTPException.

3. Update app/main.py for Auth Endpoints

Now, let’s integrate these authentication functions into our app/main.py. For simplicity, we’ll use an in-memory “database” of users for now. Later, we’ll integrate a real database.

# app/main.py (updated)

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from typing import Dict

from .auth import Hasher, create_access_token, get_current_user, ACCESS_TOKEN_EXPIRE_MINUTES, Token
from datetime import timedelta

app = FastAPI()

# --- Temporary User Storage (Replace with a real database later) ---
fake_users_db = {
    "testuser": {
        "username": "testuser",
        "hashed_password": Hasher.get_password_hash("password123"), # Hash a default password
    }
}

@app.get("/")
async def read_root():
    return {"message": "Welcome to the Real-time Chat API!"}

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

# Endpoint for user login to get a JWT
@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = fake_users_db.get(form_data.username)
    if not user or not Hasher.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_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user["username"]}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

# Protected endpoint example (requires a valid JWT)
@app.get("/users/me")
async def read_users_me(current_user: str = Depends(get_current_user)):
    return {"username": current_user, "message": "You are authenticated!"}

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            print(f"Received message: {data}")
            await websocket.send_text(f"Message text was: {data}")
    except WebSocketDisconnect:
        print("Client disconnected.")

Code Explanation (app/main.py updates):

  • Imports: We added Depends, HTTPException, status for handling HTTP dependencies and exceptions, and OAuth2PasswordRequestForm for the login endpoint.
  • fake_users_db: A simple dictionary to simulate a user database. testuser with password password123.
  • login_for_access_token endpoint:
    • This is a POST endpoint at /token.
    • It uses OAuth2PasswordRequestForm to expect username and password as form data.
    • It retrieves the user from our fake database and verifies the password using Hasher.verify_password.
    • If credentials are valid, it calls create_access_token to generate a JWT.
    • It returns the access_token and token_type (required by OAuth2 specification).
  • read_users_me endpoint:
    • This is a GET endpoint at /users/me.
    • current_user: str = Depends(get_current_user): This is the magic of FastAPI dependencies. Before this function runs, get_current_user is executed. If get_current_user successfully validates a JWT, it returns the username, which is then passed as current_user to this function. If validation fails, get_current_user raises an HTTPException, and read_users_me is never called.

4. Run and Test Authentication

  1. Start the server:

    pipenv shell
uvicorn app.main:app --reload

  2. Access Swagger UI: Open your browser to http://127.0.0.1:8000/docs.

  3. Test the /token endpoint (Login):

    • Find the /token endpoint in Swagger UI.
    • Click “Try it out”.
    • Enter testuser for username and password123 for password.
    • Click “Execute”.
    • You should get a 200 OK response with an access_token and token_type: bearer. Copy the access_token.

  4. Authorize in Swagger UI:

    • Click the “Authorize” button at the top right of the Swagger UI page.
    • In the “Value” field, enter Bearer YOUR_ACCESS_TOKEN (replace YOUR_ACCESS_TOKEN with the token you copied, ensuring there’s a space after Bearer).
    • Click “Authorize” and then “Close”.

  5. Test the /users/me (Protected Endpoint):

    • Find the /users/me endpoint.
    • Click “Try it out” and then “Execute”.
    • You should get a 200 OK response with {"username": "testuser", "message": "You are authenticated!"}.

    If you try to access /users/me without authorizing or with an invalid token, you will get a 401 Unauthorized error.

Pitfalls and Best Practices

  • Secret Key Management: The SECRET_KEY in app/auth.py is hardcoded. For production, this MUST be loaded from environment variables.
  • User Storage: fake_users_db is a simple dictionary. In a real application, this would be replaced by a database interaction to store and retrieve user data securely.
  • Token Expiration: JWTs have an expiration. Once expired, the user needs to re-authenticate (get a new token).
  • Refresh Tokens: For better UX, production apps often use short-lived access tokens and longer-lived refresh tokens to avoid frequent re-logins. We are not implementing refresh tokens in this basic guide, but it’s an important concept to be aware of.

Summary/Key Takeaways

You’ve successfully set up basic JWT-based authentication for your FastAPI application. You can now register POST and GET endpoints, verify user credentials, and issue access tokens. More importantly, you’ve learned how to protect your API endpoints using FastAPI’s dependency injection system, ensuring only authenticated users can access certain resources.

In the next chapter, we’ll extend this by managing multiple active WebSocket connections, which is essential for a functional chat application.