"""v7.55.0 — Connecteurs de l'agent : catalogue, CRUD, SSRF, statut, outil LLM.""" import asyncio from pathlib import Path import pytest from conftest import anon_csrf from app.db import get_conn from app.services import connectors from app.services.sso_provisioning import decrypt_secret ROOT = Path(__file__).resolve().parent.parent JS_PATH = ROOT / "static" / "js" / "agent_panel_2.js" PANEL_HTML = ROOT / "app" / "templates" / "agent_panel.html" PUBLIC_URL = "https://example.com/api" def _create(client, **kw): body = {"name": "Conn perso", "url": PUBLIC_URL, "secret": "sk-live-abc", **kw} r = client.post("/api/agent/connectors", json=body) assert r.status_code == 200, r.text return r.json() def test_native_connectors_always_listed(client): rows = client.get("/api/agent/connectors").json()["connectors"] native = {r["kind"]: r for r in rows if r["builtin"]} assert set(native) == {"gitea", "github", "web", "google", "ms365"} for r in native.values(): assert r["id"] is None assert r["status"] in ("ok", "missing") assert r["enabled"] is True assert native["web"]["status"] == "ok" # les connecteurs OAuth n'existent que si configurés (client_id/secret) assert native["google"]["oauth"] is True assert native["google"]["status"] == "missing" def test_create_custom_connector_never_returns_secret(client): row = _create(client) assert row["kind"] == "custom" and row["builtin"] is False assert row["has_secret"] is True assert "sk-live-abc" not in str(row) # jamais en clair dans la réponse with get_conn() as conn: stored = conn.execute( "SELECT secret_encrypted FROM agent_connectors WHERE id=?", (row["id"],) ).fetchone()["secret_encrypted"] assert stored and "sk-live-abc" not in stored # chiffré (Fernet) assert decrypt_secret(stored) == "sk-live-abc" @pytest.mark.parametrize("bad", [ "http://localhost/secret", "http://127.0.0.1:8080/admin", "http://169.254.169.254/latest/meta-data/", "ftp://exemple.com/api", "file:///etc/passwd", ]) def test_create_rejects_non_public_urls(client, bad): r = client.post("/api/agent/connectors", json={"name": "interne", "url": bad}) assert r.status_code == 400, r.text def test_patch_toggle_and_delete(client): row = _create(client) cid = row["id"] off = client.patch(f"/api/agent/connectors/{cid}", json={"enabled": False}) assert off.status_code == 200 and off.json()["enabled"] is False listed = {r["id"]: r for r in client.get("/api/agent/connectors").json()["connectors"]} assert listed[cid]["enabled"] is False assert client.delete(f"/api/agent/connectors/{cid}").status_code == 200 assert client.delete(f"/api/agent/connectors/{cid}").status_code == 404 assert client.patch(f"/api/agent/connectors/{cid}", json={"enabled": True}).status_code == 404 def test_connectors_require_session(client): anon_csrf(client) assert client.get("/api/agent/connectors").status_code == 401 assert client.post("/api/agent/connectors", json={"name": "x", "url": PUBLIC_URL}).status_code == 401 def test_probe_persists_status(client, monkeypatch): row = _create(client) cid = row["id"] async def ok_get(url, headers=None): assert url.startswith("https://example.com") # URL du connecteur assert (headers or {}).get("Authorization") == "Bearer sk-live-abc" # clé déchiffrée return 200, "{}" monkeypatch.setattr(connectors, "_get", ok_get) res = _probe(client, str(cid)) assert res["status"] == "ok", res with get_conn() as conn: st = conn.execute("SELECT status FROM agent_connectors WHERE id=?", (cid,)).fetchone() assert st["status"] == "ok" def _probe(client, target): r = client.post("/api/agent/connectors/probe", json={"connector": target}) assert r.status_code == 200, r.text return r.json() def test_probe_error_is_persisted(client, monkeypatch): row = _create(client) cid = row["id"] async def boom(url, headers=None): raise ValueError("Hôte non autorisé") monkeypatch.setattr(connectors, "_get", boom) res = _probe(client, str(cid)) assert res["status"] == "error" assert "Hôte non autorisé" in res["detail"] with get_conn() as conn: st = conn.execute("SELECT status, detail FROM agent_connectors WHERE id=?", (cid,)).fetchone() assert st["status"] == "error" def test_connector_fetch_tool_registered_and_works(client, monkeypatch): from app.services.tool_registry import ToolRegistry reg = ToolRegistry() schema = {t["name"]: t for t in reg.schema()} assert "connector_fetch" in schema assert schema["connector_fetch"]["parameters"]["required"] == ["connector"] row = _create(client) async def ok_get(url, headers=None): return 200, '{"ok": true, "source": "exemple"}' monkeypatch.setattr(connectors, "_get", ok_get) res = asyncio.run(reg.execute("connector_fetch", {"connector": str(row["id"]), "path": "/status"})) assert res.status == "success" assert "exemple" in res.data["text"] # connecteur désactivé → refusé client.patch(f"/api/agent/connectors/{row['id']}", json={"enabled": False}) res2 = asyncio.run(reg.execute("connector_fetch", {"connector": str(row["id"])})) assert res2.status == "error" assert "désactivé" in res2.message.lower() # connecteur inconnu → erreur outil (pas d'exception qui casse le run) res3 = asyncio.run(reg.execute("connector_fetch", {"connector": "999999"})) assert res3.status == "error" def test_connectors_menu_wired(client): src = JS_PATH.read_text(encoding="utf-8") root = src.split("var FD_PLUS_MENU = [", 1)[1].split("];", 1)[0] assert "action:'connectors'" in root assert root.count("disabled:true") == 0 # v7.58.0 : menu complet for action in ("'connector'", "'connector-new'", "'connector-probe'", "'connector-toggle'", "'connector-delete'"): assert action in src, action for fn in ("fetchConnectors()", "connectorCreate()", "connectorProbe()", "connectorToggle()", "connectorDelete()"): assert fn in src, fn html = PANEL_HTML.read_text(encoding="utf-8") assert "plusSection==='connector-new'" in html assert 'type="password"' in html # la clé n'est pas en clair à l'écran resp = client.get("/accounts") assert resp.status_code == 200 and "connectorForm" in resp.text