"""Durcissement : en-têtes, limitation de débit, quotas, journal d'audit. Modèle de menace retenu pour un service public exposé : - force brute sur le PIN et énumération de numéros ; - déni de service (flood HTTP, WebSocket, corps de requête géants) ; - XSS / clickjacking / injection via la PWA ; - abus du relais WebRTC pour joindre des utilisateurs arbitraires ; - élévation de privilèges via les routes d'administration. """ from __future__ import annotations import hashlib import ipaddress import logging import re import time from collections import defaultdict, deque from fastapi import HTTPException, Request from fastapi.responses import JSONResponse from starlette.middleware.base import BaseHTTPMiddleware from . import db log = logging.getLogger("security") # --- Limites --------------------------------------------------------------- MAX_BODY_BYTES = 64 * 1024 # aucune route légitime n'envoie plus (images en P2P) # Seule exception : le tableur de bénévoles importé par un superadmin. La route # est authentifiée et réservée au superadmin, le risque de flood est nul. UPLOAD_PATHS = ("/api/admin/users/import",) MAX_UPLOAD_BYTES = 2 * 1024 * 1024 MAX_WS_FRAME_BYTES = 96 * 1024 # la signalisation SDP peut être volumineuse MAX_WS_PER_USER = 5 # onglets/appareils simultanés MAX_WS_FRAMES_PER_MINUTE = 600 # Rate limit windows : (nombre de requêtes autorisées, durée en secondes) # # Attention au NAT : sur un site de festival, des centaines de téléphones # partagent la même IP publique. Un plafond par IP trop bas exclurait des # utilisateurs légitimes. On limite donc par identité quand elle existe # (porteur d'un token), et l'IP ne sert que de garde-fou large. RATE_RULES = { "auth": (60, 300), # /api/auth/* par IP : garde-fou anti-sweep "api": (600, 60), # reste de l'API, par utilisateur authentifié "admin": (300, 60), # routes d'administration } # Limite par numéro visé sur les routes d'authentification. C'est elle qui # arrête le brute force ciblé, en complément du lockout de compte (5 essais). AUTH_PER_PHONE = (10, 300) class RateLimiter: """Sliding window en mémoire. Suffisant tant qu'on tourne en instance unique.""" def __init__(self) -> None: self._hits: dict[str, deque[float]] = defaultdict(deque) self._last_sweep = time.monotonic() def check(self, key: str, limit: int, window: float) -> tuple[bool, int]: now = time.monotonic() self._sweep(now) bucket = self._hits[key] while bucket and now - bucket[0] > window: bucket.popleft() if len(bucket) >= limit: retry = int(window - (now - bucket[0])) + 1 return False, retry bucket.append(now) return True, 0 def _sweep(self, now: float) -> None: """Purge périodique pour éviter une croissance mémoire non bornée.""" if now - self._last_sweep < 300: return self._last_sweep = now for key in [k for k, v in self._hits.items() if not v or now - v[-1] > 3600]: self._hits.pop(key, None) def reset(self) -> None: """Remise à zéro complète (utilisée par les tests).""" self._hits.clear() limiter = RateLimiter() def client_ip(request: Request) -> str: """IP réelle derrière le proxy Hugging Face (X-Forwarded-For).""" forwarded = request.headers.get("x-forwarded-for", "") if forwarded: candidate = forwarded.split(",")[0].strip() try: ipaddress.ip_address(candidate) return candidate except ValueError: pass return request.client.host if request.client else "inconnu" def enforce(request: Request, bucket: str) -> None: """Applique une règle de rate limit, ou lève une 429.""" limit, window = RATE_RULES[bucket] ok, retry = limiter.check(f"{bucket}:{client_ip(request)}", limit, window) if not ok: raise HTTPException( status_code=429, detail="Trop de requêtes. Réessayez dans un instant.", headers={"Retry-After": str(retry)}, ) # --- Intergiciels ---------------------------------------------------------- # CSP stricte côté scripts : c'est là que se joue le XSS. `script-src 'self'` # interdit tout script inline et toute origine externe. # # `style-src` tolère l'inline, car les largeurs de jauges, de barres de # progression et la position des tooltips sont calculées à l'exécution. Le # risque résiduel (injection CSS) est contenu : `img-src` et `connect-src` # sont verrouillés sur l'origine, donc aucun canal d'exfiltration n'est ouvert. CSP = "; ".join( [ "default-src 'self'", "script-src 'self'", "style-src 'self' 'unsafe-inline'", # blob: pour afficher les photos reçues en P2P, data: pour le grain SVG. "img-src 'self' blob: data:", "font-src 'self'", "connect-src 'self' ws: wss:", "media-src 'self' blob:", "object-src 'none'", "base-uri 'none'", "form-action 'self'", "frame-ancestors 'none'", "manifest-src 'self'", "worker-src 'self'", "upgrade-insecure-requests", ] ) SECURITY_HEADERS = { "Content-Security-Policy": CSP, "X-Content-Type-Options": "nosniff", "X-Frame-Options": "DENY", "Referrer-Policy": "no-referrer", "Cross-Origin-Opener-Policy": "same-origin", "Cross-Origin-Resource-Policy": "same-origin", "Permissions-Policy": ( "geolocation=(), microphone=(), camera=(), payment=(), usb=(), " "magnetometer=(), accelerometer=(), interest-cohort=()" ), # HSTS : Hugging Face termine le TLS, on est toujours en HTTPS en production. "Strict-Transport-Security": "max-age=31536000; includeSubDomains", } class SecurityHeadersMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): response = await call_next(request) for key, value in SECURITY_HEADERS.items(): response.headers.setdefault(key, value) # L'en-tête `Server` est posé par uvicorn : il est retiré via # --no-server-header au lancement, pas ici. return response class BodyLimitMiddleware(BaseHTTPMiddleware): """Rejette les corps de requête surdimensionnés avant de les lire.""" async def dispatch(self, request: Request, call_next): declared = request.headers.get("content-length") ceiling = ( MAX_UPLOAD_BYTES if request.url.path.startswith(UPLOAD_PATHS) else MAX_BODY_BYTES ) if declared and declared.isdigit() and int(declared) > ceiling: return JSONResponse({"detail": "Requête trop volumineuse."}, status_code=413) return await call_next(request) def _identity(request: Request) -> str: """Clé de rate limit : l'utilisateur authentifié si possible, l'IP sinon. Le token n'est pas décodé — son empreinte suffit à distinguer deux sessions, et un token invalide retombera de toute façon sur un 401. """ header = request.headers.get("authorization", "") if header.lower().startswith("bearer "): token = header.split(" ", 1)[1].strip() if token: return "u:" + hashlib.sha256(token.encode()).hexdigest()[:20] return "ip:" + client_ip(request) def check_phone_attempt(phone: str) -> None: """Rate limit par numéro visé sur les routes d'authentification.""" limit, window = AUTH_PER_PHONE ok, retry = limiter.check(f"phone:{phone}", limit, window) if not ok: raise HTTPException( status_code=429, detail="Trop de tentatives sur ce numéro. Réessayez dans quelques minutes.", headers={"Retry-After": str(retry)}, ) def check_self_registration(ip: str) -> None: """Borne les créations spontanées de comptes depuis une même IP. Le seuil est volontairement large : sur un site de festival, tous les téléphones sortent par la même IP publique. Il n'est là que pour empêcher un script de remplir la base de comptes fantômes. """ from . import config ok, retry = limiter.check( f"selfreg:{ip}", config.SELF_REGISTRATION_PER_IP, config.SELF_REGISTRATION_WINDOW_SECONDS, ) if not ok: audit(None, "self_registration_throttled", ip=ip) raise HTTPException( status_code=429, detail="Trop d'inscriptions depuis ce réseau. Réessayez plus tard " "ou demandez à un administrateur de vous enregistrer.", headers={"Retry-After": str(retry)}, ) class RateLimitMiddleware(BaseHTTPMiddleware): """Rate limit par identité authentifiée, avec l'IP comme garde-fou.""" async def dispatch(self, request: Request, call_next): path = request.url.path bucket = None if path.startswith("/api/auth"): bucket = "auth" elif path.startswith("/api/admin"): bucket = "admin" elif path.startswith("/api"): bucket = "api" if bucket: limit, window = RATE_RULES[bucket] # Les routes d'auth sont anonymes : on ne peut s'appuyer que sur l'IP. key = f"ip:{client_ip(request)}" if bucket == "auth" else _identity(request) ok, retry = limiter.check(f"{bucket}:{key}", limit, window) if not ok: audit( None, "rate_limit", target=path, detail=f"seuil {bucket} dépassé", ip=client_ip(request), ) return JSONResponse( {"detail": "Trop de requêtes. Réessayez dans un instant."}, status_code=429, headers={"Retry-After": str(retry)}, ) return await call_next(request) # --- Nettoyage des entrées ------------------------------------------------- # Caractères de contrôle et marques de direction (usurpation visuelle). _CONTROL = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f​-‏‪-‮⁦-⁩]") def clean_text(value: str, max_length: int) -> str: """Supprime les caractères de contrôle, normalise les espaces, tronque.""" if not value: return "" cleaned = _CONTROL.sub("", str(value)) cleaned = re.sub(r"[ \t]{4,}", " ", cleaned) cleaned = re.sub(r"\n{4,}", "\n\n\n", cleaned) return cleaned.strip()[:max_length] # --- Journal d'audit ------------------------------------------------------- # Nombre d'entrées conservées : le journal ne doit pas grignoter le quota disque. AUDIT_KEEP = 5000 def audit( actor: str | None, action: str, target: str = "", detail: str = "", ip: str = "", ) -> None: """Trace un événement sensible. Ne doit jamais faire échouer l'appelant.""" try: db.execute( "INSERT INTO audit_log(at, actor, action, target, detail, ip) VALUES(?, ?, ?, ?, ?, ?)", (db.now_ms(), actor, action, target[:120], detail[:240], ip[:45]), ) except Exception: log.warning("Écriture du journal d'audit impossible (action=%s)", action) def trim_audit() -> int: """Ne conserve que les AUDIT_KEEP entrées les plus récentes.""" cur = db.execute( "DELETE FROM audit_log WHERE id <= COALESCE(" " (SELECT id FROM audit_log ORDER BY id DESC LIMIT 1 OFFSET ?), 0)", (AUDIT_KEEP,), ) return cur.rowcount or 0 def recent_audit(limit: int = 60) -> list[dict]: rows = db.query("SELECT * FROM audit_log ORDER BY id DESC LIMIT ?", (limit,)) return [ { "id": r["id"], "at": r["at"], "actor": r["actor"], "action": r["action"], "target": r["target"], "detail": r["detail"], "ip": r["ip"], } for r in rows ]