fix(sprint-f1): sécurité — path traversal, SSRF, CORS, validation inputs

Sprint F1 — fermeture de toutes les vulnérabilités identifiées :

- Path traversal profiles : validation regex profile_id [a-z0-9_-]
- Path traversal frontend : resolve() + vérification sous _STATIC_DIR
- Path traversal slug : Field(pattern=r'^[a-z0-9][a-z0-9_-]{0,63}$')
- Path traversal folio_label : sanitisation _sanitize_label()
- Upload fichiers : _sanitize_filename(), limite 100Mo, validation MIME
- SSRF manifest_url : blocklist IP privées, schéma http(s) uniquement,
limite taille manifest 10Mo
- CORS : allow_credentials=False (incompatible avec allow_origins=["*"])
- Inputs : max_length sur search (500), corrections (500k/100k),
model_id (256), restore_to_version >= 1

17 tests de sécurité ajoutés (test_security.py).
477 tests passants, 0 échecs.

https://claude.ai/code/session_015Lht7wNQRzhUaLw94dE9z9

backend/app/api/v1/corpora.py

@@ -15,7 +15,7 @@ from datetime import datetime, timezone
 
 # 2. third-party
 from fastapi import APIRouter, Depends, HTTPException
-from pydantic import BaseModel, ConfigDict
+from pydantic import BaseModel, ConfigDict, Field
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -29,9 +29,9 @@ router = APIRouter(prefix="/corpora", tags=["corpora"])
 # ── Schémas de requête / réponse ─────────────────────────────────────────────
 
 class CorpusCreate(BaseModel):
-    slug: str
-    title: str
-    profile_id: str
+    slug: str = Field(..., pattern=r"^[a-z0-9][a-z0-9_-]{0,63}$")
+    title: str = Field(..., min_length=1, max_length=256)
+    profile_id: str = Field(..., pattern=r"^[a-z0-9][a-z0-9_-]*$")
 
 
 class CorpusResponse(BaseModel):

backend/app/api/v1/ingest.py

