"""FlowDeck — Security middleware: CSP headers + rate limiting.""" from __future__ import annotations import ipaddress import re 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'; " # A20 phase 2 : chart.js/leaflet vendorisés dans /static/js/vendor # (test_csp_no_cdn_and_vendor), plus aucun hôte CDN tiers. # `unsafe-eval` reste : Alpine (x-data) + htmx (hx-on/hx-vars) en # ont besoin → retrait avec le build CSP d'Alpine (reste d'A20). "script-src 'self' 'unsafe-eval' 'nonce-{nonce}'; " "script-src-attr 'unsafe-inline'; " # ponytail: aucun @font-face Google (grep négatif) → les deux # hôtes fonts étaient morts, supprimés. "style-src 'self' 'unsafe-inline'; " # ponytail: `https:` reste ouvert — unfurls (YouTube/Vimeo/…) et # tuiles OSM sont inénumérables ; plafond assumé, à resserrer si # un proxy d'images local arrive. "img-src 'self' data: blob: https:; " "font-src 'self' data:; " # connect-src fermé : plus de `https:` (aucun fetch cross-origin # côté front — grep négatif) et websockets scopés à l'hôte de la # requête ({host}) → plus de canal d'exfil vers un tiers. "connect-src 'self' ws://{host} wss://{host}; " "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: # Host du navigateur (uvicorn rejette les Host invalides) ; # on retire quand même tout caractère hors base URL par sécurité. host = re.sub(r"[^0-9A-Za-z.\-:\[\]]", "", request.headers.get("host", "")) response.headers[self.CSP_HEADER] = self.CSP_VALUE.format( nonce=nonce, host=host ) 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//auth et spam de /f/) — 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