pip install slowapi

# Rate limiting robuste avec slowapi (basé sur limits + un backend Redis en production)
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address, storage_uri="redis://localhost:6379")
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)


@app.post("/auth/login")
@limiter.limit("5/minute")  # 5 tentatives de connexion max par IP et par minute
def login(request: Request, form_data: OAuth2PasswordRequestForm = Depends()):
    ...


@app.get("/api/public-data")
@limiter.limit("100/minute")
def public_data(request: Request):
    return {"data": "..."}

# Headers de sécurité HTTP via un middleware dédié
from starlette.middleware.base import BaseHTTPMiddleware


class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        response = await call_next(request)
        response.headers["X-Content-Type-Options"] = "nosniff"
        response.headers["X-Frame-Options"] = "DENY"
        response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
        response.headers["Strict-Transport-Security"] = "max-age=63072000; includeSubDomains"
        response.headers["Permissions-Policy"] = "camera=(), microphone=(), geolocation=()"
        return response


app.add_middleware(SecurityHeadersMiddleware)

# Gestion des secrets — jamais en dur, chargés et validés via pydantic-settings
from pydantic_settings import BaseSettings, SettingsConfigDict
from pydantic import SecretStr


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env")

    database_url: SecretStr   # SecretStr masque la valeur dans les logs/repr accidentels
    jwt_secret_key: SecretStr
    stripe_api_key: SecretStr


settings = Settings()

# Utilisation : settings.jwt_secret_key.get_secret_value() -- explicite, pas d'exposition accidentelle
print(settings.jwt_secret_key)  # affiche "SecretStr('**********')", pas la vraie valeur

# Protection contre les injections SQL — TOUJOURS via l'ORM ou des requêtes paramétrées
from sqlalchemy import text

# MAUVAIS — injection SQL possible
def bad_search(db: Session, name: str):
    return db.execute(text(f"SELECT * FROM products WHERE name = '{name}'"))  # DANGER

# BON — requête paramétrée
def good_search(db: Session, name: str):
    return db.execute(text("SELECT * FROM products WHERE name = :name"), {"name": name})

# MEILLEUR — via l'ORM, paramétrage automatique
def best_search(db: Session, name: str):
    return db.query(Product).filter(Product.name == name).all()

# Limiter la taille des payloads entrants — prévention basique de DoS applicatif
from starlette.middleware.base import BaseHTTPMiddleware

MAX_BODY_SIZE = 1 * 1024 * 1024  # 1 Mo


class LimitBodySizeMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        content_length = request.headers.get("content-length")
        if content_length and int(content_length) > MAX_BODY_SIZE:
            from starlette.responses import JSONResponse
            return JSONResponse({"error": "Payload trop volumineux"}, status_code=413)
        return await call_next(request)


app.add_middleware(LimitBodySizeMiddleware)

# Audit trail — journaliser les actions sensibles (qui, quoi, quand) pour la traçabilité
async def log_sensitive_action(db: AsyncSession, user_id: int, action: str, resource_id: int):
    audit_entry = AuditLog(
        user_id=user_id,
        action=action,             # ex. "product.delete"
        resource_id=resource_id,
        timestamp=datetime.now(timezone.utc),
    )
    db.add(audit_entry)
    await db.commit()


@router.delete("/products/{id}")
async def delete_product(
    id: int,
    db: Annotated[AsyncSession, Depends(get_db)],
    current_user: Annotated[User, Depends(require_role(Role.ADMIN))],
):
    product = await db.get(Product, id)
    await db.delete(product)
    await db.commit()
    await log_sensitive_action(db, current_user.id, "product.delete", id)
    return {"deleted": id}