flowdeck/app/middleware/security.py

"""FlowDeck — Security middleware: CSP headers + rate limiting."""
from __future__ import annotations

import ipaddress
import secrets
import time
from collections import defaultdict

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse

from app.templating import CSP_NONCE

# ── Constants ────────────────────────────────────────────────

# Allowed extensions for file uploads
ALLOWED_EXTENSIONS: frozenset[str] = frozenset({
    # Images
    ".png", ".jpg", ".jpeg", ".gif", ".webp", ".svg", ".bmp", ".ico",
    # Documents
    ".pdf", ".md", ".markdown", ".txt", ".log", ".csv",
    # Code
    ".py", ".js", ".jsx", ".ts", ".tsx", ".html", ".htm", ".xml", ".css",
    ".json", ".yaml", ".yml", ".toml", ".sql", ".sh", ".bash", ".zsh",
    ".ps1", ".rs", ".go", ".java", ".rb", ".php", ".c", ".cpp", ".h",
    ".swift", ".kt", ".scala", ".r",
    # Archives
    ".zip", ".tar", ".gz", ".rar", ".7z",
    # Misc
    ".env", ".cfg", ".conf", ".ini", ".dockerfile", ".makefile",
})

MAX_UPLOAD_SIZE = 10 * 1024 * 1024  # 10 MB


def validate_upload(filename: str, size: int) -> str | None:
    """Validate upload filename and size. Returns error message or None."""
    if size > MAX_UPLOAD_SIZE:
        return f"File '{filename}' exceeds maximum size of 10 MB"
    ext = _ext(filename)
    if ext and ext not in ALLOWED_EXTENSIONS:
        return f"File extension '{ext}' is not allowed"
    return None


def _ext(filename: str) -> str:
    """Extract lowercase extension from filename."""
    if "." in filename:
        return "." + filename.rsplit(".", 1)[-1].lower()
    return ""


# ── Content-Security-Policy Middleware ────────────────────────

class ContentSecurityPolicyMiddleware(BaseHTTPMiddleware):
    """Sets Content-Security-Policy headers on all HTML responses.

    A permissive-yet-safe policy for a Notion-style app that needs:
    - inline scripts (Alpine.js, HTMX)
    - inline styles
    - font loading
    - images from various sources
    - media (audio/video)
    - websocket connections for HMR/SSE
    """

    CSP_HEADER = "Content-Security-Policy"
    # A20 : `unsafe-inline` sort de script-src (remplacé par un nonce par
    # requête — un XSS injecté dans une page ne tourne plus). Les 74 handlers
    # inline `onclick=` restent couverts par `script-src-attr`, explicitement
    # détaché de script-src (sinon le nonce les désactiverait aussi).
    # `unsafe-eval` reste : Alpine (x-data en string) et htmx (hx-vars) en ont
    # besoin → à retirer avec le build CSP d'Alpine (voir reste d'A20).
    CSP_VALUE = (
        "default-src 'self'; "
        # ponytail: chart.js et leaflet sont chargés depuis CDN par les vues
        # chart/map de collections — l'upgrade est de les vendoriser dans
        # /static/js puis de retirer ces deux hôtes.
        "script-src 'self' 'unsafe-eval' 'nonce-{nonce}' "
        "https://cdn.jsdelivr.net https://unpkg.com; "
        "script-src-attr 'unsafe-inline'; "
        "style-src 'self' 'unsafe-inline' https://fonts.googleapis.com https://unpkg.com; "
        "img-src 'self' data: blob: https:; "
        "font-src 'self' data: https://fonts.gstatic.com; "
        "connect-src 'self' https: wss: ws:; "
        "media-src 'self' blob:; "
        "frame-src 'self'; "
        "object-src 'none'; "
        "base-uri 'self'; "
        "form-action 'self'; "
    )

    async def dispatch(self, request: Request, call_next):
        nonce = secrets.token_urlsafe(16)
        # Posé AVANT call_next : BaseHTTPMiddleware exécute le aval dans une
        # tâche créée maintenant → le contexte (donc le nonce) y est copié,
        # exactement ce que les templates liront via `csp_nonce()`.
        CSP_NONCE.set(nonce)
        response = await call_next(request)
        # Only set CSP on HTML responses
        content_type = response.headers.get("content-type", "")
        if "text/html" in content_type:
            response.headers[self.CSP_HEADER] = self.CSP_VALUE.format(nonce=nonce)
        return response


