flowdeck/tests/test_audit_p0_fixes.py

"""Non-régression de l'audit sécurité 2026-09-30 — A11 (traversal) et A18 (XSS public)."""
from conftest import anon, anon_csrf


def test_avatar_path_traversal_denied(client):
    """A11 : `:path` accepte les `/` — la lecture doit rester dans /data/avatars."""
    r = client.get("/api/settings/avatar/..%2f..%2fetc%2fpasswd")
    assert r.status_code in (403, 404), r.status_code


def test_public_view_escapes_output(client):
    """A18 : titre de base et titre de ligne interpolés dans un f-string HTML."""
    cid = client.post("/db/api", json={"name": "<script>alert(1)</script>"}).json()["id"]
    client.post(f"/db/{cid}/pages/api", json={"title": "<img src=x onerror=alert(1)>"})

    anon(client)
    r = client.get(f"/workspace/public/{cid}")
    assert r.status_code == 200
    assert "<script>alert(1)" not in r.text
    assert "&lt;script&gt;" in r.text
    assert "<img src=x" not in r.text


def test_public_view_hides_restricted_collection(client):
    """A18 : `permission_type` restricted/private → 404 (pas de fuite)."""
    cid = client.post("/db/api", json={"name": "Internal"}).json()["id"]
    from app.db import get_conn

    with get_conn() as conn:
        conn.execute("UPDATE collections SET permission_type='restricted' WHERE id=?", (cid,))
        conn.commit()

    anon(client)
    r = client.get(f"/workspace/public/{cid}")
    assert r.status_code == 404
    assert "Internal" not in r.text


def test_no_duplicate_routes():
    """A24 : deux routes même méthode+chemin → l'une écrase silencieusement l'autre."""
    from app.main import app

    seen = set()
    for route in app.routes:
        for method in getattr(route, "methods", None) or set():
            if method in ("HEAD", "OPTIONS"):
                continue
            key = (method, route.path)
            assert key not in seen, f"doublon de route: {key}"
            seen.add(key)


def test_og_metadata_rejects_private_host(client):
    """A12 : SSRF — aucun fetch vers loopback/link-local (re-vérif à chaque hop)."""
    for url in ("http://127.0.0.1/latest/meta-data/", "http://169.254.169.254/x", "http://localhost/x"):
        r = client.post("/board/api/og/metadata", json={"url": url})
        assert r.status_code == 400, (url, r.status_code, r.text[:200])


def test_automations_require_session(client):
    """A13 : CRUD, run et press-button refusent un anonymous."""
    anon(client)
    anon_csrf(client)
    assert client.post("/workspace/automations", json={"name": "x"}).status_code == 401
    assert client.post("/workspace/automations/1/run", json={}).status_code == 401
    assert client.post("/api/automations/press-button", json={}).status_code == 401
    assert client.get("/workspace/automations").status_code == 401


def test_outbound_webhook_requires_admin_and_public_url(client):
    """A15 : webhooks sortants = admin + URL publique (le scheduler POSTe le contenu)."""
    # admin de la fixture : URL privée refusée (SSRF)
    r = client.post("/workspace/webhooks", json={"url": "http://127.0.0.1/hook"})
    assert r.status_code == 400

    anon(client)
    anon_csrf(client)
    assert client.post("/workspace/webhooks", json={"url": "https://example.com/h"}).status_code == 401


def test_legacy_api_requires_auth(client):
    """A17 : le router /api legacy refuse un anonymous (health et front-error restent publics)."""
    anon(client)
    assert client.get("/api/users/me").status_code == 401
    # CSRF valide mais aucune session → la garde du router doit répondre 401.
    client.cookies.set("csrf_token", "csrf-anon")
    assert client.post("/api/move", json={}, headers={"X-CSRF-Token": "csrf-anon"}).status_code == 401
    assert client.get("/api/health").status_code == 200


def test_upload_requires_session_and_validates_files(client):
    """A22 : validate_upload branché (taille + extension) et pas d'upload anonyme."""
    from app.middleware.security import validate_upload

    assert validate_upload("note.txt", 10) is None
    assert validate_upload("virus.exe", 10) is not None
    assert validate_upload("big.txt", 11 * 1024 * 1024) is not None

    anon_csrf(client)
    r = client.post("/api/local-workspace/upload", files={"files": ("a.txt", b"x", "text/plain")})
    assert r.status_code == 401


def test_agent_providers_require_admin_and_valid_api_base(client):
    """A14 : plus de fallback `admin` — un anonymous ne dirige plus le ping."""
    # admin de la fixture : scheme non-http refusé, identifiants refusés
    r = client.patch("/api/agent/providers", json={"provider": "mistral", "api_base": "ftp://x.test/v1"})
    assert r.status_code == 400, r.text
    r2 = client.post("/api/agent/providers/test", json={"provider": "mistral", "api_base": "https://user:[email protected]/v1"})
    assert r2.status_code == 400, r2.text

    anon_csrf(client)
    assert client.patch("/api/agent/providers", json={"provider": "ollama"}).status_code == 401
    assert client.post("/api/agent/providers/test", json={"provider": "ollama"}).status_code == 401


def test_collection_rollback_when_materialize_fails(client, monkeypatch):
    """A25 : un échec de `materialize_properties` ne doit pas commiter la collection."""
    import pytest

    from app.services import db_templates

    def boom(*_a, **_k):
        raise RuntimeError("materialize boom")

    monkeypatch.setattr(db_templates, "materialize_properties", boom)
    tok = client.post("/api/v1/token").json()["token"]
    with pytest.raises(RuntimeError):
        client.post(
            "/api/v2/collections",
            json={"name": "Broken", "schema": [{"name": "Title", "type": "title"}]},
            headers={"Authorization": f"Bearer {tok}"},
        )

    from app.db import get_conn

    with get_conn() as conn:
        n = conn.execute("SELECT COUNT(*) FROM collections WHERE name='Broken'").fetchone()[0]
    assert n == 0, "la collection ne doit pas survivre à un schéma non matérialisé"


def test_exports_and_attachments_require_auth(client):
    """A16 : export + pièces jointes = session et `can_view_page` (jamais le contenu)."""
    pid = client.post("/board/api/pages?title=Secret&section=Private").json()["id"]

    anon(client)
    assert client.get(f"/api/export/markdown/{pid}").status_code == 401
    assert client.get(f"/api/export/html/{pid}").status_code == 401
    assert client.get(f"/api/pages/{pid}/download").status_code == 401
    assert client.get(f"/api/pages/{pid}/file-content").status_code == 401