auth

auth

Authentication and authorisation patterns across the stack: JWTs, API keys, Supabase Auth, and role-based access control.

1) JWT generation and verification (Python)

from jose import jwt, JWTError
from datetime import datetime, timedelta, timezone

SECRET_KEY = os.getenv("JWT_SECRET")
ALGORITHM  = "HS256"

def create_token(user_id: str, role: str, expires_min: int = 60) -> str:
    payload = {
        "sub": user_id,
        "role": role,
        "exp": datetime.now(timezone.utc) + timedelta(minutes=expires_min),
        "iat": datetime.now(timezone.utc),
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

def verify_token(token: str) -> dict:
    try:
        return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    except JWTError as e:
        raise HTTPException(401, detail=f"Invalid token: {e}")

2) FastAPI security dependency

from fastapi import Depends, HTTPException, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

bearer = HTTPBearer()

async def get_current_user(
    creds: HTTPAuthorizationCredentials = Security(bearer),
) -> dict:
    return verify_token(creds.credentials)

async def require_admin(user: dict = Depends(get_current_user)) -> dict:
    if user.get("role") != "admin":
        raise HTTPException(403, "Admin only")
    return user

# Route usage
@router.delete("/article/{slug}", dependencies=[Depends(require_admin)])
async def delete_article(slug: str): ...

3) API key authentication

from fastapi import Header

API_KEYS = set(os.getenv("API_KEYS", "").split(","))   # comma-separated in .env

async def api_key_auth(x_api_key: str = Header(...)):
    if x_api_key not in API_KEYS:
        raise HTTPException(401, "Invalid API key")
    return x_api_key

# Rotating keys — store in Redis with expiry
async def validate_api_key(key: str) -> bool:
    return bool(await redis.get(f"apikey:{key}"))

4) Supabase Auth (server-side, Python)