app/security.py

"""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
    ]