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).
549 lines
22 KiB
Python
549 lines
22 KiB
Python
"""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 "<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_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_migration_transaction_rolls_back():
|
|
"""A31 : un échec au milieu d'une migration ne laisse ni DDL partiel, ni
|
|
ligne dans schema_version → la reprise rejoue proprement."""
|
|
import sqlite3 as _sqlite3
|
|
|
|
import pytest as _pytest
|
|
|
|
from app.migrations import _apply_one, _ensure_table
|
|
|
|
conn = _sqlite3.connect(":memory:")
|
|
_ensure_table(conn)
|
|
|
|
def boom(c):
|
|
c.execute("CREATE TABLE partial_x (id INTEGER)")
|
|
raise RuntimeError("boom")
|
|
|
|
with _pytest.raises(RuntimeError, match="boom"):
|
|
_apply_one(conn, 9999, "boom", boom)
|
|
assert (
|
|
conn.execute("SELECT name FROM sqlite_master WHERE name='partial_x'").fetchone()
|
|
is None
|
|
), "DDL partiel non annulé"
|
|
assert (
|
|
conn.execute("SELECT COUNT(*) FROM schema_version WHERE version=9999").fetchone()[0]
|
|
== 0
|
|
)
|
|
# chemin nominal : DDL + marque de version dans la même transaction
|
|
_apply_one(conn, 9998, "ok", lambda c: c.execute("CREATE TABLE ok_x (id INTEGER)"))
|
|
assert (
|
|
conn.execute("SELECT COUNT(*) FROM schema_version WHERE version=9998").fetchone()[0]
|
|
== 1
|
|
)
|
|
conn.close()
|
|
|
|
|
|
def test_columns_helper_validates_table_name():
|
|
"""A31 : `columns()` remplace les 24 copies de PRAGMA table_info + valide l'identifiant."""
|
|
import sqlite3 as _sqlite3
|
|
|
|
from app.migrations import columns
|
|
|
|
conn = _sqlite3.connect(":memory:")
|
|
conn.execute("CREATE TABLE t1 (id INTEGER, nom TEXT)")
|
|
assert columns(conn, "t1") == {"id", "nom"}
|
|
try:
|
|
columns(conn, "t1; DROP TABLE users")
|
|
raise AssertionError("identifiant non validé")
|
|
except ValueError:
|
|
pass
|
|
conn.close()
|
|
|
|
|
|
def _assert_nonce(csp: str, html: str) -> str:
|
|
"""Header CSP : script-src sans unsafe-inline + TOUS les scripts inline noncés."""
|
|
import re as _re
|
|
|
|
m = _re.search(r"script-src ([^;]*);", csp)
|
|
assert m, csp
|
|
script_src = m.group(1)
|
|
nm = _re.search(r"'nonce-([^']+)'", script_src)
|
|
assert nm, script_src
|
|
nonce = nm.group(1)
|
|
assert "'unsafe-inline'" not in script_src, script_src
|
|
# A20 TERMINÉ : Alpine en build CSP (alpine.csp.min.js) + htmx allowEval=false
|
|
assert "'unsafe-eval'" not in script_src
|
|
assert "script-src-attr 'unsafe-inline'" in csp
|
|
tags = [
|
|
mm.group(0)
|
|
for mm in _re.finditer(r"<script[^>]*>", html)
|
|
if "src=" not in mm.group(0)
|
|
]
|
|
assert tags, "aucun script inline"
|
|
missing = [t for t in tags if f'nonce="{nonce}"' not in t]
|
|
assert missing == [], missing[:3]
|
|
return nonce
|
|
|
|
|
|
def test_csp_nonce_per_request(client):
|
|
"""A20 : nonce par requête — page base.html (avec meta htmx-config) et page
|
|
hors template (LOCAL_LOGIN_HTML, constante de module → nonce au rendu)."""
|
|
# 1) une page qui étend base.html (la meta htmx-config y est)
|
|
base = None
|
|
for url in ("/", "/dashboard", "/board", "/notes", "/settings", "/workspaces"):
|
|
cand = client.get(url)
|
|
if cand.status_code == 200 and "htmx-config" in cand.text:
|
|
base = cand
|
|
break
|
|
assert base is not None, "aucune page base.html atteignable"
|
|
nonce = _assert_nonce(base.headers.get("content-security-policy", ""), base.text)
|
|
assert f'"inlineScriptNonce": "{nonce}"' in base.text
|
|
# deux requêtes = deux nonces différents
|
|
other = client.get("/dashboard")
|
|
if other.status_code == 200 and "htmx-config" in other.text:
|
|
other_nonce = _re_search_nonce(other.headers.get("content-security-policy", ""))
|
|
assert other_nonce != nonce
|
|
|
|
# 2) la page de login hors template (script injecté par _with_nonce)
|
|
r = client.get("/auth/login?provider=local")
|
|
assert r.status_code == 200, r.status_code
|
|
_assert_nonce(r.headers.get("content-security-policy", ""), r.text)
|
|
|
|
|
|
def _re_search_nonce(csp: str) -> str:
|
|
import re as _re
|
|
|
|
return _re.search(r"'nonce-([^']+)'", _re.search(r"script-src ([^;]*);", csp).group(1)).group(1)
|
|
|
|
|
|
def test_csp_no_cdn_and_vendor(client):
|
|
"""A20 phase 2 : plus aucun hôte CDN tiers, chart/leaflet vendorisés,
|
|
connect-src fermé (scopé à l'hôte de la requête)."""
|
|
import pathlib as _pathlib
|
|
|
|
page = None
|
|
for url in ("/", "/dashboard", "/board", "/notes", "/settings", "/workspaces"):
|
|
cand = client.get(url)
|
|
if cand.status_code == 200 and "htmx-config" in cand.text:
|
|
page = cand
|
|
break
|
|
assert page is not None, "aucune page base.html atteignable"
|
|
csp = page.headers.get("content-security-policy", "")
|
|
assert "cdn.jsdelivr" not in csp, csp
|
|
assert "unpkg.com" not in csp, csp
|
|
assert "fonts.googleapis.com" not in csp, csp
|
|
assert "fonts.gstatic.com" not in csp, csp
|
|
import re as _re
|
|
|
|
conn = _re.search(r"connect-src ([^;]*);", csp).group(1)
|
|
assert conn == "'self' ws://testserver wss://testserver", conn
|
|
assert " https:" not in conn and not conn.startswith("https:"), conn
|
|
|
|
# vues chart/map : références locales (aucun CDN — A28 : le fichier
|
|
# collections.py est devenu un package, on balaie tous ses modules)
|
|
for src in _pathlib.Path("app/routers/collections").glob("*.py"):
|
|
text = src.read_text(encoding="utf-8")
|
|
assert "cdn.jsdelivr" not in text and "unpkg.com" not in text, src
|
|
|
|
# assets vendor servis
|
|
for path in (
|
|
"/static/js/vendor/chart.umd.js",
|
|
"/static/js/vendor/leaflet.js",
|
|
"/static/js/vendor/leaflet.css",
|
|
"/static/js/vendor/leaflet/images/marker-icon.png",
|
|
):
|
|
r = client.get(path)
|
|
assert r.status_code == 200, (path, r.status_code)
|
|
assert len(r.content) > 500, (path, len(r.content))
|
|
|
|
|
|
def test_http_client_shared_and_loop_scoped():
|
|
"""A42 : le client HTTP partagé est réutilisé dans la même boucle,
|
|
cloisonné par kwargs, et JAMAIS partagé entre deux boucles (un
|
|
AsyncClient lié à une boucle morte lèverait « Event loop is closed »)."""
|
|
import asyncio
|
|
|
|
from app.services.http_client import shared_client
|
|
|
|
async def same_loop():
|
|
async with shared_client(timeout=15) as a:
|
|
async with shared_client(timeout=15) as b:
|
|
assert a is b, "même boucle + mêmes kwargs = même client"
|
|
async with shared_client(timeout=30) as c:
|
|
assert c is not a, "kwargs différents = client différent"
|
|
return a
|
|
|
|
first = asyncio.run(same_loop())
|
|
# nouvelle boucle (façon tests : une boucle par test) → nouveau client
|
|
async def other_loop():
|
|
async with shared_client(timeout=15) as d:
|
|
assert d is not first, "client jamais réutilisé sur une boucle morte"
|
|
return d
|
|
|
|
second = asyncio.run(other_loop())
|
|
assert second is not first
|
|
|
|
|
|
def test_csrf_server_rendered_no_placeholder(client):
|
|
"""A43-1 : `hx-headers` est rendu côté serveur avec le vrai jeton (plus
|
|
de `__CSRF_PLACEHOLDER__` servi — la fenêtre de course JS disparaît),
|
|
et la valeur vaut le cookie `csrf_token` de la session."""
|
|
import json as _json
|
|
|
|
page = None
|
|
for url in ("/", "/dashboard", "/board", "/notes", "/settings", "/workspaces"):
|
|
cand = client.get(url)
|
|
if cand.status_code == 200 and "htmx-config" in cand.text:
|
|
page = cand
|
|
break
|
|
assert page is not None, "aucune page base.html atteignable"
|
|
# 1ʳᵉ visite : cookie créé dans la response → on refait un aller-retour
|
|
r = client.get(page.url if hasattr(page, "url") else "/dashboard")
|
|
if "htmx-config" not in r.text:
|
|
r = page
|
|
assert "__CSRF_PLACEHOLDER__" not in r.text, "placeholder servi au navigateur"
|
|
import re as _re
|
|
|
|
m = _re.search(r"hx-headers=\'([^\']*)\'", r.text)
|
|
assert m, "attribut hx-headers absent"
|
|
token = _json.loads(m.group(1).replace(""", '"'))["X-CSRF-Token"]
|
|
cookie = client.cookies.get("csrf_token", "")
|
|
assert cookie, "cookie csrf_token absent"
|
|
assert token == cookie, (token[:8], cookie[:8])
|
|
|
|
|
|
def test_no_duplicate_global_functions():
|
|
"""A38 : aucune fonction `function NAME` GLOBALE (profondeur 0) définie
|
|
2+ fois entre les templates et static/js — les paires à risque d'ombre
|
|
silencieuse (onDoc, escHtml, openCardDetail…) vivent dans des IIFEs ou
|
|
sont dédupliquées (openCardDetail → app.js)."""
|
|
import pathlib as _pathlib
|
|
import re as _re
|
|
|
|
files = list(_pathlib.Path("app/templates").glob("*.html")) + list(
|
|
_pathlib.Path("static/js").glob("*.js")
|
|
)
|
|
found: dict[str, list[str]] = {}
|
|
for p in files:
|
|
s = p.read_text(encoding="utf-8", errors="ignore")
|
|
depth = 0
|
|
line = 1
|
|
i = 0
|
|
state = None
|
|
n = len(s)
|
|
# ponytail: scanner naïve (strings/comments/backticks) — un faux
|
|
# positif se voit immédiatement à la lecture du nom signalé
|
|
while i < n:
|
|
c = s[i]
|
|
if c == "\n":
|
|
line += 1
|
|
if state is None:
|
|
if c in ('"', "'"):
|
|
state = c
|
|
i += 1
|
|
continue
|
|
if c == "`":
|
|
state = c
|
|
i += 1
|
|
continue
|
|
if c == "/" and i + 1 < n and s[i + 1] == "/":
|
|
state = "//"
|
|
i += 2
|
|
continue
|
|
if c == "/" and i + 1 < n and s[i + 1] == "*":
|
|
state = "/*"
|
|
i += 2
|
|
continue
|
|
if c == "{":
|
|
depth += 1
|
|
elif c == "}":
|
|
depth -= 1
|
|
else:
|
|
if c == "\\":
|
|
i += 2
|
|
continue
|
|
if (state in ('"', "'") and c == state) or (state == "`" and c == state):
|
|
state = None
|
|
elif state == "//" and c == "\n":
|
|
state = None
|
|
elif state == "/*" and c == "*" and i + 1 < n and s[i + 1] == "/":
|
|
state = None
|
|
i += 2
|
|
continue
|
|
i += 1
|
|
if state is None and depth == 0 and s.startswith("function ", i):
|
|
m = _re.match(r"function\s+([A-Za-z_]\w*)", s[i : i + 60])
|
|
if m:
|
|
found.setdefault(m.group(1), []).append(f"{p.name}:{line}")
|
|
dups = {k: v for k, v in found.items() if len(v) >= 2}
|
|
assert dups == {}, dups
|
|
|
|
|
|
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§ion=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
|