# ── Rate Limiting Middleware ──────────────────────────────────

class RateLimitMiddleware(BaseHTTPMiddleware):
    """Simple in-memory sliding-window rate limiter per IP.

    Default: 100 requests per minute per IP for API routes.
    Non-API routes (HTML pages, static files) are not rate-limited.
    """

    # Paths that should be rate-limited
    RATE_LIMITED_PREFIXES: tuple[str, ...] = (
        "/api/", "/board/api/", "/auth/",
        # A33 : préfixes qui manquaient — SCIM (brute force de jetons/IdP),
        # API workspace + collections (les endpoints mutants du legacy).
        "/scim/v2/", "/workspace/", "/db/",
    )

    # Pages publiques : seul le non-GET est plafonné (brute force de
    # /s/<slug>/auth et spam de /f/<token>) — la lecture reste libre pour les
    # visiteurs d'un site publié qui partagent une IP.
    RATE_LIMITED_NON_GET_PREFIXES: tuple[str, ...] = ("/s/", "/f/")

    # Paths exempt from rate limiting even under an API prefix
    EXEMPT_PATHS: frozenset[str] = frozenset({
        "/api/health",
        "/api/frontend-error",
        "/api/frontend-errors",
    })

    def __init__(self, app, max_requests: int | None = None, window_seconds: int = 60):
        super().__init__(app)
        # A33 : None = lire `settings.rate_limit_requests` à chaque requête (la
        # valeur de config n'était jamais lue — 100 codé en dur contre 60 annoncé).
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self._store: dict[str, tuple[float, int]] = defaultdict(lambda: (0.0, 0))
        self._last_prune = 0.0
        self._max_keys = 5000

    async def dispatch(self, request: Request, call_next):
        path = request.url.path

        # Respect the global rate-limit toggle (disabled in tests/local).
        from app.config import settings
        if not settings.rate_limit_enabled:
            return await call_next(request)

        # Only rate-limit API routes (+ non-GET sur les pages publiques)
        method = request.method.upper()
        limited = any(path.startswith(p) for p in self.RATE_LIMITED_PREFIXES) or (
            method not in ("GET", "HEAD", "OPTIONS")
            and any(path.startswith(p) for p in self.RATE_LIMITED_NON_GET_PREFIXES)
        )
        if not limited:
            return await call_next(request)

        # Exempt health check and error capture
        if path in self.EXEMPT_PATHS:
            return await call_next(request)

        limit = self.max_requests or settings.rate_limit_requests
        ip = self._client_key(request)
        now = time.time()

        # A33 : le store n'était jamais épuré → croissance mémoire par IP.
        if len(self._store) > self._max_keys and now - self._last_prune > self.window_seconds:
            self._prune(now)

        window_start, count = self._store[ip]
        if now - window_start > self.window_seconds:
            self._store[ip] = (now, 1)
            return await call_next(request)

        if count >= limit:
            return JSONResponse(
                {"error": "Rate limit exceeded", "detail": f"Max {limit} req/min per IP"},
                status_code=429,
            )

        self._store[ip] = (window_start, count + 1)
        return await call_next(request)

    def _client_key(self, request: Request) -> str:
        """A33 : X-Forwarded-For uniquement derrière un proxy local.

        Le test porte sur `is_private` (Python ≥ 3.13 : toutes les plages non
        globales, pas seulement RFC1918) — un pair non-global n'est pas un
        internaute, donc le XFF du proxy fait foi.
        ponytail: si un jour plusieurs proxies se chaînent (ou si l'app est
        exposée directement), prendre la dernière adresse non privée de la
        chaîne plutôt que la première.
        """
        host = request.client.host if request.client else "unknown"
        fwd = request.headers.get("x-forwarded-for", "")
        if fwd:
            try:
                direct = ipaddress.ip_address(host)
                if direct.is_private or direct.is_loopback:
                    return fwd.split(",")[0].strip() or host
            except ValueError:
                pass  # hôte non-IP (testserver…) → on garde la clé d'origine
        return host

    def _prune(self, now: float) -> None:
        expired = [k for k, (start, _) in self._store.items() if now - start > self.window_seconds]
        for k in expired:
            del self._store[k]
        self._last_prune = now