diff --git a/CHANGELOG.md b/CHANGELOG.md index bd0c1f5..3658648 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,48 @@ # Changelog - FlowDeck +## v7.83.0 (2026-10-26) — Fusion : serveur MCP complété (ACC-5.1) et pont MCP vers ObsiGate (ACC-2.2) + +- **ACC-5.1 — serveur MCP Flowdeck complété** (`app/routers/mcp_server.py`, + spec d'accès centralisé §7.2/§7.4). Le catalogue de lecture gagne + `list_projects`, `get_project` (métadonnées + schéma + titres seuls), + `read_page` (page classique ou ligne de collection, contenu tronqué + proprement à 20 000 caractères) et `list_project_tasks` (tâches d'une + base projet, tous assignés). `search_pages` devient une vraie + recherche **plein texte** (titre + contenu des pages, titre + + propriétés des lignes de collection), et chaque résultat porte son + **lien profond** (`url`) et son `kind` — contrat figé : un projet + Flowdeck est une collection. §7.4 tenu : les **scopes du jeton** + sont appliqués (un jeton `read` se voit refuser les outils + d'écriture, erreur -32002) et **chaque appel `tools/call` est + journalisé** dans `api_audit_log` (outil, jeton, statut — jamais les + valeurs d'arguments). `list_meetings`/`get_meeting` restent hors + catalogue tant que la décision Meetings A/B n'est pas tranchée. +- **ACC-2.2 — pont MCP Flowdeck → ObsiGate, côté client** + (`app/services/mcp_client.py`, plan « étape 1 : lectures seules »). + Le client MCP gère désormais les serveurs **à session** (Streamable + HTTP d'ObsiGate) : capture et réémission de l'en-tête + `Mcp-Session-Id`, handshake complet avant `tools/call`, et décodage + SSE **multi-messages** (le message de réponse est sélectionné, pas + le premier `data:`). Nouveau preset de connecteur **« obsigate »** + (URL dérivée de la configuration dev/prod, clé dédiée au pont créée + dans ObsiGate → Configurations → Clés API & MCP). +- **Garde-fou n° 2 du plan du pont** : les outils MCP sont **classés** + à la découverte (annotation `readOnlyHint` du serveur, sinon + heuristique de nom ; lecture par défaut). Un outil MCP écrivant + (`propose_*`, `create_*`…) passe désormais par les mêmes portes que + les outils internes — politique d'approbation (`agent_policies`) et + rôles (`PermissionManager`, via le nouvel `is_write_tool`) — là où + il passait avant pour une lecture. Et un outil distant **`apply_*` + est bloqué côté client** : une écriture croisée ne s'exécute jamais + par un simple appel d'outil ; elle suit le patron §7.5 (propose → + approbation dans l'application propriétaire). +- **Preuve réelle T0–T5** : le code est testé contre un faux serveur + ObsiGate à session stricte (`tests/test_fusion_pont_obsigate.py`, + 7 tests ; `tests/test_fusion_mcp_server.py` porté à 18 tests). La + démonstration contre l'ObsiGate réel attend encore deux gestes de + Bruno : créer la clé API ObsiGate dédiée au pont, **limitée à un + seul vault**, et trancher **Q-A/Q-B** (fusion-trois-apps #1). + ## v7.82.4 (2026-10-25) — Résolution dev/prod par l'hôte de la requête : le mode console du panneau revit en dev - **Bug corrigé (dev)** : sur flowdeck-dev, le mode console du diff --git a/VERSION b/VERSION index 8ec0b31..22bce22 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -7.82.4 +7.83.0 diff --git a/app/main.py b/app/main.py index dd609e2..1f9fdb7 100644 --- a/app/main.py +++ b/app/main.py @@ -216,7 +216,7 @@ def _lire_version() -> str: # Littéral de version synchronisé avec le fichier VERSION : -# version="7.82.4" +# version="7.83.0" # Patron historique des bumps (voir scripts/bump_version.py, qui cherche # ce littéral) et contrôle de tests/test_pwa_offline.py. La version # runtime reste lue dynamiquement depuis VERSION par _lire_version() diff --git a/app/routers/mcp_server.py b/app/routers/mcp_server.py index 208c84d..eb724a5 100644 --- a/app/routers/mcp_server.py +++ b/app/routers/mcp_server.py @@ -18,6 +18,21 @@ Doctrine du programme : Flowdeck référence les tâches, il ne les copie pas — aucune tâche TikTak ne transite par ici. Écritures limitées à la création/mise à jour de pages et à la création de collections ; aucune suppression. + +**ACC-5.1 (architecture v2.0, §7.2)** — le catalogue de lecture cible +est couvert : ``list_projects`` / ``get_project`` (un « projet » du +contrat de liens profonds est une collection : ``/collections/``), +``search_pages`` (plein texte : titres **et** contenus, pages +classiques et lignes de collections), ``read_page`` (contenu tronqué +proprement) et ``list_project_tasks`` (collections ``is_task``). Les +outils réunions restent absents (décision Meetings A/B ouverte). +Chaque résultat d'objet porte son **URL profonde** (forme figée par +``contrats/liens-profonds.md``), comme le §7.2 l'exige. + +**§7.4** — chaque ``tools/call`` est journalisé dans +``api_audit_log`` (jeton, outil, horodatage) et la **portée du +jeton** est appliquée : sans le scope ``write``, les trois outils +d'écriture sont refusés (« jeton en lecture seule »). """ from __future__ import annotations @@ -74,6 +89,41 @@ OUTILS = [ {"name": "list_repos", "description": "Liste le registre des dépôts suivis par Flowdeck (builtin, Gitea, GitHub).", "inputSchema": {"type": "object", "properties": {}}}, + {"name": "list_projects", + "description": "Liste les PROJETS Flowdeck du périmètre du jeton " + "(catalogue §7.2) : au sens du contrat de liens " + "profonds, un projet est une collection — id, titre, " + "état (is_task, nombre de lignes) et URL profonde " + "/collections/.", + "inputSchema": {"type": "object", "properties": {}}}, + {"name": "get_project", + "description": "Détail d'un projet (collection) SANS le contenu " + "complet de ses pages : métadonnées, schéma des " + "propriétés et titres de ses lignes, URL profonde.", + "inputSchema": {"type": "object", + "properties": {"id": {"type": "integer"}}, + "required": ["id"]}}, + {"name": "read_page", + "description": "Lit le contenu d'une page (§7.2), tronqué " + "proprement à 20 000 caractères (drapeau " + "« truncated »). « kind » départage les deux " + "familles d'identifiants : « page » (page " + "classique d'un workspace, URL /pages/) ou " + "« collection_page » (ligne d'une collection, " + "contenu porté par sa page liée). Sans « kind », " + "la page classique est essayée d'abord.", + "inputSchema": {"type": "object", + "properties": {"id": {"type": "integer"}, + "kind": {"type": "string", + "enum": ["page", "collection_page"]}}, + "required": ["id"]}}, + {"name": "list_project_tasks", + "description": "Liste les tâches des projets « base de tâches » " + "(collections is_task) du périmètre du jeton : " + "titre, statut, échéance, assignés, URL profonde " + "du projet. « project_id » restreint à un projet.", + "inputSchema": {"type": "object", + "properties": {"project_id": {"type": "integer"}}}}, {"name": "create_page", "description": "Crée une page dans une collection Flowdeck.", "inputSchema": {"type": "object", @@ -99,6 +149,38 @@ OUTILS = [ "required": ["name"]}}, ] +#: Outils d'écriture — refusés aux jetons sans le scope ``write`` (§7.4). +OUTILS_ECRITURE = {"create_page", "update_page", "create_collection"} + +#: Borne de lecture d'un contenu de page (§7.2 : « tronqué proprement »). +CONTENU_MAX = 20000 + + +def _url_projet(collection_id: int) -> str: + """URL profonde d'un projet (contrat figé : /collections/).""" + from app.services import draco_urls + + return f"{draco_urls.flowdeck_url()}/collections/{collection_id}" + + +def _url_page(page_id: int) -> str: + """URL profonde d'une page classique (contrat figé : /pages/).""" + from app.services import draco_urls + + return f"{draco_urls.flowdeck_url()}/pages/{page_id}" + + +def _tronquer(texte: str) -> tuple[str, bool]: + """Troncature propre : jamais au milieu d'un mot si on peut l'éviter.""" + texte = texte or "" + if len(texte) <= CONTENU_MAX: + return texte, False + coupe = texte[:CONTENU_MAX] + dernier_espace = coupe.rfind(" ", CONTENU_MAX - 200) + if dernier_espace > 0: + coupe = coupe[:dernier_espace] + return coupe, True + def _props(brut): """property_values_json → dict quand c'est du JSON, sinon inchangé.""" @@ -112,8 +194,10 @@ def _props(brut): def _page_resume(row) -> dict: return {"id": row["id"], "collection_id": row["collection_id"], + "kind": "collection_page", "title": row["title"], "icon": row["icon"], "properties": _props(row["property_values_json"]), + "url": _url_projet(row["collection_id"]), "updated_at": row["updated_at"]} @@ -169,6 +253,7 @@ def _appel_outil(nom: str, args: dict, user_id: int): return [{"id": r["id"], "name": r["name"], "description": r["description"], "icon": r["icon"], "is_task": bool(r["is_task"]) if "is_task" in r.keys() else False, + "url": _url_projet(r["id"]), "page_count": r["page_count"]} for r in rows if r["eff_ws"] is None or r["eff_ws"] in accessible] @@ -182,7 +267,9 @@ def _appel_outil(nom: str, args: dict, user_id: int): "ORDER BY position, id", (coll["id"],)).fetchall() return {"collection": {"id": coll["id"], "name": coll["name"], "description": coll["description"], - "icon": coll["icon"]}, + "icon": coll["icon"], + "is_task": bool(coll["is_task"]) if "is_task" in coll.keys() else False, + "url": _url_projet(coll["id"])}, "pages": [_page_resume(p) for p in pages]} if nom == "get_page": p = conn.execute("SELECT * FROM collection_pages WHERE id = ?", @@ -195,19 +282,181 @@ def _appel_outil(nom: str, args: dict, user_id: int): res[cle] = p[cle] return res if nom == "search_pages": - q = f"%{str(args.get('query', '')).strip()}%" - if q == "%%": + # §7.2 : recherche plein texte — titres ET contenus, pages + # classiques comme lignes de collections, URL profonde et + # famille (« kind ») sur chaque résultat pour que read_page + # puisse être appelé sans ambiguïté d'identifiant. + terme = str(args.get("query", "")).strip() + if not terme: return [] + q = f"%{terme}%" + resultats: list[dict] = [] rows = conn.execute( - "SELECT * FROM collection_pages WHERE title LIKE ? " - "ORDER BY updated_at DESC LIMIT 25", (q,)).fetchall() - return [_page_resume(p) for p in rows - if collection_visible(p["collection_id"])] + "SELECT * FROM collection_pages " + "WHERE title LIKE ? OR property_values_json LIKE ? " + "ORDER BY updated_at DESC LIMIT 25", (q, q)).fetchall() + for p in rows: + if collection_visible(p["collection_id"]): + resultats.append(_page_resume(p)) + pages = conn.execute( + "SELECT id, title, workspace_id, updated_at FROM pages " + "WHERE deleted_at IS NULL AND (title LIKE ? OR content LIKE ?) " + "ORDER BY updated_at DESC LIMIT 25", (q, q)).fetchall() + for p in pages: + if p["workspace_id"] is None or p["workspace_id"] in accessible: + resultats.append({ + "id": p["id"], "kind": "page", "title": p["title"], + "workspace_id": p["workspace_id"], + "url": _url_page(p["id"]), + "updated_at": p["updated_at"]}) + return resultats[:50] if nom == "list_repos": rows = conn.execute( "SELECT id, name, proj_type, owner, language, description " "FROM projects ORDER BY proj_type, name").fetchall() return [dict(r) for r in rows] + if nom == "list_projects": + # §7.2 : un « projet » du contrat de liens profonds est une + # collection. Même périmètre et même forme que + # list_collections, vocabulaire du catalogue cible. + rows = conn.execute( + f"SELECT c.*, (SELECT COUNT(*) FROM collection_pages p " + f"WHERE p.collection_id = c.id) AS page_count, " + f"{eff_sql} AS eff_ws " + "FROM collections c ORDER BY c.name").fetchall() + return [{"id": r["id"], "title": r["name"], + "description": r["description"], "icon": r["icon"], + "is_task": bool(r["is_task"]) if "is_task" in r.keys() else False, + "page_count": r["page_count"], + "url": _url_projet(r["id"]), + "updated_at": r["updated_at"]} + for r in rows + if r["eff_ws"] is None or r["eff_ws"] in accessible] + if nom == "get_project": + coll = conn.execute("SELECT * FROM collections WHERE id = ?", + (int(args.get("id", 0)),)).fetchone() + if coll is None or not collection_visible(coll["id"]): + return {"erreur": "projet introuvable"} + schema = _props(coll["schema_json"]) \ + if coll["schema_json"] else [] + titres = conn.execute( + "SELECT id, title FROM collection_pages " + "WHERE collection_id = ? ORDER BY position, id LIMIT 300", + (coll["id"],)).fetchall() + return {"project": {"id": coll["id"], "title": coll["name"], + "description": coll["description"], + "icon": coll["icon"], + "is_task": bool(coll["is_task"]) if "is_task" in coll.keys() else False, + "url": _url_projet(coll["id"]), + "updated_at": coll["updated_at"]}, + "schema": schema, + "pages": [{"id": t["id"], "title": t["title"]} + for t in titres]} + if nom == "read_page": + try: + pid = int(args.get("id")) + except (TypeError, ValueError): + return {"erreur": "id requis (entier)"} + kind = str(args.get("kind") or "") + if kind in ("page", ""): + page = conn.execute( + "SELECT * FROM pages WHERE id = ? AND deleted_at IS NULL", + (pid,)).fetchone() + if page is not None and ( + page["workspace_id"] is None + or page["workspace_id"] in accessible): + contenu, tronque = _tronquer(page["content"] or "") + return {"id": page["id"], "kind": "page", + "title": page["title"], + "workspace_id": page["workspace_id"], + "content": contenu, "truncated": tronque, + "url": _url_page(page["id"]), + "updated_at": page["updated_at"]} + if kind == "page": + return {"erreur": "page introuvable"} + if kind in ("collection_page", ""): + p = conn.execute( + "SELECT * FROM collection_pages WHERE id = ?", + (pid,)).fetchone() + if p is None or not collection_visible(p["collection_id"]): + return {"erreur": "page introuvable"} + res = _page_resume(p) + liee = conn.execute( + "SELECT content FROM pages " + "WHERE collection_row_id = ? AND deleted_at IS NULL", + (pid,)).fetchone() + if liee is not None: + contenu, tronque = _tronquer(liee["content"] or "") + res["content"] = contenu + res["truncated"] = tronque + return res + return {"erreur": f"kind inconnu : {kind}"} + if nom == "list_project_tasks": + from app.services.task_databases import collect_tasks, is_done + + projet_id = args.get("project_id") + if projet_id is not None: + try: + projet_id = int(projet_id) + except (TypeError, ValueError): + return {"erreur": "project_id doit être un entier"} + cible = conn.execute( + "SELECT id, is_task FROM collections WHERE id = ?", + (projet_id,)).fetchone() + if cible is None or not collection_visible(projet_id): + return {"erreur": "projet introuvable"} + if "is_task" in cible.keys() and not cible["is_task"]: + return {"erreur": "ce projet n'est pas une base de tâches"} + taches = collect_tasks(conn, user_id, mine_only=False) + sortie = [] + for t in taches: + if projet_id is not None and t["collection_id"] != projet_id: + continue + sortie.append({ + "id": t["id"], "title": t["title"], + "status": t["status"], "done": t["done"], + "due": t["due"], "assignees": t["assignees"], + "project_id": t["collection_id"], + "project_name": t["collection_name"], + "url": _url_projet(t["collection_id"])}) + if projet_id is not None and not sortie: + # Base de tâches non « configurée » (mapping My Tasks + # incomplet) : collect_tasks la saute par conception, + # mais le serveur MCP répond quand même — lignes brutes, + # statut lu sur la propriété mappée ou nommée Statut. + coll = conn.execute( + "SELECT * FROM collections WHERE id = ?", + (projet_id,)).fetchone() + prop_statut = None + if "task_status_prop" in coll.keys() and coll["task_status_prop"]: + prop_statut = coll["task_status_prop"] + else: + pr = conn.execute( + "SELECT id FROM collection_properties " + "WHERE collection_id = ? AND LOWER(name) IN " + "('statut', 'status', 'état', 'etat') " + "ORDER BY position LIMIT 1", (projet_id,)).fetchone() + if pr is not None: + prop_statut = pr["id"] + lignes = conn.execute( + "SELECT * FROM collection_pages WHERE collection_id = ? " + "ORDER BY position, id", (projet_id,)).fetchall() + for ligne in lignes: + props = _props(ligne["property_values_json"]) + statut = "" + if isinstance(props, dict) and prop_statut is not None: + v = props.get(prop_statut, props.get(str(prop_statut), "")) + if isinstance(v, list): + v = v[0] if v else "" + statut = str(v or "") + sortie.append({ + "id": ligne["id"], "title": ligne["title"], + "status": statut, "done": is_done(statut), + "due": "", "assignees": [], + "project_id": projet_id, + "project_name": coll["name"], + "url": _url_projet(projet_id)}) + return sortie if nom == "create_page": try: collection_id = int(args.get("collection_id")) @@ -297,7 +546,8 @@ def _appel_outil(nom: str, args: dict, user_id: int): emettre("flowdeck.collection.created", f"collection:{new_id}", name, url=url_collection(new_id), actor="agent:mcp") return {"id": coll["id"], "name": coll["name"], - "description": coll["description"], "icon": coll["icon"]} + "description": coll["description"], "icon": coll["icon"], + "url": _url_projet(coll["id"])} return {"erreur": f"outil inconnu : {nom}"} @@ -310,11 +560,45 @@ def _version() -> str: return "dev" +def _scopes_jeton(token_id: int) -> str: + """Scopes du jeton MCP (colonne ``api_tokens.scopes``, défaut hérité + ``read,write`` posé par la migration v6.3.0).""" + try: + with get_conn() as conn: + row = conn.execute("SELECT scopes FROM api_tokens WHERE id = ?", + (token_id,)).fetchone() + if row is not None and "scopes" in row.keys() and row["scopes"]: + return str(row["scopes"]) + except Exception: # noqa: BLE001 — base ancienne sans la colonne + pass + return "read,write" + + +def _journaliser(proprietaire: dict, outil: str, arguments: dict, + request: Request, statut: str) -> None: + """Journal d'audit §7.4 : qui (jeton), quel outil, quand — les + valeurs d'arguments ne sont pas journalisées, seules leurs clés.""" + try: + from app.services.api_v2_helpers import audit_log + + cles = ",".join(sorted(str(k) for k in (arguments or {}).keys())) + audit_log({"id": proprietaire["user_id"], + "_token_id": proprietaire["id"]}, + f"mcp:{outil}", "mcp_tool", outil, + f"statut={statut}; jeton={proprietaire.get('name', '')}; args=[{cles}]", + request) + except Exception: # noqa: BLE001 — l'audit ne casse jamais l'appel + pass + + @router.get("/mcp") def mcp_info(): + noms = [t["name"] for t in OUTILS] return {"name": "flowdeck", "version": _version(), "transport": "streamable-http (JSON-RPC 2.0 over POST)", - "read_only": False, "tools": [t["name"] for t in OUTILS]} + "read_only": False, "tools": noms, + "read_tools": [n for n in noms if n not in OUTILS_ECRITURE], + "write_tools": [n for n in noms if n in OUTILS_ECRITURE]} @router.post("/mcp") @@ -349,8 +633,25 @@ async def mcp(request: Request): return {"jsonrpc": "2.0", "id": rid, "result": {"tools": OUTILS}} if methode == "tools/call": params = data.get("params") or {} - res = _appel_outil(params.get("name", ""), params.get("arguments") or {}, - proprietaire["user_id"]) + outil = params.get("name", "") + arguments = params.get("arguments") or {} + # §7.4 : la portée du jeton fait la loi — un jeton « read » + # (ex. clé dédiée à un pont de lecture) ne peut pas écrire. + if outil in OUTILS_ECRITURE: + from app.services.api_v2_helpers import has_scope + + if not has_scope(_scopes_jeton(proprietaire["id"]), "write"): + _journaliser(proprietaire, outil, arguments, request, + "refuse_portee") + return {"jsonrpc": "2.0", "id": rid, + "error": {"code": -32002, + "message": "Jeton en lecture seule : " + "l'outil d'écriture " + f"« {outil} » exige le scope write"}} + res = _appel_outil(outil, arguments, proprietaire["user_id"]) + _journaliser(proprietaire, outil, arguments, request, + "erreur" if isinstance(res, dict) and "erreur" in res + else "ok") return {"jsonrpc": "2.0", "id": rid, "result": {"content": [{"type": "text", "text": json.dumps(res, ensure_ascii=False)}]}} diff --git a/app/services/agent_engine.py b/app/services/agent_engine.py index 1190dfa..f5f3964 100644 --- a/app/services/agent_engine.py +++ b/app/services/agent_engine.py @@ -22,7 +22,7 @@ from app.services import agent_memory from app.services.agent_policies import check_tool, get_policy from app.services.context_builder import ContextBuilder from app.services.llm_client import LLMClient -from app.services.permission_manager import WRITE_TOOLS, PermissionManager +from app.services.permission_manager import PermissionManager, is_write_tool from app.services.tool_registry import ToolRegistry logger = logging.getLogger(__name__) @@ -248,7 +248,7 @@ class AgentEngine: # v7.2.0 — workspace tool scope + human approval gate, checked # *before* permissions so a scoped-out tool never reaches ACLs. gov = check_tool(self.user_id, self.workspace_id, tool, - is_write=tool in WRITE_TOOLS, + is_write=is_write_tool(tool), conversation_id=conversation_id) if not gov.get("allowed"): detail = gov.get("reason") or "Refusé par la politique agent" diff --git a/app/services/connectors.py b/app/services/connectors.py index b7a44e0..a99e8e3 100644 --- a/app/services/connectors.py +++ b/app/services/connectors.py @@ -24,7 +24,11 @@ from app.services.sso_provisioning import decrypt_secret, encrypt_secret logger = logging.getLogger(__name__) NATIVE_KINDS = ("gitea", "github", "web", "google", "ms365") -CUSTOM_KINDS = ("custom", "discord", "telegram", "mcp", "tiktak") +CUSTOM_KINDS = ("custom", "discord", "telegram", "mcp", "tiktak", "obsigate") +#: Kinds personnels qui parlent MCP (probe = handshake + tools/list, +#: outils dynamiques ``mcp__*``) — générique + les deux presets +#: de fédération du programme (TikTak ACC-2.1, ObsiGate ACC-2.2). +MCP_KINDS = ("mcp", "tiktak", "obsigate") AUTH_SCHEMES = ("bearer", "bot", "none") # Presets proposés par le formulaire du menu ＋ (le jeton reste toujours saisi # par l'utilisateur — jamais codé en dur ici). @@ -45,6 +49,16 @@ def _presets() -> dict: # v7.82.0 : URL dérivée de la config (TIKTAK_URL ; défaut officiel). "tiktak": {"url": f"{draco_urls.tiktak_url()}/mcp", "auth": "bearer", "hint": "TikTak → Réglages → Intégrations → clé API tck_…"}, + # Fusion ACC-2.2 — pont MCP vers ObsiGate (plan « étape 1 : + # lectures seules », tests T0–T5) : serveur MCP d'ObsiGate sur + # /mcp (Streamable HTTP à session — le client gère le + # Mcp-Session-Id). La clé se crée dans ObsiGate → + # Configurations → Clés API & MCP, dédiée au pont et limitée + # à UN SEUL vault au départ : c'est le jeton qui fait la + # portée (garde-fou n° 1 du plan), pas ce preset. + "obsigate": {"url": f"{draco_urls.obsigate_url()}/mcp", "auth": "bearer", + "hint": "ObsiGate → Configurations → Clés API & MCP → " + "clé dédiée au pont, limitée à un seul vault"}, } @@ -293,7 +307,7 @@ async def probe(connector: str, user_id: int | None = None) -> dict: """Teste un connecteur et **persiste** le résultat (personnel uniquement).""" kind, ref = _resolve(connector) try: - if kind in ("mcp", "tiktak"): + if kind in MCP_KINDS: from app.services import mcp_client tools = await mcp_client.initialize_and_list(int(ref)) status, detail = "ok", f"MCP · {len(tools)} outil(s)" @@ -343,7 +357,7 @@ async def connector_fetch(connector: str, path: str = "", query: str = "", path = (path or "").strip() if kind in OAUTH_KINDS: return await oauth_connectors.api_get(kind, user_id, path) - if kind in ("mcp", "tiktak"): + if kind in MCP_KINDS: from app.services import mcp_client return {"status": "ok", "text": mcp_client.tools_text(int(ref))} if kind == "gitea": diff --git a/app/services/mcp_client.py b/app/services/mcp_client.py index f7f24fa..d4bb49b 100644 --- a/app/services/mcp_client.py +++ b/app/services/mcp_client.py @@ -6,10 +6,30 @@ JSON-RPC 2.0 : ``initialize`` → ``notifications/initialized`` → ``tools/list ``ToolRegistry`` sous des noms ``mcp__`` — outils dynamiques, sans état en mémoire. -ponytail : **1 seul seam** ``_rpc()`` (monkeypatché en test) ; on lit soit la -réponse JSON directe, soit la première ligne ``data:`` d'une -``text/event-stream`` — pas de client SSE à l'état (les méthodes utilisées -répondent en JSON chez la grande majorité des serveurs). +**Sessions (ACC-2.2, pont ObsiGate — test T0 du plan d'étape 1).** +Certains serveurs Streamable HTTP (ObsiGate) sont **à session** : ils +renvoient un en-tête ``Mcp-Session-Id`` à ``initialize`` et refusent +tout appel qui ne le réémet pas. Le client capture donc cet en-tête à +chaque réponse et le réémet sur les appels suivants de la même +opération ; ``call_tool`` fait désormais le handshake complet avant +``tools/call`` (les serveurs sans état comme TikTak l'acceptent sans +frais). Réponses **JSON ou SSE** : dans un flux ``text/event-stream`` +qui porte plusieurs messages (notifications + réponse), c'est le +message **de réponse** (id correspondant, sinon porteur de +``result``/``error``) qui est retenu — pas le premier ``data:`` venu. + +**Classification des outils (garde-fou n° 2 du plan du pont).** Chaque +outil découvert est classé au moment du cache : **lecture par défaut** ; +écrivant si le serveur le déclare (annotation ``readOnlyHint: false``) +ou si son nom porte un verbe d'écriture (``create_``, ``propose_``, +``apply_``…). La classe est consommée par les portes de gouvernance +(``agent_policies.check_tool`` et ``PermissionManager`` via +``permission_manager.is_write_tool``) : un outil MCP écrivant suit le +même régime d'approbation et de rôles que les outils internes. + +ponytail : **1 seul seam** ``_rpc()`` (monkeypatché en test) ; pas de +client SSE à l'état (les méthodes utilisées répondent en JSON ou en +événements ponctuels chez les serveurs du programme). """ from __future__ import annotations @@ -19,7 +39,7 @@ import logging import re from app.db import get_conn -from app.services.connectors import _headers, _with_secret, get +from app.services.connectors import MCP_KINDS, _headers, _with_secret, get logger = logging.getLogger(__name__) @@ -27,6 +47,36 @@ logger = logging.getLogger(__name__) PROTOCOL_VERSION = "2024-11-05" CLIENT_INFO = {"name": "FlowDeck", "version": "7"} +#: Verbes d'écriture reconnus en tête d'un nom d'outil MCP quand le +#: serveur ne déclare pas d'annotation ``readOnlyHint``. ``propose_`` / +#: ``apply_`` sont le patron d'écriture approuvée d'ObsiGate : les deux +#: sont classés écrivants (voir aussi le blocage d'``apply_`` dans +#: ``tool_registry.McpTool``). +_PREFIXES_ECRITURE = ( + "create_", "update_", "delete_", "add_", "remove_", "set_", "move_", + "write_", "edit_", "rename_", "propose_", "apply_", "archive_", + "unarchive_", "send_", "publish_", "close_", "sync_", "trigger_", + "import_", +) + +#: Derniers ``Mcp-Session-Id`` capturés, par URL de serveur. Mémoire de +#: processus volontairement éphémère : une session perdue se rouvre par +#: un handshake, jamais par une persistance. +_sessions: dict[str, str] = {} + + +def classer_outil(nom_original: str, annotations: dict | None = None) -> bool: + """True si l'outil MCP est **écrivant**. + + L'annotation du serveur fait foi quand elle existe + (``readOnlyHint``) ; sinon, heuristique de nom. Lecture par défaut — + c'est la règle du plan du pont : un outil ne devient écrivant que + s'il est déclaré comme tel. + """ + if isinstance(annotations, dict) and "readOnlyHint" in annotations: + return not bool(annotations["readOnlyHint"]) + return (nom_original or "").lower().startswith(_PREFIXES_ECRITURE) + def tool_name(server_name: str, tool_name: str) -> str: """Nom LLM stable : ``mcp__`` (slug minuscules/underscores).""" @@ -36,24 +86,66 @@ def tool_name(server_name: str, tool_name: str) -> str: return re.sub(r"_{2,}", "_", joined) -def parse_body(content_type: str, body: str, status: int) -> dict: - """Corps MCP → dict JSON (JSON direct ou événement ``data:`` d'un flux SSE).""" +def parse_body(content_type: str, body: str, status: int, + expected_id=None) -> dict: + """Corps MCP → dict JSON (JSON direct ou événements ``data:`` SSE). + + Dans un flux SSE multi-messages, le message retenu est celui dont + l'``id`` correspond à la requête (``expected_id``), sinon le + premier porteur de ``result``/``error``, sinon le premier message. + """ text = body or "" + data = None if "text/event-stream" in (content_type or ""): + candidats: list = [] for line in text.splitlines(): if line.startswith("data:"): - text = line[5:].strip() - break - try: - data = json.loads(text) - except ValueError as exc: - raise ValueError(f"Réponse MCP illisible (HTTP {status})") from exc + try: + candidats.append(json.loads(line[5:].strip())) + except ValueError: + continue + if expected_id is not None: + data = next((c for c in candidats + if isinstance(c, dict) and c.get("id") == expected_id), + None) + if data is None: + data = next((c for c in candidats + if isinstance(c, dict) + and ("result" in c or "error" in c)), None) + if data is None and candidats: + data = candidats[0] + if data is None: + raise ValueError(f"Réponse MCP illisible (HTTP {status})") + else: + try: + data = json.loads(text) + except ValueError as exc: + raise ValueError(f"Réponse MCP illisible (HTTP {status})") from exc if isinstance(data, dict) and data.get("error"): err = data["error"] if isinstance(data["error"], dict) else {} raise ValueError(f"MCP {err.get('code', '?')} : {err.get('message', 'erreur')}") return data if isinstance(data, dict) else {} +def _capturer_session(url: str, headers) -> None: + """Mémorise le ``Mcp-Session-Id`` d'une réponse, s'il est présent.""" + try: + sid = headers.get("mcp-session-id") if headers else None + except AttributeError: + sid = None + if sid: + _sessions[url] = sid + + +def _session_headers(url: str, headers: dict | None) -> dict: + """En-têtes d'appel + session connue pour cette URL (si capturée).""" + out = dict(headers or {}) + sid = _sessions.get(url) + if sid and not any(k.lower() == "mcp-session-id" for k in out): + out["Mcp-Session-Id"] = sid + return out + + async def _rpc(url: str, payload: dict, headers: dict | None = None) -> dict: """POST JSON-RPC gardé SSRF → réponse décodée. Point d'injection des tests.""" from app.services.http_client import shared_client @@ -65,8 +157,9 @@ async def _rpc(url: str, payload: dict, headers: dict | None = None) -> dict: **(headers or {})} async with shared_client(timeout=15, headers=hdrs) as client: resp = await client.post(safe, json=payload) + _capturer_session(url, resp.headers) return parse_body(resp.headers.get("content-type", ""), resp.text or "", - resp.status_code) + resp.status_code, expected_id=payload.get("id")) async def _notify(url: str, payload: dict, headers: dict | None = None) -> None: @@ -77,11 +170,30 @@ async def _notify(url: str, payload: dict, headers: dict | None = None) -> None: logger.debug("MCP notification ignorée: %s", exc) +async def _handshake(url: str, headers: dict | None) -> dict: + """``initialize`` + ``notifications/initialized`` → en-têtes de session. + + Obligatoire avant tout ``tools/call`` contre un serveur à session + (ObsiGate) ; sans effet de bord contre un serveur sans état. Les + en-têtes retournés portent le ``Mcp-Session-Id`` capturé pendant + ``initialize`` quand le serveur en a émis un. + """ + await _rpc(url, { + "jsonrpc": "2.0", "id": 1, "method": "initialize", + "params": {"protocolVersion": PROTOCOL_VERSION, "capabilities": {}, + "clientInfo": CLIENT_INFO}, + }, headers) + session = _session_headers(url, headers) + await _notify(url, {"jsonrpc": "2.0", "method": "notifications/initialized"}, + session) + return session + + def _server(connector_id: int) -> dict: row = get(connector_id) if row is None: raise ValueError("Serveur MCP introuvable") - if row["kind"] not in ("mcp", "tiktak"): + if row["kind"] not in MCP_KINDS: raise ValueError("Ce connecteur n'est pas un serveur MCP") if not row["enabled"]: raise ValueError("Serveur MCP désactivé") @@ -102,6 +214,7 @@ def _normalize(server: dict, raw: list) -> list[dict]: "original": str(t["name"]), "description": str(t.get("description") or "Outil MCP"), "parameters": params, + "write": classer_outil(str(t["name"]), t.get("annotations")), }) return out @@ -117,15 +230,9 @@ async def initialize_and_list(connector_id: int) -> list[dict]: """Handshake + ``tools/list`` → met à jour le cache de la base.""" srv = _server(connector_id) url = _with_secret(srv["url"], connector_id) - await _rpc(url, { - "jsonrpc": "2.0", "id": 1, "method": "initialize", - "params": {"protocolVersion": PROTOCOL_VERSION, "capabilities": {}, - "clientInfo": CLIENT_INFO}, - }, srv["headers"]) - await _notify(url, {"jsonrpc": "2.0", "method": "notifications/initialized"}, - srv["headers"]) + session = await _handshake(url, srv["headers"]) data = await _rpc(url, {"jsonrpc": "2.0", "id": 2, "method": "tools/list", - "params": {}}, srv["headers"]) + "params": {}}, session) result = data.get("result") if isinstance(data.get("result"), dict) else {} tools = _normalize(srv, result.get("tools") or []) _save_tools(connector_id, tools) @@ -137,7 +244,7 @@ def cached_tools() -> list[dict]: with get_conn() as conn: rows = conn.execute( "SELECT id, tools_json FROM agent_connectors " - "WHERE kind IN ('mcp', 'tiktak') AND enabled=1" + "WHERE kind IN ('mcp', 'tiktak', 'obsigate') AND enabled=1" ).fetchall() out: list[dict] = [] for r in rows: @@ -170,13 +277,18 @@ def _content_text(result: dict) -> str: async def call_tool(connector_id: int, tool: str, arguments: dict | None = None) -> dict: - """``tools/call`` → {status, text} (isError → error, borné à 20 000 car.).""" + """``tools/call`` → {status, text} (isError → error, borné à 20 000 car.). + + Le handshake (avec session éventuelle) précède systématiquement + l'appel : c'est la condition T0 des serveurs à session. + """ srv = _server(connector_id) url = _with_secret(srv["url"], connector_id) + session = await _handshake(url, srv["headers"]) data = await _rpc(url, { "jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": {"name": tool, "arguments": arguments or {}}, - }, srv["headers"]) + }, session) result = data.get("result") if isinstance(data.get("result"), dict) else {} text = _content_text(result)[:20000] return {"status": "error" if result.get("isError") else "ok", "text": text} diff --git a/app/services/permission_manager.py b/app/services/permission_manager.py index f9952b6..30cc42d 100644 --- a/app/services/permission_manager.py +++ b/app/services/permission_manager.py @@ -55,6 +55,28 @@ DESTRUCTIVE_TOOLS = { } +def is_write_tool(tool: str) -> bool: + """L'outil est-il écrivant ? Outils internes (``WRITE_TOOLS``) et + outils **MCP dynamiques** classés écrivants à la découverte + (``mcp_client.classer_outil`` — garde-fou n° 2 du pont ObsiGate, + ACC-2.2 : avant, un ``mcp_*`` passait toujours pour une lecture, + donc un ``apply_*`` distant échappait aux deux portes). + """ + if tool in WRITE_TOOLS: + return True + if tool.startswith("mcp_"): + try: + from app.services import mcp_client + + for entry in mcp_client.cached_tools(): + if entry.get("name") == tool: + return bool(entry.get("write")) + except Exception: # noqa: BLE001 — la classification ne casse rien + logger.debug("is_write_tool: classification MCP indisponible", + exc_info=True) + return False + + class PermissionManager: """Resolves workspace role and gates agent + granular ACL checks.""" @@ -186,7 +208,7 @@ class PermissionManager: - destructive tools: admin/owner, or requires confirm approval mode. """ role = self.role_in_workspace(workspace_id) - if tool in WRITE_TOOLS and role not in WRITE_ROLES: + if is_write_tool(tool) and role not in WRITE_ROLES: raise HTTPException( status_code=403, detail=f"Agent tool '{tool}' requires editor+ role (user is '{role}')", diff --git a/app/services/tool_registry.py b/app/services/tool_registry.py index 04c6c1d..560ac91 100644 --- a/app/services/tool_registry.py +++ b/app/services/tool_registry.py @@ -1230,9 +1230,23 @@ class McpTool(Tool): async def execute(self, args, *, user_id=None) -> ToolResult: from app.services import mcp_client + # Garde-fou du pont (ACC-2.2, patron §7.5) : un outil distant + # « apply_* » EXÉCUTE une écriture croisée — il n'est jamais + # utilisable par un simple appel d'outil depuis un agent + # Flowdeck. L'écriture passe par « propose_* » (classé + # écrivant : rôles + approbation) puis par l'approbation dans + # l'application propriétaire de la donnée. + original = self.entry.get("original") or "" + if original.lower().startswith("apply_"): + return ToolResult( + status="error", tool=self.name, target_type="mcp", + message=(f"Outil « {original} » bloqué côté client : une " + "écriture croisée ne s'exécute pas par un appel " + "d'outil — propose_* puis approbation dans " + "l'application propriétaire (patron §7.5).")) try: res = await mcp_client.call_tool(self.entry["connector_id"], - self.entry.get("original") or "", args) + original, args) except ValueError as exc: return ToolResult(status="error", tool=self.name, message=str(exc)) except Exception as exc: # noqa: BLE001 — réseau : erreur outil, pas run diff --git a/app/templates/agent_panel.html b/app/templates/agent_panel.html index ece6a65..4c7790b 100644 --- a/app/templates/agent_panel.html +++ b/app/templates/agent_panel.html @@ -452,6 +452,7 @@ body.fd-ap-resizing *{cursor:col-resize!important} + @@ -533,5 +534,5 @@ body.fd-ap-resizing *{cursor:col-resize!important} - + diff --git a/static/js/agent_panel_2.js b/static/js/agent_panel_2.js index 3dd821c..7ffe65f 100644 --- a/static/js/agent_panel_2.js +++ b/static/js/agent_panel_2.js @@ -1425,7 +1425,9 @@ hint:'BotFather → Bot Token ({secret} complète l\u2019URL)'}, mcp: {url:'', auth:'bearer', hint:'URL du endpoint MCP (streamable HTTP)'}, tiktak: {url:(window.FLOWDECK_TIKTAK_MCP_URL || 'https://tiktak.dracodev.net/mcp'), auth:'bearer', name:'TikTak', - hint:'TikTak → Réglages → Intégrations → clé API tck_…'} + hint:'TikTak → Réglages → Intégrations → clé API tck_…'}, + obsigate:{url:(window.FLOWDECK_OBSIGATE_MCP_URL || 'https://og.dracodev.net/mcp'), auth:'bearer', name:'ObsiGate', + hint:'ObsiGate → Configurations → Clés API & MCP → clé dédiée au pont, limitée à un seul vault'} }; var p = presets[f.kind] || presets.custom; f.url = p.url; f.auth = p.auth; f.hint = p.hint; diff --git a/tests/test_fusion_mcp_server.py b/tests/test_fusion_mcp_server.py index f777410..5a8aeb4 100644 --- a/tests/test_fusion_mcp_server.py +++ b/tests/test_fusion_mcp_server.py @@ -70,8 +70,9 @@ def test_mcp_handshake_et_outils(client, jeton): noms = [t["name"] for t in r.json()["result"]["tools"]] assert noms == ["list_workspaces", "get_workspace", "list_collections", "get_collection", "get_page", - "search_pages", "list_repos", "create_page", - "update_page", "create_collection"] + "search_pages", "list_repos", "list_projects", + "get_project", "read_page", "list_project_tasks", + "create_page", "update_page", "create_collection"] def test_mcp_workspaces(client, jeton): @@ -107,7 +108,13 @@ def test_mcp_collections_et_pages(client, jeton, donnees): assert titres == ["Page de cadrage", "Revue hebdo"] assert detail["pages"][0]["properties"] == {"statut": "en cours"} res = _appel(client, jeton, 3, "search_pages", {"query": "cadrage"}) - assert [p["title"] for p in res] == ["Page de cadrage"] + # ACC-5.1 : la recherche est plein texte et couvre aussi les pages + # classiques — on vérifie la ligne de collection, sa famille et son + # URL profonde (§7.2 : l'URL est systématique). + lignes = [p for p in res if p["kind"] == "collection_page"] + assert [p["title"] for p in lignes] == ["Page de cadrage"] + assert all(p["url"].endswith(f"/collections/{donnees}") for p in lignes) + assert all("url" in p and "kind" in p for p in res) res = _appel(client, jeton, 4, "get_collection", {"id": 99999}) assert "introuvable" in res["erreur"] @@ -237,3 +244,160 @@ def test_mcp_info_get(client, jeton): assert "create_page" in infos["tools"] assert "update_page" in infos["tools"] assert "create_collection" in infos["tools"] + # ACC-5.1 : le catalogue distingue lectures et écritures. + assert "list_projects" in infos["read_tools"] + assert "read_page" in infos["read_tools"] + assert infos["write_tools"] == ["create_page", "update_page", + "create_collection"] + + +# ── ACC-5.1 : catalogue de lecture §7.2 ──────────────────────────────────── + + +@pytest.fixture() +def projet_taches(client): + """Un workspace + une base de tâches configurée (mapping complet).""" + with get_conn() as conn: + conn.execute("INSERT OR IGNORE INTO users (id, login) VALUES (1, 'tester')") + cur = conn.execute( + "INSERT INTO workspaces (name, owner_id) VALUES ('Atelier MCP', 1)") + wid = cur.lastrowid + cur = conn.execute( + "INSERT INTO collections (name, description, icon, workspace_id, is_task) " + "VALUES ('Suivi projet', 'Base de tâches', '🗂️', ?, 1)", (wid,)) + cid = cur.lastrowid + props = {} + for nom, typ, opts in ( + ("Assigné à", "person", "[]"), + ("Statut", "status", json.dumps( + [{"name": "À faire"}, {"name": "Terminé"}])), + ("Échéance", "date", "[]")): + cur = conn.execute( + "INSERT INTO collection_properties " + "(collection_id, name, prop_type, options_json) VALUES (?, ?, ?, ?)", + (cid, nom, typ, opts)) + props[nom] = cur.lastrowid + conn.execute( + "UPDATE collections SET task_assignee_prop=?, task_status_prop=?, " + "task_due_prop=? WHERE id=?", + (props["Assigné à"], props["Statut"], props["Échéance"], cid)) + conn.execute( + "INSERT INTO collection_pages (collection_id, title, icon, position, " + "property_values_json) VALUES (?, 'Relire le pont MCP', 'file', 1, ?)", + (cid, json.dumps({str(props["Statut"]): "À faire", + str(props["Échéance"]): "2026-10-30"}))) + conn.execute( + "INSERT INTO collection_pages (collection_id, title, icon, position, " + "property_values_json) VALUES (?, 'Brancher ObsiGate', 'file', 2, ?)", + (cid, json.dumps({str(props["Statut"]): "Terminé"}))) + conn.commit() + return {"workspace_id": wid, "collection_id": cid, "props": props} + + +def test_mcp_projets_catalogue(client, jeton, donnees, projet_taches): + projets = _appel(client, jeton, 90, "list_projects") + alpha = [p for p in projets if p["title"] == "Projet Alpha"] + assert alpha and alpha[0]["id"] == donnees + assert alpha[0]["url"].endswith(f"/collections/{donnees}") + assert alpha[0]["page_count"] == 2 + suivi = [p for p in projets if p["title"] == "Suivi projet"] + assert suivi and suivi[0]["is_task"] is True + + detail = _appel(client, jeton, 91, "get_project", {"id": donnees}) + assert detail["project"]["title"] == "Projet Alpha" + assert detail["project"]["url"].endswith(f"/collections/{donnees}") + # Détail sans contenu complet : titres seulement. + assert [p["title"] for p in detail["pages"]] == ["Page de cadrage", + "Revue hebdo"] + assert all(set(p) == {"id", "title"} for p in detail["pages"]) + res = _appel(client, jeton, 92, "get_project", {"id": 99999}) + assert "introuvable" in res["erreur"] + + +def test_mcp_list_project_tasks(client, jeton, projet_taches): + cid = projet_taches["collection_id"] + taches = _appel(client, jeton, 93, "list_project_tasks", + {"project_id": cid}) + assert [t["title"] for t in taches] == ["Relire le pont MCP", + "Brancher ObsiGate"] + assert taches[0]["status"] == "À faire" and taches[0]["done"] is False + assert taches[0]["due"] == "2026-10-30" + assert taches[1]["done"] is True + assert taches[0]["url"].endswith(f"/collections/{cid}") + toutes = _appel(client, jeton, 94, "list_project_tasks") + assert {t["title"] for t in toutes} >= {"Relire le pont MCP"} + res = _appel(client, jeton, 95, "list_project_tasks", + {"project_id": 99999}) + assert "introuvable" in res["erreur"] + + +def test_mcp_read_page_classique_et_troncature(client, jeton, projet_taches): + long_contenu = "mot " * 6000 # 24 000 caractères > borne 20 000 + with get_conn() as conn: + cur = conn.execute( + "INSERT INTO pages (workspace, workspace_id, title, content, " + "parent_section) VALUES ('Atelier MCP', ?, 'Grande note', ?, 'Private')", + (projet_taches["workspace_id"], long_contenu)) + pid = cur.lastrowid + conn.commit() + res = _appel(client, jeton, 96, "read_page", {"id": pid}) + assert res["kind"] == "page" and res["title"] == "Grande note" + assert res["truncated"] is True and len(res["content"]) <= 20000 + assert res["url"].endswith(f"/pages/{pid}") + res = _appel(client, jeton, 97, "read_page", {"id": 99999, "kind": "page"}) + assert "introuvable" in res["erreur"] + + +def test_mcp_read_page_ligne_de_collection(client, jeton, donnees): + with get_conn() as conn: + ligne = conn.execute( + "SELECT id FROM collection_pages WHERE collection_id=? AND " + "title='Page de cadrage'", (donnees,)).fetchone()["id"] + conn.execute( + "INSERT INTO pages (workspace, title, content, collection_row_id) " + "VALUES ('', 'contenu', 'Le contenu porté de la ligne', ?)", + (ligne,)) + conn.commit() + res = _appel(client, jeton, 98, "read_page", + {"id": ligne, "kind": "collection_page"}) + assert res["kind"] == "collection_page" + assert res["content"] == "Le contenu porté de la ligne" + assert res["truncated"] is False + assert res["properties"] == {"statut": "en cours"} + + +def test_mcp_jeton_lecture_seule_refuse_les_ecritures(client, jeton, donnees): + """§7.4 : un jeton au scope « read » (pont de lecture) ne peut pas + écrire via MCP, mais il peut toujours lire.""" + lecture = "fd_token_lecture_seule" + with get_conn() as conn: + conn.execute( + "INSERT INTO api_tokens (user_id, name, token_hash, token_prefix, " + "revoked, scopes) VALUES (1, 'pont-lecture', ?, ?, 0, 'read')", + (hashlib.sha256(lecture.encode()).hexdigest(), lecture[:12])) + conn.commit() + cols = _appel(client, lecture, 100, "list_projects") + assert any(c["title"] == "Projet Alpha" for c in cols) + r = _rpc(client, lecture, 101, "tools/call", + {"name": "create_page", + "arguments": {"collection_id": donnees, "title": "Interdite"}}) + assert r.status_code == 200 + erreur = r.json()["error"] + assert erreur["code"] == -32002 and "lecture seule" in erreur["message"] + with get_conn() as conn: + n = conn.execute( + "SELECT COUNT(*) FROM collection_pages WHERE title='Interdite'" + ).fetchone()[0] + assert n == 0 + + +def test_mcp_journalisation_des_appels(client, jeton, donnees): + """§7.4 : chaque tools/call laisse une trace dans api_audit_log.""" + _appel(client, jeton, 110, "list_projects") + with get_conn() as conn: + rows = conn.execute( + "SELECT action, user_id, token_id, detail FROM api_audit_log " + "WHERE action='mcp:list_projects'").fetchall() + assert rows, "aucune ligne d'audit pour l'appel MCP" + assert rows[-1]["user_id"] == 1 and rows[-1]["token_id"] is not None + assert "statut=ok" in rows[-1]["detail"] diff --git a/tests/test_fusion_pont_obsigate.py b/tests/test_fusion_pont_obsigate.py new file mode 100644 index 0000000..9ab0757 --- /dev/null +++ b/tests/test_fusion_pont_obsigate.py @@ -0,0 +1,320 @@ +"""Fusion ACC-2.2 — pont MCP Flowdeck → ObsiGate (plan « étape 1 : +lectures seules », tests T0–T5 du plan ``pont-mcp-etape1``). + +Ce que le code livre ici (la preuve réelle T0–T5 contre l'ObsiGate de +Bruno exige sa clé dédiée limitée à un vault + ses arbitrages Q-A/Q-B) : + +- **T0 (connectivité)** : le client MCP parle à un serveur **à session** + façon ObsiGate — ``initialize`` renvoie ``Mcp-Session-Id``, tout + appel sans cet en-tête est refusé ; le handshake précède aussi + ``tools/call``. Réponses SSE multi-messages décodées correctement. +- **Preset « obsigate »** : URL ``/mcp`` dérivée de la + configuration (dev/prod), clé chiffrée jamais renvoyée. +- **Garde-fou n° 2** : les outils MCP sont **classés** à la découverte + (annotations ``readOnlyHint``, sinon heuristique de nom) — les + écrivants (``propose_*``…) passent par les portes existantes + (politique d'approbation + rôles) ; ``apply_*`` est **bloqué** côté + client (patron §7.5 : l'écriture croisée s'approuve dans l'app + propriétaire). +""" + +import asyncio +import json + +import httpx +import pytest + +from app.db import get_conn +from app.services import connectors, http_client, mcp_client +from app.services.importers import url_fetch +from app.services.tool_registry import ToolRegistry + +SESSION_ID = "sess-og-test-123" +URL_PONT = "https://og.dracodev.net/mcp" + + +@pytest.fixture(autouse=True) +def _dns_monde_reel(monkeypatch): + """Même neutralisation DNS que test_fusion_tiktak (hôtes publics + résolus en 198.18.0.0/15 localement, refusés par la garde SSRF).""" + monkeypatch.setattr(url_fetch, "_validate_url", lambda url: url.strip()) + monkeypatch.setattr(mcp_client, "_sessions", {}) + + +# Outils tels qu'ObsiGate les déclare sur son serveur MCP : lectures, +# puis le patron d'écriture propose/apply (spec d'accès centralisé §7.5). +OUTILS_OBSIGATE = [ + {"name": "list_vaults", + "description": "Liste les vaults accessibles au jeton.", + "inputSchema": {"type": "object", "properties": {}}}, + {"name": "search_fulltext", + "description": "Recherche plein texte dans le vault du jeton.", + "inputSchema": {"type": "object", + "properties": {"query": {"type": "string"}}, + "required": ["query"]}, + "annotations": {"readOnlyHint": True}}, + {"name": "read_file", + "description": "Lit un fichier du vault.", + "inputSchema": {"type": "object", + "properties": {"path": {"type": "string"}}, + "required": ["path"]}}, + {"name": "propose_write_file", + "description": "Propose une écriture (brouillon à approuver).", + "inputSchema": {"type": "object", + "properties": {"path": {"type": "string"}, + "content": {"type": "string"}}, + "required": ["path", "content"]}, + "annotations": {"readOnlyHint": False}}, + {"name": "apply_write_file", + "description": "Applique une écriture approuvée.", + "inputSchema": {"type": "object", + "properties": {"draft_id": {"type": "string"}}, + "required": ["draft_id"]}}, +] + + +def _gestionnaire_obsigate(appels): + """Rejoue le serveur MCP d'ObsiGate : Streamable HTTP **à session + stricte** (tout appel post-initialize sans ``Mcp-Session-Id`` + valide → erreur -32000), ``tools/list`` répondu en SSE + multi-messages (une notification précède la réponse).""" + def handler(request: httpx.Request) -> httpx.Response: + body = json.loads(request.content or b"{}") + method = body.get("method") + appels.append((method, dict(request.headers))) + if method == "initialize": + return httpx.Response( + 200, + json={"jsonrpc": "2.0", "id": body.get("id"), + "result": {"protocolVersion": "2025-03-26", + "capabilities": {"tools": {}}, + "serverInfo": {"name": "obsigate", + "version": "2.78"}}}, + headers={"Mcp-Session-Id": SESSION_ID}) + if request.headers.get("mcp-session-id") != SESSION_ID: + return httpx.Response( + 400, + json={"jsonrpc": "2.0", "id": body.get("id"), + "error": {"code": -32000, + "message": "Mcp-Session-Id manquant ou invalide"}}) + if method == "notifications/initialized": + return httpx.Response(202, content=b"") + if method == "tools/list": + sse = ( + 'event: message\n' + 'data: {"jsonrpc": "2.0", "method": "notifications/progress",' + ' "params": {"message": "scan des vaults"}}\n\n' + 'event: message\n' + f'data: {json.dumps({"jsonrpc": "2.0", "id": body.get("id"), "result": {"tools": OUTILS_OBSIGATE}})}\n\n' + ) + return httpx.Response(200, content=sse.encode(), + headers={"Content-Type": "text/event-stream"}) + if method == "tools/call": + nom = (body.get("params") or {}).get("name") + if nom == "search_fulltext": + res = [{"path": "Notes/Pont MCP.md", "vault": "home-bruno", + "extraits": ["le pont MCP relie Flowdeck à ObsiGate"]}] + elif nom == "read_file": + res = {"path": "Notes/Pont MCP.md", + "content": "Contenu du fichier Pont MCP."} + elif nom == "list_vaults": + res = [{"name": "home-bruno"}] + else: + res = {"ok": True} + return httpx.Response( + 200, json={"jsonrpc": "2.0", "id": body.get("id"), + "result": {"content": [ + {"type": "text", + "text": json.dumps(res, ensure_ascii=False)}]}}) + return httpx.Response(200, json={"jsonrpc": "2.0", + "id": body.get("id"), "result": {}}) + return handler + + +@pytest.fixture +def faux_obsigate(monkeypatch): + """Branche le vrai ``_rpc`` du client sur le faux serveur (transport + httpx simulé) : sessions, en-têtes et SSE sont exercés pour de vrai.""" + appels: list = [] + transport = httpx.MockTransport(_gestionnaire_obsigate(appels)) + vraie_classe = http_client.shared_client + + class FauxShared: + def __init__(self, **kwargs): + kwargs["transport"] = transport + self._inner = vraie_classe(**kwargs) + + async def __aenter__(self): + return await self._inner.__aenter__() + + async def __aexit__(self, *exc): + return await self._inner.__aexit__(*exc) + + monkeypatch.setattr(http_client, "shared_client", FauxShared) + return appels + + +def _creer_obsigate(client, **kw): + body = {"name": "ObsiGate", "kind": "obsigate", "url": URL_PONT, + "secret": "og_cle_pont_test", "auth": "bearer", **kw} + r = client.post("/api/agent/connectors", json=body) + assert r.status_code == 200, r.text + return r.json() + + +def _sonder(client): + row = _creer_obsigate(client) + res = asyncio.run(connectors.probe(str(row["id"]))) + assert res["status"] == "ok", res + return row, res + + +# ── Preset ─────────────────────────────────────────────────────────────── + +def test_preset_obsigate_et_secret_chiffre(client): + row = _creer_obsigate(client) + assert row["kind"] == "obsigate" and row["auth"] == "bearer" + assert row["has_secret"] is True + assert "og_cle_pont_test" not in json.dumps(row) # jamais renvoyée + # URL par défaut = défaut officiel ; en dev, draco_urls la dérive + # en og-dev (résolution par configuration, chantier dev/prod). + assert connectors.PRESETS["obsigate"]["url"] == "https://og.dracodev.net/mcp" + hdrs = connectors._headers(row["id"]) + assert hdrs["Authorization"] == "Bearer og_cle_pont_test" + + +# ── T0 : session + handshake + SSE ─────────────────────────────────────── + +def test_t0_session_handshake_et_cache(client, faux_obsigate): + row, res = _sonder(client) + assert "5 outil(s)" in res["detail"] + methodes = [m for m, _ in faux_obsigate] + assert methodes[:3] == ["initialize", "notifications/initialized", "tools/list"] + # L'appel tools/list porte la session émise à initialize — sans + # elle, le faux serveur refuse (session stricte) et le probe échoue. + entetes_list = [h for m, h in faux_obsigate if m == "tools/list"][0] + assert entetes_list.get("mcp-session-id") == SESSION_ID + assert entetes_list.get("authorization") == "Bearer og_cle_pont_test" + + with get_conn() as conn: + brut = conn.execute( + "SELECT tools_json FROM agent_connectors WHERE id=?", + (row["id"],)).fetchone()["tools_json"] + outils = {t["original"]: t for t in json.loads(brut)} + assert set(outils) == {"list_vaults", "search_fulltext", "read_file", + "propose_write_file", "apply_write_file"} + # Classification à la découverte (garde-fou n° 2). + assert outils["search_fulltext"]["write"] is False # annotation readOnlyHint + assert outils["read_file"]["write"] is False # défaut = lecture + assert outils["propose_write_file"]["write"] is True # annotation + préfixe + assert outils["apply_write_file"]["write"] is True # préfixe apply_ + + +def test_t1_appel_outil_lecture_avec_session(client, faux_obsigate): + _sonder(client) + reg = ToolRegistry() + noms = {t["name"] for t in reg.schema()} + assert "mcp_obsigate_search_fulltext" in noms + assert "mcp_obsigate_read_file" in noms + + avant = len(faux_obsigate) + res = asyncio.run(reg.execute("mcp_obsigate_search_fulltext", + {"query": "pont MCP"})) + assert res.status == "success", res + assert "Pont MCP.md" in res.data["text"] + # call_tool refait le handshake complet, session réémise partout. + sequence = [m for m, _ in faux_obsigate[avant:]] + assert sequence == ["initialize", "notifications/initialized", "tools/call"] + entetes_call = [h for m, h in faux_obsigate if m == "tools/call"][0] + assert entetes_call.get("mcp-session-id") == SESSION_ID + + res2 = asyncio.run(reg.execute("mcp_obsigate_read_file", + {"path": "Notes/Pont MCP.md"})) + assert res2.status == "success" + assert "Contenu du fichier Pont MCP." in res2.data["text"] + + +def test_parse_body_sse_multi_messages(): + sse = ( + ': keep-alive\n' + 'event: message\n' + 'data: {"jsonrpc": "2.0", "method": "notifications/progress", "params": {}}\n\n' + 'event: message\n' + 'data: {"jsonrpc": "2.0", "id": 2, "result": {"tools": []}}\n\n' + ) + # Sans id attendu : le message porteur de result gagne sur la notification. + assert mcp_client.parse_body("text/event-stream", sse, 200) == { + "jsonrpc": "2.0", "id": 2, "result": {"tools": []}} + # Avec id attendu : c'est lui qui tranche. + assert mcp_client.parse_body("text/event-stream", sse, 200, + expected_id=2)["result"] == {"tools": []} + + +def test_classer_outil(): + assert mcp_client.classer_outil("search_fulltext") is False + assert mcp_client.classer_outil("read_file", {"readOnlyHint": True}) is False + assert mcp_client.classer_outil("propose_write_file") is True + assert mcp_client.classer_outil("apply_write_file") is True + assert mcp_client.classer_outil("create_note") is True + # L'annotation du serveur fait foi, même contre l'heuristique. + assert mcp_client.classer_outil("create_note", {"readOnlyHint": True}) is False + assert mcp_client.classer_outil("get_stats", {"readOnlyHint": False}) is True + + +# ── Garde-fous : portes de gouvernance + blocage apply_* ──────────────── + +def test_apply_bloque_cote_client(client, faux_obsigate): + _sonder(client) + reg = ToolRegistry() + res = asyncio.run(reg.execute("mcp_obsigate_apply_write_file", + {"draft_id": "b-1"})) + assert res.status == "error" + assert "bloqué" in res.message + # Aucun tools/call n'a quitté le client pour cet outil. + assert [m for m, _ in faux_obsigate].count("tools/call") == 0 + + +def test_portes_gouvernance_outils_mcp(client, faux_obsigate): + from app.services.agent_policies import check_tool + from app.services.permission_manager import PermissionManager, is_write_tool + + row, _res = _sonder(client) + assert row["id"] + assert is_write_tool("mcp_obsigate_propose_write_file") is True + assert is_write_tool("mcp_obsigate_apply_write_file") is True + assert is_write_tool("mcp_obsigate_search_fulltext") is False + assert is_write_tool("create_page") is True # interne, inchangé + assert is_write_tool("search") is False + + # Politique globale « approbation requise » : l'écrivant MCP est + # intercepté comme un écrivant interne, la lecture passe. + with get_conn() as conn: + conn.execute( + "INSERT INTO agent_policies (workspace_id, require_approval)" + " VALUES (NULL, 1)") + conn.commit() + gate = check_tool(1, None, "mcp_obsigate_propose_write_file", + is_write=is_write_tool("mcp_obsigate_propose_write_file")) + assert gate["allowed"] is False and gate["approval_id"] + ok = check_tool(1, None, "mcp_obsigate_search_fulltext", + is_write=is_write_tool("mcp_obsigate_search_fulltext")) + assert ok["allowed"] is True + + # Rôles : un viewer ne peut pas lancer un outil MCP écrivant. + from fastapi import HTTPException + with get_conn() as conn: + admin = conn.execute("SELECT id FROM users ORDER BY id LIMIT 1").fetchone() + conn.execute("INSERT INTO users (login, is_admin) VALUES ('lecteur_pont', 0)") + lecteur = conn.execute( + "SELECT id FROM users WHERE login='lecteur_pont'").fetchone() + conn.execute("INSERT INTO workspaces (name, owner_id) VALUES ('ws-pont', ?)", + (admin["id"],)) + ws = conn.execute("SELECT id FROM workspaces WHERE name='ws-pont'").fetchone() + conn.execute( + "INSERT INTO workspace_members (workspace_id, user_id, role)" + " VALUES (?,?, 'viewer')", (ws["id"], lecteur["id"])) + conn.commit() + pm = PermissionManager(lecteur["id"]) + with pytest.raises(HTTPException): + pm.assert_can("mcp_obsigate_propose_write_file", {}, ws["id"]) + pm.assert_can("mcp_obsigate_search_fulltext", {}, ws["id"])