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_rate_limit_key_and_prune():
    """A33 : XFF ignoré depuis une IP publique (anti-bypass), épurage du store."""
    from types import SimpleNamespace

    from app.middleware.security import RateLimitMiddleware

    mw = RateLimitMiddleware(None)

    def req(host, fwd=None):
        headers = {"x-forwarded-for": fwd} if fwd else {}
        return SimpleNamespace(headers=headers, client=SimpleNamespace(host=host))

    # IP publique (globale) : le client peut spoofer XFF autant qu'il veut → clé d'origine
    assert mw._client_key(req("8.8.8.8", "1.2.3.4")) == "8.8.8.8"
    # Derrière un proxy local : on prend le premier hop XFF
    assert mw._client_key(req("10.0.0.1", "198.51.100.7, 10.0.0.2")) == "198.51.100.7"
    # Sans XFF / hôte non IP (testserver)
    assert mw._client_key(req("testserver")) == "testserver"

    # Épurage : les fenêtres expirées sortent du store
    import time

    now = time.time()
    mw._store["old"] = (now - 3600, 5)
    mw._store["fresh"] = (now, 1)
    mw._prune(now)
    assert "old" not in mw._store and "fresh" in mw._store


def test_cors_no_star(client):
    """A37 : plus de `*` — origine refusée n'a pas d'ACAO, origine autorisée oui."""
    r = client.get("/api/health", headers={"Origin": "https://evil.example"})
    assert "access-control-allow-origin" not in r.headers
    r2 = client.get("/api/health", headers={"Origin": "http://localhost:8080"})
    assert r2.headers.get("access-control-allow-origin") == "http://localhost:8080"


def test_asset_version_single_source():
    """A40 : une seule source de version d'assets = le fichier VERSION."""
    import re as _re
    from pathlib import Path

    root = Path(__file__).resolve().parents[1]
    version = (root / "VERSION").read_text(encoding="utf-8").strip()
    from app.templating import ASSET_VERSION, ENV

    assert ASSET_VERSION == version
    assert ENV.globals["asset_version"] == version
    src = (root / "app/templates/base.html").read_text(encoding="utf-8")
    assert "app.css?v={{ asset_version }}" in src
    assert "app.js?v={{ asset_version }}" in src
    # plus aucun littéral de version première main dans les templates
    literals = _re.findall(
        r"(?:app|design-tokens|components|offline|flowdeck)\.(?:css|js)\?v=\d", src
    )
    assert literals == [], literals


def test_publish_service_shared_and_safe(client):
    """A29 : les 3 routers déléguent — 404 sur page absente, slug unique,
    dépublication qui ne touche pas aux partages manuels."""
    from app.db import get_conn

    with get_conn() as conn:
        cur = conn.execute(
            "INSERT INTO pages (workspace, title, content, content_format, share_mode) "
            "VALUES (1, 'Publie moi', 'contenu', 'markdown', 'anyone')",
        )
        pid = cur.lastrowid
        conn.commit()
    try:
        r = client.post(f"/api/pages/{pid}/publish")
        assert r.status_code == 200, r.text
        slug = r.json()["publish_slug"]
        assert slug  # slugify du titre
        with get_conn() as conn:
            row = conn.execute(
                "SELECT is_published, publish_slug, share_mode FROM pages WHERE id=?", (pid,)
            ).fetchone()
        assert row["is_published"] == 1 and row["publish_slug"] == slug
        assert row["share_mode"] == "anyone"  # intouché (share dialog propriétaire)

        r2 = client.delete(f"/api/pages/{pid}/publish")
        assert r2.status_code == 200
        with get_conn() as conn:
            row = conn.execute(
                "SELECT is_published, publish_slug, share_mode FROM pages WHERE id=?", (pid,)
            ).fetchone()
        assert row["is_published"] == 0 and row["publish_slug"] == ""
        assert row["share_mode"] == "anyone"  # dépublier ne révoque pas le partage

        # 404 sur page inexistante — les deux chemins passent par le service
        assert client.post("/api/pages/999999/publish").status_code == 404
        assert client.delete("/api/pages/999999/publish").status_code == 404
    finally:
        with get_conn() as conn:
            conn.execute("DELETE FROM pages WHERE id=?", (pid,))
            conn.commit()


def test_users_me_no_secret_columns(client):
    """A29-byproduct : GET /api/users/me (v1) ne doit plus renvoyer password_hash."""
    r = client.get("/api/users/me")
    assert r.status_code == 200, r.text
    body = r.json()
    assert "password_hash" not in body, list(body)
    assert "locked_until" not in body and "login_attempts" not in body
    assert body.get("login")  # la réponse reste exploitable


def test_gitea_cache_evicts_expired():
    """A42 : les entrées expirées sortent du cache à chaque écriture."""
    from datetime import datetime, timedelta

    from app.services.gitea_client import GiteaClient

    c = GiteaClient.__new__(GiteaClient)  # sans appel réseau
    c._cache = {}
    c._ttl = timedelta(seconds=1)
    c._set_cache("k", "v")
    assert c._cached("k") == "v"
    # expire l'entrée puis force une autre écriture → la précédente est évacuée
    c._cache["k"] = (datetime.now() - timedelta(seconds=1), "v")
    c._set_cache("k2", "v2")
    assert "k" not in c._cache and c._cache["k2"][1] == "v2"


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