From ae290b5421d17e9c682ba0bec42940e4a9e17027 Mon Sep 17 00:00:00 2001 From: stroblme Date: Tue, 6 Oct 2026 12:16:48 +0200 Subject: [PATCH] Quote every path parameter the MCP tools put into a URL httpx resolves `..` before sending, so save_flow("../ai/config") reached PUT /ai/config. Each segment is now percent-encoded, dots included. --- backend/fluksio/mcp/server.py | 52 +++++++++++++++++++---------- backend/tests/mcp/test_mcp_tools.py | 41 +++++++++++++++++++++++ 2 files changed, 76 insertions(+), 17 deletions(-) create mode 100644 backend/tests/mcp/test_mcp_tools.py diff --git a/backend/fluksio/mcp/server.py b/backend/fluksio/mcp/server.py index 540bde0..94432ce 100644 --- a/backend/fluksio/mcp/server.py +++ b/backend/fluksio/mcp/server.py @@ -15,6 +15,7 @@ from __future__ import annotations import logging from contextvars import ContextVar from typing import Any +from urllib.parse import quote import httpx from mcp.server.fastmcp import FastMCP @@ -85,6 +86,17 @@ async def forward_caller_auth(request: httpx.Request) -> None: request.headers["authorization"] = authorization +def _seg(value: str) -> str: + """*value* as one path segment, and nothing more. + + httpx resolves ``..`` before sending, so a name put into a path as written + could climb out of the route it was meant for: ``save_flow("../ai/config")`` + was ``PUT /ai/config``. ``quote`` covers the slashes but leaves dots alone, + hence encoding those too; the server decodes both. + """ + return quote(value, safe="").replace(".", "%2E") + + async def _call(method: str, path: str, **kwargs: Any) -> Any: """One REST call, with failures handed back as data rather than raised. @@ -129,7 +141,7 @@ async def get_flow(name: str) -> Any: The definition is the working copy — unpublished edits included — which is what to base a change on. """ - return await _call("GET", f"/flows/{name}") + return await _call("GET", f"/flows/{_seg(name)}") @mcp.tool() @@ -141,19 +153,19 @@ async def list_node_types() -> Any: @mcp.tool() async def get_node_source(name: str, node_id: str) -> Any: """The Python source of one node.""" - return await _call("GET", f"/flows/{name}/nodes/{node_id}/source") + return await _call("GET", f"/flows/{_seg(name)}/nodes/{_seg(node_id)}/source") @mcp.tool() async def get_flow_state(name: str) -> Any: """The last value seen on every message of a flow.""" - return await _call("GET", f"/flows/{name}/state") + return await _call("GET", f"/flows/{_seg(name)}/state") @mcp.tool() async def get_message_history(name: str, message: str) -> Any: """Recent numeric values of one message, oldest first.""" - return await _call("GET", f"/flows/{name}/history/{message}") + return await _call("GET", f"/flows/{_seg(name)}/history/{_seg(message)}") @mcp.tool() @@ -198,7 +210,7 @@ async def save_flow(name: str, definition: dict[str, Any]) -> Any: Nothing here reaches the engine until `publish_flow`. """ - return await _call("PUT", f"/flows/{name}", json=definition) + return await _call("PUT", f"/flows/{_seg(name)}", json=definition) @mcp.tool() @@ -209,32 +221,34 @@ async def save_node_source(name: str, node_id: str, code: str) -> Any: one per setting, and returns a dict keyed by output port. """ return await _call( - "PUT", f"/flows/{name}/nodes/{node_id}/source", json={"code": code} + "PUT", f"/flows/{_seg(name)}/nodes/{_seg(node_id)}/source", json={"code": code} ) @mcp.tool() async def publish_flow(name: str, version: int) -> Any: """Deploy a flow's unpublished changes: this is what puts them live.""" - return await _call("POST", f"/flows/{name}/publish", json={"version": version}) + return await _call( + "POST", f"/flows/{_seg(name)}/publish", json={"version": version} + ) @mcp.tool() async def discard_draft(name: str) -> Any: """Throw unpublished changes away and go back to what is running.""" - return await _call("POST", f"/flows/{name}/discard-draft") + return await _call("POST", f"/flows/{_seg(name)}/discard-draft") @mcp.tool() async def delete_flow(name: str) -> Any: """Delete a flow and the code of its nodes.""" - return await _call("DELETE", f"/flows/{name}") + return await _call("DELETE", f"/flows/{_seg(name)}") @mcp.tool() async def validate_flow(name: str) -> Any: """What would keep this flow from running: loops, unconnected inputs.""" - return await _call("POST", f"/flows/{name}/validate") + return await _call("POST", f"/flows/{_seg(name)}/validate") # ----------------------------------------------------------------------------- @@ -249,7 +263,9 @@ async def run_flow(name: str, inputs: dict[str, Any] | None = None) -> Any: With unpublished changes this runs the draft, so a change can be tried before it is published. """ - return await _call("POST", f"/flows/{name}/run", json={"inputs": inputs or {}}) + return await _call( + "POST", f"/flows/{_seg(name)}/run", json={"inputs": inputs or {}} + ) @mcp.tool() @@ -258,38 +274,40 @@ async def trigger_node( ) -> Any: """Feed values into one node and run everything downstream of it.""" return await _call( - "POST", f"/flows/{name}/nodes/{node_id}/trigger", json={"values": values or {}} + "POST", + f"/flows/{_seg(name)}/nodes/{_seg(node_id)}/trigger", + json={"values": values or {}}, ) @mcp.tool() async def start_flow(name: str) -> Any: """Let the engine run this flow: subscriptions, schedules and webhooks.""" - return await _call("POST", f"/flows/{name}/start") + return await _call("POST", f"/flows/{_seg(name)}/start") @mcp.tool() async def stop_flow(name: str) -> Any: """Take a flow off the engine. Nothing of it stays subscribed or scheduled.""" - return await _call("POST", f"/flows/{name}/stop") + return await _call("POST", f"/flows/{_seg(name)}/stop") @mcp.tool() async def pause_flow(name: str) -> Any: """Hold a flow's nodes while its incoming values keep arriving.""" - return await _call("POST", f"/flows/{name}/pause") + return await _call("POST", f"/flows/{_seg(name)}/pause") @mcp.tool() async def resume_flow(name: str) -> Any: """Let a paused flow carry on, running whatever was held back.""" - return await _call("POST", f"/flows/{name}/resume") + return await _call("POST", f"/flows/{_seg(name)}/resume") @mcp.tool() async def cancel_node(name: str, node_id: str) -> Any: """Stop a node's code while it is running. Nothing to stop is not an error.""" - return await _call("POST", f"/flows/{name}/nodes/{node_id}/cancel") + return await _call("POST", f"/flows/{_seg(name)}/nodes/{_seg(node_id)}/cancel") # ----------------------------------------------------------------------------- diff --git a/backend/tests/mcp/test_mcp_tools.py b/backend/tests/mcp/test_mcp_tools.py new file mode 100644 index 0000000..7d4b27c --- /dev/null +++ b/backend/tests/mcp/test_mcp_tools.py @@ -0,0 +1,41 @@ +"""What the MCP tools send, whatever an agent hands them.""" + +import asyncio + +import httpx + +from fluksio.mcp import server + + +def test_a_name_cannot_climb_out_of_its_route() -> None: + """httpx resolves ``..``, so ``save_flow("../ai/config")`` was ``PUT /ai/config``. + + A prompt-injected assistant acting for a superuser could repoint the + provider that way while the stored key stayed. + """ + seen: list[str] = [] + + def record(request: httpx.Request) -> httpx.Response: + seen.append(f"{request.method} {request.url.raw_path.decode()}") + return httpx.Response(200, json={}) + + async def calls() -> None: + previous = server._client + mock = httpx.AsyncClient( + transport=httpx.MockTransport(record), base_url="http://test" + ) + server.set_client(mock) + try: + await server.save_flow("../ai/config", {}) + await server.delete_flow("..") + await server.get_node_source("..", "../source") + finally: + server.set_client(previous) + await mock.aclose() + + asyncio.run(calls()) + assert seen == [ + "PUT /api/v1/flows/%2E%2E%2Fai%2Fconfig", + "DELETE /api/v1/flows/%2E%2E", + "GET /api/v1/flows/%2E%2E/nodes/%2E%2E%2Fsource/source", + ]