Serve MCP and hooks with a secret through the portal tunnel

The connector now serves /mcp, its OAuth documents and /hooks/ besides
/api/v1/, refuses any dot segment (httpx resolves them, which carried
/api/v1/../../mcp onto the MCP mount), addresses calls to the hub's host,
and bridges served sockets by calling the app with a websocket scope.
A hook without a secret answers the tunnel with the 404 (or 1008) of a
wrong secret. Requests through the hub get the hub's issuer, resource and
consent page in the OAuth metadata and the 401, and the MCP host check
admits the hub while enrolled.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WgBfkm4rCmtd1T4vHWjEPx

backend/tests/api/routes/test_oauth.py

@@ -80,6 +80,44 @@ def test_the_metadata_says_how_to_get_in(client: TestClient) -> None:
     }
 
 
+@pytest.mark.usefixtures("enrolled")
+def test_through_the_hub_the_metadata_and_resource_are_the_hubs(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    """An agent at /i/{id}/mcp on the hub is sent to the hub's addresses."""
+    from fluksio.cloud import config as cloud_config
+    from tests.utils.portal import INSTANCE_ID, ISSUER
+
+    via = {cloud_config.VIA_HEADER: cloud_config.VIA_PORTAL}
+    base = f"{ISSUER}/i/{INSTANCE_ID}"
+    document = client.get("/.well-known/oauth-authorization-server", headers=via).json()
+    assert document["issuer"] == base
+    assert document["token_endpoint"] == f"{base}{PREFIX}/token"
+    # No link is attached here to name the portal's page, so the hub's own.
+    assert document["authorization_endpoint"] == f"{base}/oauth/authorize"
+    # A request made here hears nothing about the hub.
+    local = client.get("/.well-known/oauth-authorization-server").json()
+    assert local["issuer"] == settings.oauth_issuer
+
+    _verifier, challenge = pkce()
+    body = {
+        "client_id": register(client)["client_id"],
+        "redirect_uri": REDIRECT,
+        "code_challenge": challenge,
+        "code_challenge_method": "S256",
+        "resource": f"{base}/mcp",
+    }
+    through = client.post(
+        f"{PREFIX}/authorize", headers={**superuser_token_headers, **via}, json=body
+    )
+    assert through.status_code == 200, through.text
+    # Only through the hub is that a resource this instance serves.
+    here = client.post(
+        f"{PREFIX}/authorize", headers=superuser_token_headers, json=body
+    )
+    assert here.json()["error"] == "invalid_target"
+
+
 def test_a_full_authorization_reaches_the_api(
     client: TestClient, superuser_token_headers: dict[str, str]
 ) -> None:

backend/tests/mcp/test_mcp_http.py

@@ -97,6 +97,55 @@ def test_the_resource_metadata_names_the_authorization_server(over_mcp) -> None:
     assert settings.oauth_issuer in listed
 
 
+@pytest.mark.usefixtures("enrolled")
+def test_through_the_hub_an_agent_is_shown_the_hubs_addresses(
+    over_mcp,
+    client: TestClient,
+    superuser_token_headers: dict[str, str],
+    agent: uuid.UUID,
+) -> None:
+    """What the connector sends: its marker, and the hub as the Host."""
+    from fluksio.cloud import config as cloud_config
+    from tests.utils.portal import INSTANCE_ID, ISSUER
+
+    via = {cloud_config.VIA_HEADER: cloud_config.VIA_PORTAL, "Host": "hub.example.test"}
+    base = f"{ISSUER}/i/{INSTANCE_ID}"
+    me = client.get(
+        f"{settings.API_V1_STR}/users/me", headers=superuser_token_headers
+    ).json()
+    token = mcp_token(uuid.UUID(me["id"]), agent)
+
+    async def block(http: httpx.AsyncClient) -> list[httpx.Response]:
+        anonymous = rpc(None, "tools/list")
+        signed = rpc(token, "tools/list")
+        return [
+            await http.post(
+                "/mcp", json=anonymous["json"], headers={**anonymous["headers"], **via}
+            ),
+            await http.get("/.well-known/oauth-protected-resource/mcp", headers=via),
+            await http.post(
+                "/mcp", json=signed["json"], headers={**signed["headers"], **via}
+            ),
+            # Any other name is still refused.
+            await http.post(
+                "/mcp",
+                json=signed["json"],
+                headers={**signed["headers"], "Host": "elsewhere.test"},
+            ),
+        ]
+
+    unauthenticated, document, listed, elsewhere = over_mcp(block)
+    assert unauthenticated.status_code == 401
+    assert (
+        f'resource_metadata="{ISSUER}/.well-known/oauth-protected-resource/i/{INSTANCE_ID}/mcp"'
+        in unauthenticated.headers["www-authenticate"]
+    )
+    assert document.json()["resource"] == f"{base}/mcp"
+    assert document.json()["authorization_servers"] == [base]
+    assert listed.status_code == 200
+    assert elsewhere.status_code == 421
+
+
 def test_a_browser_token_is_not_an_agent_token(
     over_mcp, superuser_token_headers: dict[str, str]
 ) -> None:

backend/tests/test_tunnel_reach.py

@@ -0,0 +1,144 @@
+"""What the portal tunnel reaches on this instance, and what it refuses.
+
+Requests are handed to the connector as the hub frames them. Besides the API,
+which the hub opens only to a portal session, the tunnel serves the MCP
+endpoint and the flows' hooks, and a hook only with a secret.
+"""
+
+import asyncio
+import base64
+import json
+from typing import Any
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from fluksio.cloud.connector import CloudConnector
+from fluksio.flow.controller import HOOK_PREFIX
+from fluksio.flow.messages import DType, MessageSpec
+from fluksio.flow.nodes import HttpNode, WebSocketNode
+from fluksio.flow.pipeline import Pipeline
+
+SECRET = "tunnel-key"
+PORT = [MessageSpec(name="temp", dtype=DType.FLOAT)]
+
+
+class _Socket:
+    """The connector's side of the tunnel, recording what it sends the hub."""
+
+    def __init__(self) -> None:
+        self.frames: list[dict[str, Any]] = []
+
+    async def send(self, text: str) -> None:
+        self.frames.append(json.loads(text))
+
+
+def _through_the_tunnel(app: FastAPI, path: str, method: str = "POST") -> int | None:
+    """The status a call through the hub gets, or None when it is refused."""
+    socket = _Socket()
+    body = base64.b64encode(json.dumps({"temp": 21.5}).encode()).decode()
+    frame = {
+        "id": "c1",
+        "method": method,
+        "path": path,
+        "query": "",
+        "headers": {"content-type": "application/json"},
+        "body": body,
+    }
+    asyncio.run(CloudConnector(app)._serve(socket, frame))
+    head = socket.frames[0]
+    return None if head["op"] == "err" else int(head["status"])
+
+
+def _hooks(*nodes: HttpNode | WebSocketNode) -> tuple[FastAPI, Pipeline]:
+    """Mount nodes the way the controller does, behind an app at the root."""
+    pipeline = Pipeline(nodes=list(nodes))
+    app = FastAPI()
+    app.mount("/", FastAPI())  # the MCP catch-all
+    for node in nodes:
+        node.url = f"{HOOK_PREFIX}/house/{node.url.lstrip('/')}"
+        node.register_route(app)
+    return app, pipeline
+
+
+def _webhook(name: str, **params: object) -> HttpNode:
+    node = HttpNode(provides=PORT, params={"url": f"/{name}", **params}, name=name)
+    node.assign_flow("house", name)
+    return node
+
+
+def _socket(name: str, **params: object) -> WebSocketNode:
+    node = WebSocketNode(provides=PORT, params={"url": f"/{name}", **params}, name=name)
+    node.assign_flow("house", name)
+    return node
+
+
+def test_a_webhook_answers_the_tunnel_only_with_its_secret() -> None:
+    app, pipeline = _hooks(_webhook("open"), _webhook("guarded", secret=SECRET))
+
+    # On this network a hook without a secret stays open, as it always was.
+    assert TestClient(app).post("/hooks/house/open", json={}).status_code == 200
+    # Through the hub it looks like a wrong secret does.
+    assert _through_the_tunnel(app, "/hooks/house/open") == 404
+    assert _through_the_tunnel(app, "/hooks/house/guarded/wrong") == 404
+    assert "house.temp" not in pipeline.values()
+
+    assert _through_the_tunnel(app, f"/hooks/house/guarded/{SECRET}") == 200
+    assert pipeline.values()["house.temp"]["value"] == 21.5
+
+
+@pytest.mark.parametrize(
+    "path",
+    [
+        # httpx resolves a literal dot segment, which would carry a path the hub
+        # forwarded without a session onto the API.
+        "/hooks/../api/v1/login/access-token",
+        "/api/v1/../../mcp",
+        "/hooks/%2e%2e/api/v1/utils/health-check/",
+        "/docs",
+        "/.well-known/openid-configuration",
+        "/mcpx",
+    ],
+)
+def test_the_tunnel_refuses_what_it_does_not_serve(path: str) -> None:
+    assert _through_the_tunnel(FastAPI(), path, method="GET") is None
+
+
+def test_a_served_socket_answers_the_tunnel_only_with_its_secret() -> None:
+    app, pipeline = _hooks(_socket("open"), _socket("guarded", secret=SECRET))
+
+    async def scenario() -> list[list[dict[str, Any]]]:
+        connector = CloudConnector(app)
+        refused = []
+        for stream, path in [
+            ("w1", "/hooks/house/open"),
+            ("w2", "/hooks/house/guarded/wrong"),
+            ("w3", "/hooks/house/nothing"),
+        ]:
+            socket = _Socket()
+            await connector._stream(socket, {"id": stream, "path": path, "query": ""})
+            refused.append(socket.frames)
+
+        socket = _Socket()
+        frame = {"id": "w4", "path": f"/hooks/house/guarded/{SECRET}", "query": ""}
+        task = asyncio.create_task(connector._stream(socket, frame))
+        while not socket.frames:
+            await asyncio.sleep(0.01)
+        assert socket.frames == [{"op": "ws_open_ok", "id": "w4"}]
+        # What the session loop does with the hub's ws_msg.
+        connector._inboxes["w4"].put_nowait(
+            {"type": "websocket.receive", "text": '{"temp": 21.5}'}
+        )
+        while "house.temp" not in pipeline.values():
+            await asyncio.sleep(0.01)
+        task.cancel()
+        return refused
+
+    refused = asyncio.run(scenario())
+    # Refused before opening, every way alike.
+    assert refused == [
+        [{"op": "ws_close", "id": stream, "code": 1008}]
+        for stream in ("w1", "w2", "w3")
+    ]
+    assert pipeline.values()["house.temp"]["value"] == 21.5