Recap — Token-Based Auth Flow

# Token-Based Auth Flow JWT (JSON Web Tokens) + OAuth2's password flow is the dominant auth pattern for APIs. Sign in once; get a token; include the token in subsequent requests. ## The flow ``` 1. POST /signup { email, password } → Server hashes password with bcrypt → Saves user → Returns user info (no token) 2. POST /login { email, password } → Server verifies password against bcrypt hash → Generates JWT with user id + expiry → Returns { access_token, token_type: "bearer" } 3. GET /me (with `Authorization: Bearer `) → Server verifies JWT signature + expiry → Decodes user id from payload → Looks up user; returns ``` ## Hash passwords (never store plaintext) ```python from passlib.context import CryptContext pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") def hash_password(password: str) -> str: return pwd_context.hash(password) def verify_password(plain: str, hashed: str) -> bool: return pwd_context.verify(plain, hashed) ``` Bcrypt is slow on purpose (~100ms per hash). Prevents brute-force attacks even if the hash leaks. For 2026, argon2 is also a great choice. ## Generate JWT ```python from datetime import datetime, timedelta from jose import jwt # python-jose library SECRET_KEY = settings.secret_key ALGORITHM = "HS256" def create_access_token(user_id: int, expires_delta: timedelta = timedelta(hours=1)) -> str: expire = datetime.utcnow() + expires_delta payload = {"sub": str(user_id), "exp": expire} return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM) ``` `sub` (subject) = the user ID. `exp` = expiry timestamp. ## OAuth2 password flow setup ```python from fastapi import Depends, FastAPI, HTTPException, status from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from sqlalchemy.orm import Session app = FastAPI() oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login") class LoginResponse(BaseModel): access_token: str token_type: str @app.post("/login", response_model=LoginResponse) def login( form: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db), ): user = db.execute(select(User).where(User.email == form.username)).scalar_one_or_none() if not user or not verify_password(form.password, user.password_hash): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials", headers={"WWW-Authenticate": "Bearer"}, ) token = create_access_token(user.id) return LoginResponse(access_token=token, token_type="bearer") ``` `OAuth2PasswordRequestForm` expects form-encoded `username` and `password`. By convention, the `username` field holds the email. ## get_current_user dependency ```python def get_current_user( token: str = Depends(oauth2_scheme), db: Session = Depends(get_db), ) -> User: 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]) user_id = int(payload.get("sub")) except (jwt.JWTError, ValueError): raise credentials_exception user = db.get(User, user_id) if not user: raise credentials_exception return user ``` Used in any protected endpoint: ```python @app.get("/me", response_model=UserResponse) def me(current_user: User = Depends(get_current_user)): return current_user ``` ## Signup ```python class UserCreate(BaseModel): email: EmailStr password: str name: str @app.post("/signup", response_model=UserResponse, status_code=201) def signup(payload: UserCreate, db: Session = Depends(get_db)): existing = db.execute(select(User).where(User.email == payload.email)).scalar_one_or_none() if existing: raise HTTPException(409, "Email already registered") user = User( email=payload.email, password_hash=hash_password(payload.password), name=payload.name, ) db.add(user) db.commit() db.refresh(user) return user ``` ## Refresh tokens Short-lived access tokens + long-lived refresh tokens: ```python @app.post("/login", response_model=LoginResponse) def login(...): access_token = create_access_token(user.id, expires_delta=timedelta(minutes=15)) refresh_token = create_refresh_token(user.id, expires_delta=timedelta(days=30)) return {"access_token": access_token, "refresh_token": refresh_token, "token_type": "bearer"} @app.post("/refresh", response_model=LoginResponse) def refresh(refresh_token: str): # Validate refresh token # Issue new access token ... ``` For revocation: store refresh tokens in DB; check on each refresh. ## Client usage ```bash # Login curl -X POST http://localhost:8000/login \ -F "username=alice@example.com" \ -F "password=secret" # Returns: {"access_token": "eyJ