@@ -11,13 +11,14 @@ Règle : ingestion = création des PageModel en BDD uniquement.
 """
 # 1. stdlib
 import logging
+import re
 import uuid
 from pathlib import Path
 
 # 2. third-party
 import httpx
 from fastapi import APIRouter, Depends, File, HTTPException, UploadFile
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -30,6 +31,28 @@ logger = logging.getLogger(__name__)
 
 router = APIRouter(tags=["ingestion"])
 
+# ── Constantes de sécurité ────────────────────────────────────────────────────
+
+_SAFE_LABEL_RE = re.compile(r"^[\w\-\.]+$")
+_MAX_UPLOAD_BYTES = 100 * 1024 * 1024  # 100 Mo par fichier
+_ALLOWED_MIME_PREFIXES = ("image/",)
+
+
+def _sanitize_label(label: str) -> str:
+    """Nettoie un folio_label : garde uniquement alphanum, -, _, ."""
+    clean = Path(label).name  # retire tout chemin
+    if not _SAFE_LABEL_RE.match(clean) or not clean:
+        clean = re.sub(r"[^\w\-\.]", "_", clean) or "page"
+    return clean
+
+
+def _sanitize_filename(name: str) -> str:
+    """Nettoie un nom de fichier uploadé : garde uniquement le basename sûr."""
+    clean = Path(name).name
+    if not _SAFE_LABEL_RE.match(clean) or not clean:
+        clean = f"{uuid.uuid4().hex[:12]}.bin"
+    return clean
+
 
 # ── Schémas ───────────────────────────────────────────────────────────────────
 
@@ -38,8 +61,8 @@ class IIIFManifestRequest(BaseModel):
 
 
 class IIIFImagesRequest(BaseModel):
-    urls: list[str]
-    folio_labels: list[str]
+    urls: list[str] = Field(..., max_length=5000)
+    folio_labels: list[str] = Field(..., max_length=5000)
 
 
 class IngestResponse(BaseModel):
@@ -144,11 +167,31 @@ _MANIFEST_HEADERS = {
 }
 
 
+_MAX_MANIFEST_BYTES = 10 * 1024 * 1024  # 10 Mo max pour un manifest JSON
+
+
+def _validate_url(url: str) -> None:
+    """Rejette les URLs non-HTTP et les cibles réseau privé (SSRF)."""
+    from urllib.parse import urlparse
+
+    parsed = urlparse(url)
+    if parsed.scheme not in ("http", "https"):
+        raise ValueError(f"Schéma non autorisé : {parsed.scheme!r}")
+    host = (parsed.hostname or "").lower()
+    # Bloquer les adresses privées / locales
+    blocked = ("localhost", "127.0.0.1", "0.0.0.0", "[::1]", "metadata.google.internal")
+    if host in blocked or host.startswith("169.254.") or host.startswith("10.") or host.startswith("192.168."):
+        raise ValueError(f"Hôte interdit : {host}")
+
+
 async def _fetch_json_manifest(url: str) -> dict:
-    """Télécharge un manifest IIIF. Fonction isolée pour faciliter les tests."""
+    """Télécharge un manifest IIIF avec protections SSRF + taille max."""
+    _validate_url(url)
     async with httpx.AsyncClient() as client:
         resp = await client.get(url, headers=_MANIFEST_HEADERS, follow_redirects=True, timeout=30.0)
         resp.raise_for_status()
+        if len(resp.content) > _MAX_MANIFEST_BYTES:
+            raise ValueError(f"Manifest trop volumineux ({len(resp.content)} octets)")
         return resp.json()
 
 
@@ -202,16 +245,32 @@ async def ingest_files(
     seq = await _next_sequence(db, ms.id)
 
     # Collect labels and detect duplicates
-    labels = [Path(f.filename or f"file_{i}").stem for i, f in enumerate(files)]
+    labels = [_sanitize_label(Path(f.filename or f"file_{i}").stem) for i, f in enumerate(files)]
     dupes = _find_duplicate_labels(labels)
 
     created: list[PageModel] = []
     skipped = 0
     for i, upload in enumerate(files):
-        filename = Path(upload.filename or f"file_{i}").name
+        # Validation MIME type
+        ctype = upload.content_type or ""
+        if not any(ctype.startswith(p) for p in _ALLOWED_MIME_PREFIXES):
+            raise HTTPException(
+                status_code=422,
+                detail=f"Type MIME non autorisé : {ctype!r}. Seules les images sont acceptées.",
+            )
+
+        filename = _sanitize_filename(upload.filename or f"file_{i}.bin")
         folio_label = labels[i]
         page_id = _make_page_id(corpus.slug, folio_label, seq + i, dupes)
 
+        content = await upload.read()
+        # Validation taille
+        if len(content) > _MAX_UPLOAD_BYTES:
+            raise HTTPException(
+                status_code=413,
+                detail=f"Fichier trop volumineux ({len(content)} octets). Maximum : {_MAX_UPLOAD_BYTES}.",
+            )
+
         master_dir = (
             _config_module.settings.data_dir
             / "corpora"
@@ -221,7 +280,6 @@ async def ingest_files(
         )
         master_dir.mkdir(parents=True, exist_ok=True)
         master_path = master_dir / filename
-        content = await upload.read()
         master_path.write_bytes(content)
 
         page = await _create_page(
@@ -260,6 +318,8 @@ async def ingest_iiif_manifest(
 
     try:
         manifest = await _fetch_json_manifest(body.manifest_url)
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc))
     except httpx.HTTPStatusError as exc:
         raise HTTPException(
             status_code=502,
@@ -302,7 +362,7 @@ async def ingest_iiif_manifest(
     seq = await _next_sequence(db, ms.id)
 
     # Collect labels and detect duplicates
-    labels = [_extract_canvas_label(canvas, i) for i, canvas in enumerate(canvases)]
+    labels = [_sanitize_label(_extract_canvas_label(canvas, i)) for i, canvas in enumerate(canvases)]
     dupes = _find_duplicate_labels(labels)
 
     created: list[PageModel] = []
@@ -358,11 +418,12 @@ async def ingest_iiif_images(
     ms = await _get_or_create_manuscript(db, corpus_id)
     seq = await _next_sequence(db, ms.id)
 
-    dupes = _find_duplicate_labels(body.folio_labels)
+    sanitized_labels = [_sanitize_label(lbl) for lbl in body.folio_labels]
+    dupes = _find_duplicate_labels(sanitized_labels)
 
     created: list[PageModel] = []
     skipped = 0
-    for i, (url, folio_label) in enumerate(zip(body.urls, body.folio_labels)):
+    for i, (url, folio_label) in enumerate(zip(body.urls, sanitized_labels)):
         page_id = _make_page_id(corpus.slug, folio_label, seq + i, dupes)
         page = await _create_page(
             db, ms.id, page_id, folio_label, seq + i,

backend/app/api/v1/models_api.py

@@ -17,7 +17,7 @@ from datetime import datetime, timezone
 
 # 2. third-party
 from fastapi import APIRouter, Depends, HTTPException
-from pydantic import BaseModel, ConfigDict
+from pydantic import BaseModel, ConfigDict, Field
 from sqlalchemy.ext.asyncio import AsyncSession
 
 # 3. local
@@ -42,9 +42,9 @@ class ProviderInfo(BaseModel):
 
 
 class ModelSelectRequest(BaseModel):
-    model_id: str
-    provider_type: str
-    display_name: str = ""
+    model_id: str = Field(..., min_length=1, max_length=256)
+    provider_type: str = Field(..., min_length=1, max_length=64)
+    display_name: str = Field("", max_length=256)
 
 
 class ModelConfigResponse(BaseModel):

backend/app/api/v1/pages.py

@@ -18,7 +18,7 @@ from typing import Any
 
 # 2. third-party
 from fastapi import APIRouter, Depends, HTTPException
-from pydantic import BaseModel, ConfigDict, ValidationError
+from pydantic import BaseModel, ConfigDict, Field, ValidationError
 from sqlalchemy.ext.asyncio import AsyncSession
 
 # 3. local
@@ -42,12 +42,12 @@ class CorrectionsRequest(BaseModel):
     indiquée est restaurée (avec incrémentation de editorial.version).
     """
 
