diff --git a/backend/fluksio/api/routes/oauth.py b/backend/fluksio/api/routes/oauth.py index fe4b797..b5df695 100644 --- a/backend/fluksio/api/routes/oauth.py +++ b/backend/fluksio/api/routes/oauth.py @@ -42,6 +42,7 @@ from fluksio.api.deps import ( get_current_active_superuser, get_current_user, ) +from fluksio.cloud import config as cloud_config from fluksio.core import security from fluksio.core.config import settings from fluksio.models import ( @@ -301,13 +302,22 @@ def authorize_validate( @router.post("/authorize", response_model=OAuthAuthorizeResponse) def authorize( + request: Request, body: OAuthAuthorizeRequest, current_user: CurrentUser, session: SessionDep, ) -> Any: - """Approve a client, on behalf of the signed-in user.""" + """Approve a client, on behalf of the signed-in user. + + Through the hub the endpoint is the one there, which is the resource an + agent that came that way asks for. + """ if not settings.MCP_ENABLED: return _error("access_denied", "The MCP endpoint is switched off.", 403) + resources = {settings.mcp_resource} + base = cloud_config.portal_base(request.headers) + if base is not None: + resources.add(f"{base}/mcp") client = _load_client(session, body.client_id) if client is None: @@ -318,7 +328,7 @@ def authorize( return _error("invalid_request", "PKCE must use S256.") if not _PKCE_RE.match(body.code_challenge): return _error("invalid_request", "Malformed code challenge.") - if body.resource and body.resource.rstrip("/") != settings.mcp_resource: + if body.resource and body.resource.rstrip("/") not in resources: return _error("invalid_target", "Unknown resource.") code = secrets.token_urlsafe(32) diff --git a/backend/fluksio/cloud/config.py b/backend/fluksio/cloud/config.py index 34d0dce..d561014 100644 --- a/backend/fluksio/cloud/config.py +++ b/backend/fluksio/cloud/config.py @@ -20,6 +20,7 @@ import json import logging import os import secrets +from collections.abc import Mapping from dataclasses import dataclass from typing import Any @@ -114,6 +115,23 @@ def delete() -> None: _cache = None +def via_portal(headers: Mapping[str, str]) -> bool: + """Whether the connector replayed this request off the tunnel.""" + return secrets.compare_digest(headers.get(VIA_HEADER, ""), VIA_PORTAL) + + +def portal_base(headers: Mapping[str, str]) -> str | None: + """Where this instance answers at the hub, for a request that came through it. + + The hub serves an instance under ``/i/{id}`` on the host it signs as, the + issuer enrolment pinned. ``None`` for a request made here. + """ + config = _read() + if config is None or not via_portal(headers): + return None + return f"{config.issuer.rstrip('/')}/i/{config.instance_id}" + + def decode_portal_token(token: str) -> dict[str, Any]: """Validate a token the portal minted for this instance. diff --git a/backend/fluksio/cloud/connector.py b/backend/fluksio/cloud/connector.py index 10473c9..e981614 100644 --- a/backend/fluksio/cloud/connector.py +++ b/backend/fluksio/cloud/connector.py @@ -22,6 +22,7 @@ import logging import random import time import uuid +from collections.abc import MutableMapping from dataclasses import replace from datetime import timedelta from typing import Any @@ -53,9 +54,20 @@ PING_INTERVAL_S = 20.0 PING_TIMEOUT_S = 20.0 CHUNK_BYTES = 256 * 1024 MAX_WS_MESSAGE = 1024 * 1024 -#: Only the versioned API is served over the tunnel. The MCP mount and the -#: OAuth endpoints live outside it and stay local-only. -ALLOWED_PREFIX = "/api/v1/" +#: What the tunnel serves: the versioned API, which the hub opens only to a +#: portal session; the MCP endpoint and the OAuth documents leading an agent +#: to it, which take the agent's own token; and the flows' hooks, of which +#: only those with a secret answer a request from the tunnel. Everything else, +#: the API docs included, stays on this machine. +ALLOWED_PREFIXES = ("/api/v1/", "/hooks/") +ALLOWED_PATHS = frozenset( + { + "/mcp", + "/.well-known/oauth-protected-resource/mcp", + "/.well-known/oauth-authorization-server", + "/.well-known/jwks.json", + } +) #: Proxied calls, and bridged streams, this instance will run at once. #: Neither dict was bounded, and an id the hub reused silently dropped the #: reference to a task that was still running. @@ -143,6 +155,11 @@ class CloudConnector: self._revoked = False #: The portal account that owns this instance, as the last welcome said. self._owner: str | None = None + #: Where the portal opens this instance, as the last welcome said. An + #: older hub does not say. + self.app_url: str | None = None + #: What the hub sent each bridged hook socket, waiting to be received. + self._inboxes: dict[str, asyncio.Queue[dict[str, Any]]] = {} @property def revoked(self) -> bool: @@ -240,6 +257,7 @@ class CloudConnector: ) self._owner = welcome.get("owner") + self.app_url = welcome.get("app_url") self._adopt_owner(config, self._owner) self._adopt_jwks(welcome.get("jwks")) @@ -260,6 +278,15 @@ class CloudConnector: self._cancel(str(frame.get("id") or "")) elif op == "ws_open": self._start_stream(socket, frame) + elif op == "ws_msg": + inbox = self._inboxes.get(str(frame.get("id") or "")) + if inbox is not None: + inbox.put_nowait( + { + "type": "websocket.receive", + "text": str(frame.get("text") or ""), + } + ) elif op == "ws_close": self._cancel_stream(str(frame.get("id") or "")) elif op == "member_removed": @@ -506,10 +533,10 @@ class CloudConnector: call_id = str(frame["id"]) path = str(frame.get("path") or "") try: - if not path.startswith(ALLOWED_PREFIX): - # Defence in depth: the hub only ever forwards API paths, and - # an instance that trusted anything else would be an SSRF - # gadget aimed at its own process. + if not _allowed(path): + # Defence in depth: the hub forwards nothing else, and an + # instance that trusted it to would be an SSRF gadget aimed at + # its own process. await socket.send( _dump({"op": "err", "id": call_id, "message": "path not allowed"}) ) @@ -525,6 +552,9 @@ class CloudConnector: # is the only thing that makes it true, and the value is this # process's own so nothing off the network can imitate it. headers[cloud_config.VIA_HEADER] = cloud_config.VIA_PORTAL + # Addressed as the caller addressed it, which is the hub. The MCP + # endpoint checks the Host against the names it answers to. + headers["host"] = _hub_host() async with self._client() as client: request = client.build_request( @@ -619,13 +649,16 @@ class CloudConnector: stream_id = str(frame["id"]) path = str(frame.get("path") or "") query = str(frame.get("query") or "") + if _allowed(path) and path.startswith("/hooks/"): + await self._hook_stream(socket, stream_id, path, query) + return # The browser's ticket when it brought one; the hub restates the # session it checked as `token` either way, which is the tunnel's own # channel and stays when the public `?token=` goes. ticket = _query_value(query, "ticket") token = redeem_ticket(ticket) if ticket else _query_value(query, "token") - if not path.startswith(ALLOWED_PREFIX) or not path.endswith("/flows/ws"): + if not path.startswith("/api/v1/") or not path.endswith("/flows/ws"): await socket.send(_dump({"op": "ws_close", "id": stream_id, "code": 1008})) return with Session(engine) as session: @@ -687,12 +720,101 @@ class CloudConnector: _dump({"op": "ws_close", "id": stream_id, "code": 1011}) ) + async def _hook_stream( + self, socket: Any, stream_id: str, path: str, query: str + ) -> None: + """Serve a flow's socket over the tunnel, as uvicorn would serve it here. + + The app is called with a websocket scope of its own, so the route a hook + installed answers exactly as it does a local client, secret included. + """ + inbox: asyncio.Queue[dict[str, Any]] = asyncio.Queue() + inbox.put_nowait({"type": "websocket.connect"}) + self._inboxes[stream_id] = inbox + state = {"accepted": False, "closed": False} + + async def send(message: MutableMapping[str, Any]) -> None: + kind = message["type"] + if kind == "websocket.accept": + state["accepted"] = True + await socket.send(_dump({"op": "ws_open_ok", "id": stream_id})) + elif kind == "websocket.send": + text = message.get("text") + if text is None: + text = (message.get("bytes") or b"").decode(errors="replace") + await socket.send( + _dump({"op": "ws_msg", "id": stream_id, "text": text}) + ) + elif kind == "websocket.close": + state["closed"] = True + # Refused before it was accepted is a refusal whatever the code, + # so no route, a wrong secret and no secret all look alike. + code = message.get("code", 1000) if state["accepted"] else 1008 + await socket.send( + _dump({"op": "ws_close", "id": stream_id, "code": code}) + ) + + scope = { + "type": "websocket", + "asgi": {"version": "3.0"}, + "http_version": "1.1", + "scheme": "ws", + "path": path, + "raw_path": path.encode(), + "root_path": "", + "query_string": query.encode(), + "headers": [ + (b"host", _hub_host().encode()), + (cloud_config.VIA_HEADER.encode(), cloud_config.VIA_PORTAL.encode()), + ], + "client": ("127.0.0.1", 0), + "server": None, + "subprotocols": [], + } + code = 1000 + try: + await self._app(scope, inbox.get, send) + except asyncio.CancelledError: + raise + except Exception as exc: + logger.debug("Hook stream %s ended: %s", stream_id, exc) + code = 1011 + finally: + self._inboxes.pop(stream_id, None) + if not state["closed"]: + with contextlib.suppress(Exception): + await socket.send( + _dump({"op": "ws_close", "id": stream_id, "code": code}) + ) + def _cancel_stream(self, stream_id: str) -> None: task = self._streams.pop(stream_id, None) if task is not None: task.cancel() +def _allowed(path: str) -> bool: + """Whether the tunnel may serve this path. + + A dot segment is refused, literal or encoded: httpx resolves a literal one + before dispatching, so ``/hooks/../api/v1/…`` would otherwise reach the API + as a path the hub forwarded without a session. + """ + from urllib.parse import unquote + + if {".", ".."} & set(unquote(path).split("/")): + return False + return path in ALLOWED_PATHS or path.startswith(ALLOWED_PREFIXES) + + +def _hub_host() -> str: + """The host a caller addressed through the tunnel, the hub's own.""" + from urllib.parse import urlsplit + + config = cloud_config.load() + return urlsplit(config.issuer).netloc if config is not None else "instance.local" + + def _is_revoked(exc: Exception) -> bool: """Whether the hub closed the link as revoked, at the handshake or later.""" from websockets.exceptions import ConnectionClosed diff --git a/backend/fluksio/flow/nodes/http.py b/backend/fluksio/flow/nodes/http.py index 9ceba3d..0877cf1 100644 --- a/backend/fluksio/flow/nodes/http.py +++ b/backend/fluksio/flow/nodes/http.py @@ -400,11 +400,18 @@ class HttpNode(Node): :returns: JSON response with trigger result. :rtype: JSONResponse """ - if self.secret and not hmac.compare_digest( - str(request.path_params.get("secret", "")).encode(), - self.secret.encode(), - ): - # A wrong secret must look like no such hook at all. + from fluksio.cloud.config import via_portal + + if ( + self.secret + and not hmac.compare_digest( + str(request.path_params.get("secret", "")).encode(), + self.secret.encode(), + ) + ) or (not self.secret and via_portal(request.headers)): + # A wrong secret must look like no such hook at all, and so must + # a hook without one reached through the portal, which would + # otherwise be open to the whole internet. return JSONResponse(content={"detail": "Not Found"}, status_code=404) try: diff --git a/backend/fluksio/flow/nodes/websocket.py b/backend/fluksio/flow/nodes/websocket.py index 6593758..3d3b796 100644 --- a/backend/fluksio/flow/nodes/websocket.py +++ b/backend/fluksio/flow/nodes/websocket.py @@ -342,11 +342,17 @@ class WebSocketNode(Node): async def _serve(self, websocket: WebSocket) -> None: """One client of a served socket, for as long as it stays.""" - if self.secret and not hmac.compare_digest( - str(websocket.path_params.get("secret", "")).encode(), - self.secret.encode(), - ): + from fluksio.cloud.config import via_portal + + if ( + self.secret + and not hmac.compare_digest( + str(websocket.path_params.get("secret", "")).encode(), + self.secret.encode(), + ) + ) or (not self.secret and via_portal(websocket.headers)): # Closed before accepting: the same 403 as a path with no route. + # Through the portal a socket needs a secret, as a webhook does. await websocket.close(code=1008) return await websocket.accept() diff --git a/backend/fluksio/main.py b/backend/fluksio/main.py index bb0836b..dfc87c8 100644 --- a/backend/fluksio/main.py +++ b/backend/fluksio/main.py @@ -540,19 +540,27 @@ async def rebuild_busy(request: Request, exc: RebuildBusy) -> JSONResponse: # n include_in_schema=False, tags=["oauth"], ) -def oauth_authorization_server() -> JSONResponse: +def oauth_authorization_server(request: Request) -> JSONResponse: """RFC 8414 metadata, so an agent can find its way in unaided. The authorization endpoint is the dashboard rather than the API: approving a client needs a signed-in human, and the browser session lives there. + Through the hub that is the dashboard the portal opens, which the hub names + when the link attaches; the hub's own address is the guess an older one + leaves. """ - issuer = settings.oauth_issuer + from fluksio.cloud import config as cloud_config + + base = cloud_config.portal_base(request.headers) + issuer = base or settings.oauth_issuer + dashboard = settings.FRONTEND_HOST.rstrip("/") + if base is not None: + connector = getattr(request.app.state, "cloud_connector", None) + dashboard = getattr(connector, "app_url", None) or base return JSONResponse( content={ "issuer": issuer, - "authorization_endpoint": ( - f"{settings.FRONTEND_HOST.rstrip('/')}/oauth/authorize" - ), + "authorization_endpoint": f"{dashboard}/oauth/authorize", "token_endpoint": f"{issuer}{settings.API_V1_STR}/oauth/token", "registration_endpoint": f"{issuer}{settings.API_V1_STR}/oauth/register", "jwks_uri": f"{issuer}/.well-known/jwks.json", diff --git a/backend/fluksio/mcp/http.py b/backend/fluksio/mcp/http.py index 3e1cc22..265062c 100644 --- a/backend/fluksio/mcp/http.py +++ b/backend/fluksio/mcp/http.py @@ -4,6 +4,9 @@ The transport is streamable HTTP, stateless, one JSON response per call: no sticky sessions to route and nothing for a proxy to buffer. The SDK serves both ``/mcp`` and the protected-resource metadata beside it, which is why the sub-app is mounted at the root rather than under ``/mcp``. + +An enrolled instance also answers at ``/i/{id}/mcp`` on its hub, so an agent +that came through the tunnel is shown the hub's addresses instead. """ from __future__ import annotations @@ -14,12 +17,16 @@ from urllib.parse import urlparse import jwt from fastapi.concurrency import run_in_threadpool from mcp.server.auth.provider import AccessToken +from mcp.server.auth.routes import build_resource_metadata_url from mcp.server.auth.settings import AuthSettings from mcp.server.transport_security import TransportSecuritySettings from pydantic import AnyHttpUrl -from starlette.applications import Starlette +from starlette.datastructures import Headers +from starlette.responses import JSONResponse +from starlette.types import ASGIApp, Message, Receive, Scope, Send from fluksio.api.deps import oauth_client_lives +from fluksio.cloud import config as cloud_config from fluksio.core import security from fluksio.core.config import settings from fluksio.mcp import server @@ -67,6 +74,20 @@ class _JWTVerifier: ) +class _AndTheHub(list[str]): + """The hosts named here, plus the hub's while this instance is enrolled. + + A call through the tunnel is addressed to the hub, as its Host says. The + hub is looked up per request, since enrolment comes and goes while the + endpoint is up. Its origin is not needed: the hub never forwards one. + """ + + def __contains__(self, host: object) -> bool: + config = cloud_config.load() + hub = urlparse(config.issuer).netloc if config is not None else None + return super().__contains__(host) or (hub is not None and host == hub) + + def _transport_security() -> TransportSecuritySettings: """Which Host headers to trust. @@ -87,14 +108,61 @@ def _transport_security() -> TransportSecuritySettings: "testserver", ] allowed_origins += ["http://localhost", "http://127.0.0.1"] - return TransportSecuritySettings( + # Unvalidated, since validating would copy the hosts into a plain list. + return TransportSecuritySettings.model_construct( enable_dns_rebinding_protection=True, - allowed_hosts=allowed_hosts, + allowed_hosts=_AndTheHub(allowed_hosts), allowed_origins=allowed_origins, ) -def build_http_app(app: object) -> Starlette: +class _ThroughThePortal: + """Show an agent that came through the hub the hub's addresses. + + The SDK builds its metadata once, for the API host, and names it in every + 401. An agent at ``/i/{id}/mcp`` on the hub cannot reach that host, so its + discovery has to lead to the hub's documents instead. + """ + + def __init__(self, app: ASGIApp) -> None: + self.app = app + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + base = None + if scope["type"] == "http": + base = cloud_config.portal_base(Headers(scope=scope)) + if base is None: + await self.app(scope, receive, send) + return + if scope["path"] == "/.well-known/oauth-protected-resource/mcp": + document = { + "resource": f"{base}/mcp", + "authorization_servers": [base], + "bearer_methods_supported": ["header"], + } + headers = {"Cache-Control": "public, max-age=3600"} + await JSONResponse(document, headers=headers)(scope, receive, send) + return + here = str(build_resource_metadata_url(AnyHttpUrl(settings.mcp_resource))) + there = str(build_resource_metadata_url(AnyHttpUrl(f"{base}/mcp"))) + + async def send_theirs(message: Message) -> None: + if message["type"] == "http.response.start": + message["headers"] = [ + ( + key, + value.replace(here.encode(), there.encode()) + if key == b"www-authenticate" + else value, + ) + for key, value in message["headers"] + ] + await send(message) + + await self.app(scope, receive, send_theirs) + + +def build_http_app(app: object) -> ASGIApp: """The MCP sub-app, wired to this app's API and its OAuth server.""" server.mcp.settings.stateless_http = True server.mcp.settings.json_response = True @@ -109,4 +177,4 @@ def build_http_app(app: object) -> Starlette: server.mcp._token_verifier = _JWTVerifier() # noqa: SLF001 server.connect(app) - return server.mcp.streamable_http_app() + return _ThroughThePortal(server.mcp.streamable_http_app()) diff --git a/backend/tests/api/routes/test_oauth.py b/backend/tests/api/routes/test_oauth.py index 100e05c..de21684 100644 --- a/backend/tests/api/routes/test_oauth.py +++ b/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: diff --git a/backend/tests/mcp/test_mcp_http.py b/backend/tests/mcp/test_mcp_http.py index 6d04e49..7c9d92e 100644 --- a/backend/tests/mcp/test_mcp_http.py +++ b/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: diff --git a/backend/tests/test_tunnel_reach.py b/backend/tests/test_tunnel_reach.py new file mode 100644 index 0000000..7214127 --- /dev/null +++ b/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 diff --git a/docs/code/agents.md b/docs/code/agents.md index 4ce4ad4..69080ae 100644 --- a/docs/code/agents.md +++ b/docs/code/agents.md @@ -22,8 +22,12 @@ Manage registered clients under **Admin → Agents**. Revoking a client invalidates its access. MCP uses a separate signing key at `OAUTH_PRIVATE_KEY_FILE`; rotating that key invalidates all MCP tokens. -MCP is not forwarded by the [portal tunnel](../interface/portal.md). -Connect directly to the instance's API host. +An instance connected to a [portal](../interface/portal.md) also answers +through it, at `https://hub.fluksio.com/i//mcp` (the portal's +own `hub.` host for a custom one). The portal forwards the endpoint and its +OAuth, and the instance checks the agent's token as it does locally. The +consent page opens in the instance as the portal serves it, so open the +instance from the portal first if this browser has not recently. ## Available tools diff --git a/docs/interface/portal.md b/docs/interface/portal.md index 67e4c58..ed6c540 100644 --- a/docs/interface/portal.md +++ b/docs/interface/portal.md @@ -56,6 +56,22 @@ Somebody admitted by a link gets their local account the first time they open the instance. Until then they are listed under **Admin → Users → Pending portal members**, where you can revoke the invitation. +## Hooks and agents through the portal + +The portal also forwards a flow's hooks and the MCP endpoint, so they work +without an inbound port: + +- A webhook or a served WebSocket: + `https://hub.fluksio.com/i//hooks///`, + with `wss://` for a socket. +- An agent: `https://hub.fluksio.com/i//mcp`, see + [Agents over MCP](../code/agents.md). + +Through the portal a hook needs a [secret](../reference/node-types.md#http). +A hook without one answers only on the instance's own address; through the +portal it gets the same 404, or closed socket, as a wrong secret. The instance +id is the one in the address when you open the instance from the portal. + ## Manage or end remote access **Settings → Remote access** shows the connection status and the latest diff --git a/docs/reference/node-types.md b/docs/reference/node-types.md index ea57a43..1551c55 100644 --- a/docs/reference/node-types.md +++ b/docs/reference/node-types.md @@ -80,6 +80,8 @@ publishes nothing. The canvas flags this as an advisory issue rather than stopping the flow, because a webhook on a private network is a legitimate thing to want. + Through the [portal](../interface/portal.md#hooks-and-agents-through-the-portal) + a webhook or served socket answers only with a secret. ### WebSocket diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts index 293dd18..4c659a3 100644 --- a/frontend/src/client/sdk.gen.ts +++ b/frontend/src/client/sdk.gen.ts @@ -1737,6 +1737,9 @@ export class OauthService { /** * Authorize * Approve a client, on behalf of the signed-in user. + * + * Through the hub the endpoint is the one there, which is the resource an + * agent that came that way asks for. * @param data The data for the request. * @param data.requestBody * @returns OAuthAuthorizeResponse Successful Response