bruno
5a537f5dc3
- A12 — `og_fetcher` : GET sans `follow_redirects`, `_is_public_host` revérifié à chaque saut (max 5) ; `POST /board/api/og/metadata` → 400 sur hôte privé/loopback - A13 — router automations sous `Depends(_require_session)` (CRUD, run, press-button) + `created_by` sans fallback ; action `webhook` validée par `_is_public_host` avant POST (SSRF) - A15 — webhooks sortants : `_require_admin` sur GET/POST/DELETE + `_is_public_host` sur l'URL en création - A17 — router legacy `/api` sous `Depends(_require_session_or_bearer)` (session ou Bearer `/api/v1`), allowlist explicite `/api/health` + `/api/frontend-error` - A22 — les 2 uploads locales : session exigée (`_require_user_id`) + `validate_upload` branché (taille + extension) + `FLOWDECK_DATA_DIR` au lieu de `/data` codé en dur - A23 — N+1 : COUNT→`GROUP BY` (dashboard), cards→`executemany` (board sync), duplicata de propriétés→`executemany` + remap des ids par SELECT (collections) - A24 — 2 routes écrasées supprimées : `GET /api/projects` (api.py) et `GET /workspace` (workspace.py) + test « aucun doublon méthode+chemin » - Tests : +9 dans `tests/test_audit_p0_fixes.py` (SSRF, 401s, validate_upload, doublons de routes) ; tests OG sur hôtes résolubles (la garde fait du DNS) - suite **1025/1025** · `ruff check app tests` OK
100 lines
4.1 KiB
Python
100 lines
4.1 KiB
Python
"""Non-régression de l'audit sécurité 2026-09-30 — A11 (traversal) et A18 (XSS public)."""
|
|
from conftest import anon
|
|
|
|
|
|
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 "<script>" 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)
|
|
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)
|
|
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(client)
|
|
r = client.post("/api/local-workspace/upload", files={"files": ("a.txt", b"x", "text/plain")})
|
|
assert r.status_code == 401
|