-    ocr_diplomatic_text: str | None = None
-    editorial_status: str | None = None
-    commentary_public: str | None = None
-    commentary_scholarly: str | None = None
+    ocr_diplomatic_text: str | None = Field(None, max_length=500_000)
+    editorial_status: str | None = Field(None, max_length=50)
+    commentary_public: str | None = Field(None, max_length=100_000)
+    commentary_scholarly: str | None = Field(None, max_length=100_000)
     region_validations: dict[str, str] | None = None
-    restore_to_version: int | None = None
+    restore_to_version: int | None = Field(None, ge=1)
 
 
 class VersionInfo(BaseModel):

backend/app/api/v1/profiles.py

@@ -10,6 +10,7 @@ Ils sont validés par CorpusProfile avant d'être retournés.
 # 1. stdlib
 import json
 import logging
+import re
 from pathlib import Path
 
 # 2. third-party
@@ -57,9 +58,14 @@ async def list_profiles() -> list[dict]:
     return profiles
 
 
+_SAFE_ID_RE = re.compile(r"^[a-z0-9][a-z0-9_-]*$")
+
+
 @router.get("/{profile_id}", response_model=dict)
 async def get_profile(profile_id: str) -> dict:
     """Retourne un profil par son id (nom du fichier sans extension)."""
+    if not _SAFE_ID_RE.match(profile_id):
+        raise HTTPException(status_code=400, detail="profile_id invalide")
     path = settings.profiles_dir / f"{profile_id}.json"
     if not path.exists():
         raise HTTPException(status_code=404, detail="Profil introuvable")

backend/app/api/v1/search.py

