FastAPI Cheatsheet 05 — Authentication and Authorization

from passlib.context import CryptContext
pwd = CryptContext(schemes=["argon2"], deprecated="auto")

hashed = pwd.hash(plain)
ok = pwd.verify(plain, hashed)

from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from datetime import datetime, timedelta
import jwt

SECRET = settings.secret_key
ALG = "HS256"
ACCESS_MIN = 30

oauth2 = OAuth2PasswordBearer(tokenUrl="token")

@app.post("/token")
async def token(form: OAuth2PasswordRequestForm = Depends(), db = Depends(get_db)):
    user = await find_user(db, form.username)
    if not user or not pwd.verify(form.password, user.hashed_password):
        raise HTTPException(401, "Invalid creds")
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_MIN)
    t = jwt.encode({"sub": str(user.id), "exp": expire}, SECRET, ALG)
    return {"access_token": t, "token_type": "bearer"}

async def current_user(token: str = Depends(oauth2), db = Depends(get_db)) -> User:
    try:
        payload = jwt.decode(token, SECRET, algorithms=[ALG])   # always pin algorithms!
    except jwt.PyJWTError:
        raise HTTPException(401, "Invalid token")
    user = await db.get(User, int(payload["sub"]))
    if not user: raise HTTPException(401, "User missing")
    return user

class RefreshToken(Base):
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int]
    token: Mapped[str] = mapped_column(unique=True, index=True)
    expires_at: Mapped[datetime]
    revoked: Mapped[bool] = mapped_column(default=False)

@app.post("/refresh")
async def refresh(req: RefreshIn, db = Depends(get_db)):
    rec = await db.scalar(select(RefreshToken).where(
        RefreshToken.token == req.refresh_token,
        RefreshToken.revoked == False,
        RefreshToken.expires_at > datetime.utcnow(),
    ))
    if not rec: raise HTTPException(401)
    return {"access_token": make_access_token(rec.user_id), "token_type": "bearer"}

from itsdangerous import URLSafeSerializer
serializer = URLSafeSerializer(SECRET)

@app.post("/login")
async def login(form: OAuth2PasswordRequestForm = Depends(), response: Response = None, db = Depends(get_db)):
    user = await authenticate(form.username, form.password, db)
    sid = secrets.token_urlsafe(32)
    await redis.set(f"sess:{sid}", str(user.id), ex=86400 * 30)
    response.set_cookie("session", sid, httponly=True, secure=True, samesite="lax")
    return {"ok": True}

async def current_user_session(session: str = Cookie(...), db = Depends(get_db)) -> User:
    uid = await redis.get(f"sess:{session}")
    if not uid: raise HTTPException(401)
    return await db.get(User, int(uid))

from fastapi.security import APIKeyHeader

api_key_h = APIKeyHeader(name="X-API-Key")

async def get_api_key(key: str = Depends(api_key_h), db = Depends(get_db)) -> ApiKey:
    rec = await db.scalar(select(ApiKey).where(
        ApiKey.key_hash == sha256(key), ApiKey.revoked == False,
    ))
    if not rec: raise HTTPException(401, "Invalid key")
    return rec

oauth2 = OAuth2PasswordBearer(
    tokenUrl="token",
    scopes={"read": "...", "write": "...", "admin": "..."},
)

from fastapi import Security
from fastapi.security import SecurityScopes

async def user_with_scope(security_scopes: SecurityScopes, token: str = Depends(oauth2)):
    payload = jwt.decode(token, SECRET, algorithms=[ALG])
    granted = payload.get("scopes", [])
    for needed in security_scopes.scopes:
        if needed not in granted:
            raise HTTPException(403, f"missing scope: {needed}")
    return await db.get(User, int(payload["sub"]))

@app.post("/items", dependencies=[Security(user_with_scope, scopes=["write"])])
async def create_(...): ...

from enum import Enum
class Role(str, Enum):
    USER = "user"; EDITOR = "editor"; ADMIN = "admin"

def require_role(*roles: Role):
    def dep(u: User = Depends(current_user)):
        if u.role not in roles: raise HTTPException(403)
        return u
    return dep

@app.delete("/posts/{id}")
async def delete_(id: int, u: User = Depends(require_role(Role.EDITOR, Role.ADMIN))): ...

async def get_editable_post(id: int, u: User = Depends(current_user), db = Depends(get_db)) -> Post:
    p = await db.get(Post, id)
    if not p: raise HTTPException(404)
    if p.author_id != u.id and not u.is_admin:
        raise HTTPException(403)
    return p

@app.patch("/posts/{id}")
async def update_(p: Post = Depends(get_editable_post), data: PostUpdate = Body(...)):
    ...

from authlib.integrations.starlette_client import OAuth

oauth = OAuth()
oauth.register(
    name="google",
    server_metadata_url="https://accounts.google.com/.well-known/openid-configuration",
    client_id=settings.google_client_id,
    client_secret=settings.google_client_secret,
    client_kwargs={"scope": "openid email profile"},
)

@app.get("/auth/google")
async def google_login(req: Request):
    return await oauth.google.authorize_redirect(req, req.url_for("google_callback"))

@app.get("/auth/google/callback", name="google_callback")
async def google_cb(req: Request, db = Depends(get_db)):
    token = await oauth.google.authorize_access_token(req)
    info = token.get("userinfo")
    user = await find_or_create_user(db, info["email"])
    # issue your own session/JWT
    return RedirectResponse("/")

from slowapi import Limiter
limiter = Limiter(key_func=lambda req: req.client.host)

@app.post("/token")
@limiter.limit("5/minute")
async def token(...): ...

@app.post("/forgot-password")
async def forgot(req: ForgotIn, db = Depends(get_db)):
    user = await db.scalar(select(User).where(User.email == req.email))
    if user:                       # always same response shape
        t = secrets.token_urlsafe(32)
        await redis.set(f"pwreset:{t}", str(user.id), ex=3600)
        await send_email(user.email, f"...?token={t}")
    return {"ok": True}            # don't leak whether email exists

@app.middleware("http")
async def sec(req, call_next):
    resp = await call_next(req)
    resp.headers["X-Content-Type-Options"] = "nosniff"
    resp.headers["X-Frame-Options"] = "DENY"
    resp.headers["Strict-Transport-Security"] = "max-age=63072000; includeSubDomains; preload"
    return resp

from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://yourdomain.com"],
    allow_credentials=True, allow_methods=["*"], allow_headers=["*"],
)