FastAPI xác thực bằng JWT – OAuth2PasswordBearer, PyJWT, pwdlib

uv add "fastapi[standard]" pyjwt "pwdlib[argon2]"
export JWT_SECRET="$(openssl rand -hex 32)"     # production: lấy từ secret manager
export DEMO_PASSWORD='***'                       # chỉ cho ví dụ dưới

import os
from datetime import UTC, datetime, timedelta
from typing import Annotated
 
import jwt
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jwt.exceptions import InvalidTokenError
from pwdlib import PasswordHash
from pydantic import BaseModel
 
SECRET_KEY = os.environ["JWT_SECRET"]          # openssl rand -hex 32
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE = timedelta(minutes=30)
 
password_hash = PasswordHash.recommended()     # Argon2
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
DUMMY_HASH = password_hash.hash("dummy")
 
# Demo: thay bằng bảng users trong DB
USERS = {
    "netops": {
        "username": "netops",
        "hashed_password": password_hash.hash(os.environ["DEMO_PASSWORD"]),
        "scopes": ["read"],
        "disabled": False,
    }
}
 
 
class Token(BaseModel):
    access_token: str
    token_type: str = "bearer"
 
 
class User(BaseModel):
    username: str
    scopes: list[str] = []
    disabled: bool = False
 
 
def authenticate(username: str, password: str) -> User | None:
    row = USERS.get(username)
    if row is None:
        password_hash.verify(password, DUMMY_HASH)  # tốn thời gian như user thật
        return None
    if not password_hash.verify(password, row["hashed_password"]):
        return None
    return User(**row)
 
 
 
def create_access_token(sub: str, expires: timedelta = ACCESS_TOKEN_EXPIRE) -> str:
    now = datetime.now(UTC)
    payload = {"sub": sub, "iat": now, "exp": now + expires}
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
 
 
async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]) -> User:
    cred_exc = 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])  # check chữ ký + exp
        username = payload.get("sub")
    except InvalidTokenError:
        raise cred_exc from None
    row = USERS.get(username) if username else None
    if row is None:
        raise cred_exc
    return User(**row)
 
 
async def get_active_user(user: Annotated[User, Depends(get_current_user)]) -> User:
    if user.disabled:
        raise HTTPException(status_code=403, detail="Inactive user")
    return user
 
 
CurrentUser = Annotated[User, Depends(get_active_user)]
app = FastAPI()
 
 
@app.post("/token")
async def login(form: Annotated[OAuth2PasswordRequestForm, Depends()]) -> Token:
    user = authenticate(form.username, form.password)
    if user is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return Token(access_token=create_access_token(user.username))
 
 
@app.get("/users/me")
async def me(user: CurrentUser) -> User:
    return user

TOKEN=$(curl -s -X POST localhost:8000/token \
  -d username=netops -d "password=$DEMO_PASSWORD" | jq -r .access_token)
curl -s localhost:8000/users/me -H "Authorization: Bearer $TOKEN"
# Giải mã phần payload để xem (KHÔNG kiểm chữ ký, chỉ để debug)
uv run python -c 'import jwt,sys; print(jwt.decode(sys.argv[1], options={"verify_signature": False}))' "$TOKEN"

from datetime import timedelta
 
from fastapi.testclient import TestClient
 
from main import app, create_access_token
 
client = TestClient(app)
 
 
def test_expired_token_rejected():
    expired = create_access_token("netops", expires=timedelta(seconds=-1))
    r = client.get("/users/me", headers={"Authorization": f"Bearer {expired}"})
    assert r.status_code == 401