@@ -95,7 +95,7 @@ def _score_master(data: dict, query_normalized: str) -> tuple[int, str]:
 
 @router.get("/search", response_model=list[SearchResult])
 async def search_pages(
-    q: str = Query(..., min_length=2, description="Requête de recherche (min. 2 caractères)"),
+    q: str = Query(..., min_length=2, max_length=500, description="Requête de recherche (2–500 caractères)"),
 ) -> list[SearchResult]:
     """Recherche plein texte dans les master.json de tous les corpus.
 

backend/app/main.py

@@ -65,11 +65,11 @@ app = FastAPI(
     lifespan=lifespan,
 )
 
-# ── CORS (dev : tous les origines autorisés) ──────────────────────────────────
+# ── CORS (dev : toutes les origines autorisées, sans credentials) ──────────────
 app.add_middleware(
     CORSMiddleware,
     allow_origins=["*"],
-    allow_credentials=True,
+    allow_credentials=False,
     allow_methods=["*"],
     allow_headers=["*"],
 )
@@ -97,8 +97,9 @@ async def serve_frontend(full_path: str) -> FileResponse | RedirectResponse:
     if full_path.startswith("api/"):
         raise HTTPException(status_code=404, detail=f"Endpoint not found: /{full_path}")
     if _STATIC_DIR.is_dir():
-        candidate = _STATIC_DIR / full_path
-        if candidate.is_file():
+        candidate = (_STATIC_DIR / full_path).resolve()
+        # Empêcher le path traversal : le fichier résolu doit être sous _STATIC_DIR
+        if candidate.is_file() and str(candidate).startswith(str(_STATIC_DIR.resolve())):
             return FileResponse(candidate)
         index = _STATIC_DIR / "index.html"
         if index.exists():

backend/tests/test_security.py

@@ -0,0 +1,215 @@
+"""
+Tests de sécurité — Sprint F1.
+
+Vérifie que toutes les vulnérabilités identifiées sont corrigées :
+- Path traversal sur profiles, slug, folio_label, frontend serving
+- SSRF sur manifest_url
+- Validation des entrées (taille, format)
+"""
+# 1. stdlib
+import pytest
+
+# 2. third-party — fixtures API
+from tests.conftest_api import async_client, db_session  # noqa: F401
+
+
+# ---------------------------------------------------------------------------
+# Path traversal — profiles
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_profile_path_traversal_dotdot(async_client):
+    """Un profile_id contenant '..' doit être rejeté (400)."""
+    resp = await async_client.get("/api/v1/profiles/..passwd")
+    assert resp.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_profile_path_traversal_slash(async_client):
+    """Un profile_id avec un slash (même encodé) doit être rejeté (400 ou 404)."""
+    # FastAPI normalise les chemins, donc un slash dans l'ID ne sera pas transmis.
+    # On teste avec un ID contenant des caractères spéciaux interdits.
+    resp = await async_client.get("/api/v1/profiles/UPPER_CASE")
+    assert resp.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_profile_path_traversal_special_chars(async_client):
+    """Un profile_id avec des caractères spéciaux doit être rejeté."""
+    resp = await async_client.get("/api/v1/profiles/test@profile")
+    assert resp.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_profile_valid_id_not_found(async_client):
+    """Un profile_id valide mais inexistant retourne 404 (pas 400)."""
+    resp = await async_client.get("/api/v1/profiles/does-not-exist")
+    assert resp.status_code == 404
+
+
+# ---------------------------------------------------------------------------
+# Path traversal — corpus slug
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_corpus_slug_path_traversal(async_client):
+    """Un slug avec ../ doit être rejeté par la validation Pydantic."""
+    resp = await async_client.post("/api/v1/corpora", json={
+        "slug": "../../malicious",
+        "title": "Test",
+        "profile_id": "medieval-illuminated",
+    })
+    assert resp.status_code == 422
+
+
+@pytest.mark.asyncio
+async def test_corpus_slug_with_spaces(async_client):
+    """Un slug avec des espaces doit être rejeté."""
+    resp = await async_client.post("/api/v1/corpora", json={
+        "slug": "my corpus",
+        "title": "Test",
+        "profile_id": "medieval-illuminated",
+    })
+    assert resp.status_code == 422
+
+
+@pytest.mark.asyncio
+async def test_corpus_slug_uppercase(async_client):
+    """Un slug avec des majuscules doit être rejeté (lowercase only)."""
+    resp = await async_client.post("/api/v1/corpora", json={
+        "slug": "MyCorpus",
+        "title": "Test",
+        "profile_id": "medieval-illuminated",
+    })
+    assert resp.status_code == 422
+
+
+@pytest.mark.asyncio
+async def test_corpus_slug_valid(async_client):
+    """Un slug valide doit être accepté."""
+    resp = await async_client.post("/api/v1/corpora", json={
+        "slug": "my-corpus-01",
+        "title": "Test",
+        "profile_id": "medieval-illuminated",
+    })
+    assert resp.status_code == 201
+
+
+@pytest.mark.asyncio
+async def test_corpus_slug_empty(async_client):
+    """Un slug vide doit être rejeté."""
+    resp = await async_client.post("/api/v1/corpora", json={
+        "slug": "",
+        "title": "Test",
+        "profile_id": "medieval-illuminated",
+    })
+    assert resp.status_code == 422
+
+
+@pytest.mark.asyncio
+async def test_corpus_title_too_long(async_client):
+    """Un titre trop long (>256 chars) doit être rejeté."""
+    resp = await async_client.post("/api/v1/corpora", json={
+        "slug": "test-long",
+        "title": "x" * 300,
+        "profile_id": "medieval-illuminated",
+    })
+    assert resp.status_code == 422
+
+
+# ---------------------------------------------------------------------------
+# SSRF — manifest_url
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_ssrf_localhost(async_client):
+    """Un manifest_url pointant vers localhost doit être rejeté."""
+    # Créer un corpus d'abord
+    create = await async_client.post("/api/v1/corpora", json={
+        "slug": "ssrf-test", "title": "SSRF", "profile_id": "test",
+    })
+    cid = create.json()["id"]
+
+    resp = await async_client.post(f"/api/v1/corpora/{cid}/ingest/iiif-manifest", json={
+        "manifest_url": "http://localhost:8000/secret",
+    })
+    assert resp.status_code == 400
+    assert "interdit" in resp.json()["detail"].lower() or "localhost" in resp.json()["detail"].lower()
+
+
+@pytest.mark.asyncio
+async def test_ssrf_metadata_ip(async_client):
+    """Un manifest_url vers 169.254.x.x (cloud metadata) doit être rejeté."""
+    create = await async_client.post("/api/v1/corpora", json={
+        "slug": "ssrf-meta", "title": "SSRF", "profile_id": "test",
+    })
+    cid = create.json()["id"]
+
+    resp = await async_client.post(f"/api/v1/corpora/{cid}/ingest/iiif-manifest", json={
+        "manifest_url": "http://169.254.169.254/latest/meta-data/",
+    })
+    assert resp.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_ssrf_file_scheme(async_client):
+    """Un manifest_url avec file:// doit être rejeté."""
+    create = await async_client.post("/api/v1/corpora", json={
+        "slug": "ssrf-file", "title": "SSRF", "profile_id": "test",
+    })
+    cid = create.json()["id"]
+
+    resp = await async_client.post(f"/api/v1/corpora/{cid}/ingest/iiif-manifest", json={
+        "manifest_url": "file:///etc/passwd",
+    })
+    assert resp.status_code == 400
+
+
+# ---------------------------------------------------------------------------
+# Input validation — search
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_search_query_too_long(async_client):
+    """Une requête de recherche >500 chars doit être rejetée."""
+    resp = await async_client.get("/api/v1/search", params={"q": "x" * 501})
+    assert resp.status_code == 422
+
+
+@pytest.mark.asyncio
+async def test_search_query_max_length_ok(async_client):
+    """Une requête de recherche de 500 chars doit être acceptée (0 résultat)."""
+    resp = await async_client.get("/api/v1/search", params={"q": "x" * 500})
+    assert resp.status_code == 200
+
+
+# ---------------------------------------------------------------------------
+# Input validation — model selection
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_model_id_too_long(async_client):
+    """Un model_id >256 chars doit être rejeté."""
+    create = await async_client.post("/api/v1/corpora", json={
+        "slug": "model-test", "title": "T", "profile_id": "test",
+    })
+    cid = create.json()["id"]
+
+    resp = await async_client.put(f"/api/v1/corpora/{cid}/model", json={
+        "model_id": "x" * 300,
+        "provider_type": "google_ai_studio",
+    })
+    assert resp.status_code == 422
+
+
+# ---------------------------------------------------------------------------
+# Input validation — corrections
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_corrections_restore_negative_version(async_client):
+    """restore_to_version < 1 doit être rejeté."""
+    resp = await async_client.post("/api/v1/pages/fake-page/corrections", json={
+        "restore_to_version": 0,
+    })
+    assert resp.status_code == 422