diff --git a/tests/test_coffre_reference_flowdeck.py b/tests/test_coffre_reference_flowdeck.py new file mode 100644 index 0000000..d0a5b4b --- /dev/null +++ b/tests/test_coffre_reference_flowdeck.py @@ -0,0 +1,282 @@ +"""Phase 4 (issue ObsiGate #22) — connecteurs Flowdeck à référence ObsiGate. + +Couverture : format de référence, création liée à un workspace, isolation +(A ne distribue pas avec l'octroi de B — le faux distributeur n'est même +pas appelé), distribution en mémoire sans persistance ni cache, octroi +révoqué, ObsiGate indisponible (aucun repli), connecteurs historiques +chiffrés toujours valides, routes API. + +Le distributeur est un FAUX endpoint (appelable injecté), clairement +identifié comme tel : aucun test ne parle à une vraie instance ObsiGate. +Les valeurs de test sont fictives (``TEST-VALUE-XXXX``). +""" + +import pytest +from conftest import anon_csrf + +from app.db import get_conn +from app.services import coffre_reference, connectors +from app.services.coffre_reference import ( + CoffreAccesRefuse, + CoffreIndisponible, + CoffreReferenceInvalide, + CoffreWorkspaceMismatch, +) + +PUBLIC_URL = "https://example.com/api" +REFERENCE = "coffre:flowdeck/gitea-lecture" +FAKE_VALUE = "TEST-VALUE-XXXX" + + +@pytest.fixture(autouse=True) +def _faux_jeton_consommateur(): + """Le distributeur exige un jeton consommateur : valeur fictive de test. + + Sans elle, ``distribuer`` refuserait avant même d'appeler le faux + endpoint (comportement voulu en production). + """ + import app.config + + app.config.settings.obsigate_consumer_token = "TEST-CONSUMER-TOKEN-XXXX" + yield + + +def _workspaces(): + with get_conn() as conn: + conn.execute( + "INSERT OR IGNORE INTO users (id, login, full_name, is_admin) " + "VALUES (1, 'tester', 'Tester', 1)" + ) + conn.execute( + "INSERT INTO workspaces (name, owner_id) VALUES ('Atelier A', 1)" + ) + conn.execute( + "INSERT INTO workspaces (name, owner_id) VALUES ('Atelier B', 1)" + ) + conn.commit() + rows = conn.execute( + "SELECT id FROM workspaces ORDER BY id" + ).fetchall() + return int(rows[0]["id"]), int(rows[1]["id"]) + + +class _FakeDistributeur: + """Faux endpoint ObsiGate : compte les appels, ne persiste rien.""" + + def __init__(self, behaviour="ok"): + self.behaviour = behaviour + self.calls = [] + + def __call__(self, url, payload, headers): + self.calls.append((url, dict(payload))) + assert url.endswith("/api/coffres/distribuer"), url + assert headers.get("Authorization", "").startswith("Bearer "), headers + if self.behaviour == "ok": + return 200, {"reference": payload["reference"], "valeur": FAKE_VALUE, + "version": 1} + if self.behaviour == "revoque": + return 403, {"detail": "octroi_revoque"} + if self.behaviour == "indisponible": + raise ConnectionError("faux ObsiGate injoignable") + raise AssertionError(self.behaviour) + + +def _create_coffre(client, ws, **kw): + body = {"name": "Conn coffre", "url": PUBLIC_URL, + "workspace_id": ws, "coffre_reference": REFERENCE, **kw} + r = client.post("/api/agent/connectors", json=body) + assert r.status_code == 200, r.text + return r.json() + + +# ── Format ───────────────────────────────────────────────────────────────── + +@pytest.mark.parametrize("bad", [ + "gitea-lecture", "coffre:", "coffre:slug-seul", "coffre:AB/mauvais", + "coffre:a/b", "coffre:ok/mauvais nom", "coffre:ok/", +]) +def test_reference_invalide_rejetee(bad): + with pytest.raises(CoffreReferenceInvalide): + coffre_reference.parse_reference(bad) + + +def test_reference_normalisee_minuscules(): + assert coffre_reference.parse_reference("Coffre:FlowDeck/Gitea-Lecture") == ( + "flowdeck", "gitea-lecture") + + +# ── Création liée au workspace ───────────────────────────────────────────── + +def test_creation_coffre_exige_workspace(client): + r = client.post("/api/agent/connectors", + json={"name": "Sans WS", "url": PUBLIC_URL, + "coffre_reference": REFERENCE}) + assert r.status_code == 400, r.text + assert "workspace" in r.text.lower() + + +def test_creation_coffre_refuse_secret_et_reference(client): + ws_a, _ = _workspaces() + r = client.post("/api/agent/connectors", + json={"name": "Ambigu", "url": PUBLIC_URL, "secret": "x", + "workspace_id": ws_a, "coffre_reference": REFERENCE}) + assert r.status_code == 400, r.text + + +def test_creation_coffre_ne_stocke_aucun_jeton(client): + ws_a, _ = _workspaces() + row = _create_coffre(client, ws_a) + assert row["coffre_reference"] == REFERENCE + assert row["workspace_id"] == ws_a + assert row["secret_source"] == "coffre" + assert row["has_secret"] is True + assert FAKE_VALUE not in str(row) + with get_conn() as conn: + stored = conn.execute( + "SELECT secret_encrypted, coffre_reference FROM agent_connectors " + "WHERE id=?", (row["id"],) + ).fetchone() + assert stored["secret_encrypted"] == "" # aucun jeton stocké + assert stored["coffre_reference"] == REFERENCE # métadonnée seulement + + +def test_creation_workspace_introuvable(client): + r = client.post("/api/agent/connectors", + json={"name": "WS ?", "url": PUBLIC_URL, + "workspace_id": 999999, "coffre_reference": REFERENCE}) + assert r.status_code == 400, r.text + + +# ── Résolution + isolation (critère central) ─────────────────────────────── + +def test_resolve_bon_workspace_distribue_en_memoire(client): + ws_a, _ = _workspaces() + row = _create_coffre(client, ws_a) + fake = _FakeDistributeur() + valeur = connectors.resolve_secret(row["id"], workspace_id=ws_a, + distribute=fake) + assert valeur == FAKE_VALUE + assert len(fake.calls) == 1 + _, payload = fake.calls[0] + assert payload["reference"] == REFERENCE + assert payload["contexte_type"] == "flowdeck_workspace" + assert payload["contexte_id"] == str(ws_a) + # Rien n'a été persisté ni mis en cache par la distribution. + with get_conn() as conn: + stored = conn.execute( + "SELECT secret_encrypted FROM agent_connectors WHERE id=?", + (row["id"],) + ).fetchone() + assert stored["secret_encrypted"] == "" + assert FAKE_VALUE not in str(connectors.get(row["id"])) + + +def test_isolation_workspace_a_ne_distribue_pas_octroi_de_b(client): + ws_a, ws_b = _workspaces() + row = _create_coffre(client, ws_b) # octroi du workspace B + fake = _FakeDistributeur() + with pytest.raises(CoffreWorkspaceMismatch) as exc: + connectors.resolve_secret(row["id"], workspace_id=ws_a, + distribute=fake) + assert "ne peut pas utiliser" in str(exc.value) + assert fake.calls == [] # distributeur jamais appelé + + +def test_resolve_sans_workspace_refuse(client): + ws_a, _ = _workspaces() + row = _create_coffre(client, ws_a) + fake = _FakeDistributeur() + with pytest.raises(ValueError, match="workspace_id"): + connectors.resolve_secret(row["id"], distribute=fake) + assert fake.calls == [] + + +def test_headers_bon_workspace_mauvais_workspace(client): + ws_a, ws_b = _workspaces() + row = _create_coffre(client, ws_a) + fake = _FakeDistributeur() + headers = connectors._headers(row["id"], workspace_id=ws_a, + distribute=fake) + assert headers["Authorization"] == f"Bearer {FAKE_VALUE}" + with pytest.raises(CoffreWorkspaceMismatch): + connectors._headers(row["id"], workspace_id=ws_b, distribute=fake) + + +# ── Échecs propres ───────────────────────────────────────────────────────── + +def test_octroi_revoque_message_explicite_sans_repli(client): + ws_a, _ = _workspaces() + row = _create_coffre(client, ws_a) + fake = _FakeDistributeur(behaviour="revoque") + with pytest.raises(CoffreAccesRefuse) as exc: + connectors.resolve_secret(row["id"], workspace_id=ws_a, + distribute=fake) + assert "révoqué" in str(exc.value) + assert "ré-accorder" in str(exc.value) + assert exc.value.detail == "octroi_revoque" + + +def test_obsigate_indisponible_sans_valeur_en_cache(client): + ws_a, _ = _workspaces() + row = _create_coffre(client, ws_a) + fake = _FakeDistributeur(behaviour="indisponible") + for _ in range(2): # deux appels : aucun ne doit renvoyer une valeur + with pytest.raises(CoffreIndisponible): + connectors.resolve_secret(row["id"], workspace_id=ws_a, + distribute=fake) + assert len(fake.calls) == 2 # chaque appel retente, rien en cache + + +# ── Migration opportuniste ───────────────────────────────────────────────── + +def test_connecteur_historique_chiffre_toujours_valide(client): + body = {"name": "Conn perso", "url": PUBLIC_URL, "secret": "sk-live-abc"} + row = client.post("/api/agent/connectors", json=body).json() + assert row["secret_source"] == "chiffre_local" + assert connectors.resolve_secret(row["id"]) == "sk-live-abc" + + +def test_bascule_patch_vers_reference_puis_retour(client): + ws_a, ws_b = _workspaces() + row = client.post("/api/agent/connectors", + json={"name": "Migrable", "url": PUBLIC_URL, + "secret": "sk-live-abc"}).json() + cid = row["id"] + # 1. rattacher au workspace, 2. basculer vers la référence. + r = client.patch(f"/api/agent/connectors/{cid}", + json={"workspace_id": ws_a}) + assert r.status_code == 200, r.text + r = client.patch(f"/api/agent/connectors/{cid}", + json={"coffre_reference": REFERENCE}) + assert r.status_code == 200, r.text + assert r.json()["secret_source"] == "coffre" + # Le workspace est immuable : pas de transfert d'octroi vers B. + r = client.patch(f"/api/agent/connectors/{cid}", + json={"workspace_id": ws_b}) + assert r.status_code == 400, r.text + # Effacer la référence réactive le jeton local dormant. + r = client.patch(f"/api/agent/connectors/{cid}", + json={"coffre_reference": ""}) + assert r.status_code == 200, r.text + assert r.json()["secret_source"] == "chiffre_local" + assert connectors.resolve_secret(cid) == "sk-live-abc" + + +def test_probe_coffre_sans_workspace_echoue_proprement(client, monkeypatch): + ws_a, _ = _workspaces() + row = _create_coffre(client, ws_a) + + async def ok_get(url, headers=None): + return 200, "{}" + + monkeypatch.setattr(connectors, "_get", ok_get) + r = client.post("/api/agent/connectors/probe", + json={"connector": str(row["id"])}) + assert r.status_code == 200 + assert r.json()["status"] == "error" # refus, pas de fuite inter-workspace + + +def test_routes_exigent_session(client): + anon_csrf(client) + assert client.post("/api/agent/connectors", + json={"name": "x", "url": PUBLIC_URL}).status_code == 401