coffre ObsiGate Phase 4 (issue #22) : tests/test_coffre_reference_flowdeck.py

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