Chapter 14 · Security Basics: OAuth2 and JWT

Chapter goals: Distinguish authentication from authorization, run through the complete OAuth2 password grant flow ("login for token → carry token to access → validate and extract user"), and use the official current recommended libraries: PyJWT and pwdlib.

14.1 Authentication vs Authorization

• Authentication: Who are you? — verifying username/password, token, etc.;
• Authorization: What are you allowed to do? — verifying whether you have permission to access a resource.

This chapter implements authentication + the most basic authorization (return 401 if not logged in); the next chapter's scopes handle fine-grained authorization. FastAPI provides several security utility classes in fastapi.security — they not only help parse credentials but also write the corresponding security scheme into the OpenAPI docs (the "Authorize" button auto-appears in Swagger UI).

14.2 Officially Recommended Libraries

Version evolution

Early official tutorials used passlib[bcrypt] for password hashing and python-jose for JWT. The current official tutorial has switched to pwdlib (recommended Argon2 algorithm) and PyJWT. python-jose is no longer maintained; passlib has not been updated for years and has compatibility issues with bcrypt 4.x. New projects should use pwdlib + PyJWT directly.

uv add pyjwt "pwdlib[argon2]"

14.3 OAuth2 Password Grant Full Flow

The overall flow has four steps:

1. Client submits username + password in form format to /token;
2. Server validates and issues a JWT (with expiration);
3. Client carries the token in subsequent requests via Authorization: Bearer <token> header;
4. Dependency extracts and decodes the token from the request, returning the current user.

Start with the skeleton (dependency injection part):

from datetime import datetime, timedelta, timezone
from typing import Annotated

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

app = FastAPI()

SECRET_KEY = "09d25e094faa..."   # Generate with openssl rand -hex 32 in production; put in env var!
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# tokenUrl points to the token endpoint, used by Swagger UI's Authorize form
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


class Token(BaseModel):
    access_token: str
    token_type: str


class User(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    disabled: bool = False


class UserInDB(User):
    hashed_password: str


async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},  # Spec requires this header on 401
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except jwt.InvalidTokenError:   # Covers expired, bad signature, all invalid cases
        raise credentials_exception
    user = get_user_from_db(username)
    if user is None:
        raise credentials_exception
    return user


async def get_current_active_user(
    current_user: Annotated[User, Depends(get_current_user)],
) -> User:
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user


@app.get("/users/me/")
async def read_me(current_user: Annotated[User, Depends(get_current_active_user)]):
    return current_user

OAuth2PasswordBearer(tokenUrl="token") itself does not validate anything — it is just an "extractor" dependency: it pulls the token string from the Authorization: Bearer xxx header; missing the header raises 401. Real validation happens in your get_current_user.

14.4 Token Creation and Password Hashing

from datetime import datetime, timedelta
from typing import Annotated
import jwt
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from passlib.context import CryptContext
from pydantic import BaseModel

app = FastAPI()

SECRET_KEY = "09d25e094faa..."
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")


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


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


def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)


# Fake user DB
fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "hashed_password": get_password_hash("secret"),
        "disabled": False,
    }
}


@app.post("/token", response_model=Token)
async def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
    user_dict = fake_users_db.get(form_data.username)
    if not user_dict or not verify_password(form_data.password, user_dict["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_dict["username"]}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

14.5 Chapter Summary

• Authentication = "who are you"; Authorization = "what can you do";
• OAuth2PasswordBearer extracts the Bearer token from the Authorization header;
• PyJWT encodes/decodes JWT; pwdlib (or passlib) hashes passwords;
• Token includes sub (subject/username) and exp (expiration);
• Dependency chain: get_current_user → get_current_active_user → endpoint.

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. What does OAuth2PasswordBearer do?

2. Why use PyJWT instead of python-jose?

3. What must the 401 response include for OAuth2 compliance?

4. What is stored in the JWT sub (subject) claim?

🛠️ Hands-On Exercises

1. Implement the full login flow: POST /token with OAuth2PasswordRequestForm, verify password with pwdlib, issue a JWT token; then use the token to access GET /users/me.
2. Deliberately send an expired token; verify the 401 response with WWW-Authenticate header.
3. Add a disabled flag to the user model; verify that disabled users receive a 400 error when using an otherwise valid token.

Proceed to Chapter 15: Security Advanced — Scopes and API Keys.