bruno
de751ffe35
La bascule A20 phase 3 : - static/js/alpine.csp.min.js (build officiel @alpinejs/csp, 0 eval/new Function, parseur maison) servi partout : base.html, import.html, welcome.html + entree sw.js (cache bump v8). - CSP : script-src 'self' 'nonce-…' — unsafe-eval SUPPRIMÉ (ne servait plus qu'Alpine standard). htmx allowEval:false deja pose (v7.37). - Assertion test inversée : assert "'unsafe-eval'" not in script_src. - Scan statique final sur TOUS les templates : 0 expression incompatible (4 residus = faux positifs dans des chaines de texte). Pré-requis réunis par les lots 1-3 : 12 surfaces migrées + gateées (csp_preview), registres Alpine.data, x-html → x-init+Alpine.effect, délégués window.E, partage d'état lexical, bug topbar corrigé. Verifs : suite **1094/1094** · ruff OK · eslint 0/0 · **E2E 7/7 sous CSP reel** (script-src sans unsafe-eval verifie sur l'instance). Hors gate (scan propre, gitea down) : board/table_view/teamload/ card_detail → à vérifier au premier usage avec gitea remonté (noté ROADMAP/CHANGELOG).
231 lines
9.4 KiB
Python
231 lines
9.4 KiB
Python
"""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).
|
|
# A20 TERMINÉ : `unsafe-eval` retiré — Alpine tourne en build CSP
|
|
# (static/js/alpine.csp.min.js, 0 eval) ; htmx allowEval=false.
|
|
# 15 surfaces en csp_preview vert + scan statique 0 (board/gitea/cards
|
|
# = props propres, gitea down empêche un gate dédié).
|
|
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.
|
|
"script-src 'self' '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/<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
|