fix(mcp): pin OAuth return_url + sanitise read-scope status

Addresses ten findings on the Phase 4 OAuth-MCP commit: four from the
PR #478 review surface, plus six surfaced by a follow-up multi-stage
review of the first round of fixes. Two of the latter were genuine
security regressions in the very code that claimed to close those
holes.

Security
--------

- _validate_return_url now pins return_url same-origin against the
  configured oidc_config.redirect_base instead of request.url. Behind
  a permissive front proxy that did not normalise Host, an attacker
  could spoof Host and provide a matching absolute return_url to mint
  an open redirect off /api/mcp/oauth/start. Same fix pattern as
  PR #476 OIDC.
- Reject return_url values containing literal backslashes or starting
  with `//` up front. urlparse leaves backslashes inside `path`, so a
  value like `/\evil.example/foo` slipped through the path-only branch
  and became the protocol-relative `//evil.example/foo` after WHATWG-
  conformant browsers normalised the backslash — re-introducing the
  open redirect the same-origin pin was meant to close.
- internal_mcp_status (read-scoped) projects through a new
  _strip_server_status_for_read helper that drops the verbose `error`
  text and replaces it with a coarse `has_error` boolean. The error
  string is built as `f"{type(exc).__name__}: {exc}"` and so carries
  stdio binary paths (FileNotFoundError) or internal MCP URLs
  (httpx.ConnectError) — equivalent to leaking command/url, which
  this same patch deliberately strips. Approve-scoped refresh and
  reconnect callers continue to receive the full `error` text via
  the existing _strip_server_status helper.
- internal_mcp_status now returns the projected (sanitised) entries
  for every server in mcp_mgr.get_all_server_status() instead of
  emitting the un-sanitised dict that included `command` (stdio argv)
  and `url` (remote MCP endpoint). Sibling refresh/reconnect endpoints
  already used _public_server_status to strip these.
- internal_mcp_status docstring documents the trust boundary — server
  enumeration to read scope is intentional so dashboards can render
  per-server indicators; verbose error detail and command/url remain
  approve-scoped.

Correctness / UX
----------------

- _validate_return_url comparison normalises (scheme, host, port)
  before equality. Lowercases hostname and collapses the scheme's
  default port, so `https://App.Example.COM/x` and
  `https://app.example.com:443/x` are recognised as same-origin
  with `redirect_base = https://app.example.com` instead of being
  silently downgraded to the `/` fallback.
- mcp_crypto startup-gate error message now names both
  `mcp_token_encryption_keys` (rotation list) and
  `mcp_token_encryption_key` (single) so an operator using rotation
  isn't misled into thinking only the singular form is valid.

Cleanup
-------

- Delete the unused _KNOWN_TRUSTED_ENDPOINT_HOSTS legacy re-export
  shim in oidc.py (zero callers — a no-op that survived the Phase 4
  oauth_ssrf extraction). Sphinx :data: docstring reference at
  validate_discovered_endpoint updated to point at
  turnstone.core.oauth_ssrf.KNOWN_TRUSTED_OAUTH_ENDPOINT_HOSTS
  directly. The Google multi-origin allowlist is unaffected — it
  lives at the canonical name and is read from oauth_ssrf.py:164.
- test_mcp_oauth_handlers TestValidateReturnUrl imports
  _validate_return_url at module level instead of repeating the
  import inside each test method.
- test_server_lifespan_mcp_crypto replaces a fragile
  `messages.count("mcp_token_encryption_key") >= 2` substring trick
  with `re.search(r"mcp_token_encryption_key(?!s)", messages)` —
  asserts the singular form directly via negative lookahead.

Tests
-----

5448 pass (+13 vs the prior tip):

- TestValidateReturnUrl gains backslash-bypass, protocol-relative,
  default-port, uppercase-host, and explicit-port-mismatch cases
  alongside the original same-origin / cross-origin / scheme-
  mismatch / path-only cases.
- TestInternalMcpStatusEndpoint asserts the `error` text never
  reaches the read-scope wire (binary-path FileNotFoundError no
  longer appears anywhere in the rendered response) and that the
  coarse `has_error` boolean lights up correctly on the failed
  server.
- TestInternalMcpStatusEndpoint also pins the no-mcp-client path to
  `{"servers": {}}`.
- _routes_with_internal extended to include the
  /api/_internal/mcp-status route so the new tests can exercise it
  through TestClient.
- Existing test_startup_aborts_with_oauth_user_row_and_no_key
  strengthened to require both singular and plural key names appear
  in the error log.

(cherry picked from commit 62bbc332af)

tests/test_mcp_admin_api.py

@@ -142,6 +142,7 @@ def _routes_with_internal() -> list[Mount]:
         internal_mcp_reconnect_one,
         internal_mcp_refresh_one,
         internal_mcp_reload,
+        internal_mcp_status,
     )
 
     return [
@@ -150,6 +151,7 @@ def _routes_with_internal() -> list[Mount]:
             routes=[
                 *_ROUTES[0].routes,  # type: ignore[union-attr]
                 Route("/api/_internal/mcp-reload", internal_mcp_reload, methods=["POST"]),
+                Route("/api/_internal/mcp-status", internal_mcp_status),
                 Route(
                     "/api/_internal/mcp-refresh/{name}",
                     internal_mcp_refresh_one,
@@ -1712,3 +1714,93 @@ class TestInternalMcpReconnectOneEndpoint:
         assert r.status_code == 400
         assert "invalid" in r.json()["error"].lower()
         mgr.reconnect_sync.assert_not_called()
+
+
+# ---------------------------------------------------------------------------
+# Node fan-out status endpoint: GET /v1/api/_internal/mcp-status
+# ---------------------------------------------------------------------------
+
+
+class TestInternalMcpStatusEndpoint:
+    """HTTP-level tests for the node-side aggregate status endpoint.
+
+    The endpoint falls through to ``read`` scope (a deliberate choice
+    so dashboards can render status indicators for non-admin
+    operators). Because of that, the response must strip ``command``
+    (stdio argv) and ``url`` (remote MCP endpoint) — both admin-only
+    context — before it leaves the process.
+    """
+
+    @pytest.fixture()
+    def node_app_factory(self, storage: SQLiteBackend):
+        def _make(mgr: Any) -> TestClient:
+            app = Starlette(
+                routes=_routes_with_internal(),
+                middleware=[Middleware(_InjectAuthMiddleware)],
+            )
+            app.state.auth_storage = storage
+            if mgr is not None:
+                app.state.mcp_client = mgr
+            return TestClient(app, raise_server_exceptions=False)
+
+        return _make
+
+    def test_status_strips_command_url_and_error(self, node_app_factory) -> None:
+        mgr = MagicMock()
+        mgr.get_all_server_status.return_value = {
+            "srv-stdio": {
+                "connected": False,
+                "tools": 0,
+                "resources": 0,
+                "prompts": 0,
+                # Error text would carry the binary path even after
+                # ``command`` is stripped — read scope must not see it.
+                "error": "FileNotFoundError: [Errno 2] No such file or "
+                "directory: '/usr/local/bin/secret-mcp-bin'",
+                "transport": "stdio",
+                "command": ["/usr/local/bin/secret-mcp-bin", "--token", "abc"],
+                "url": "",
+                "circuit_open": True,
+                "consecutive_failures": 5,
+            },
+            "srv-http": {
+                "connected": True,
+                "tools": 1,
+                "resources": 1,
+                "prompts": 0,
+                "error": "",
+                "transport": "streamable-http",
+                "command": "",
+                "url": "https://internal-mcp.example/mcp",
+                "circuit_open": False,
+                "consecutive_failures": 0,
+            },
+        }
+        c = node_app_factory(mgr)
+        r = c.get("/v1/api/_internal/mcp-status")
+        assert r.status_code == 200
+        servers = r.json()["servers"]
+        assert set(servers) == {"srv-stdio", "srv-http"}
+        for entry in servers.values():
+            assert "command" not in entry
+            assert "url" not in entry
+            # ``error`` text is replaced by ``has_error`` boolean so a
+            # FileNotFoundError binary path or httpx URL cannot leak
+            # through verbose exception messages at read scope.
+            assert "error" not in entry
+            assert "has_error" in entry
+        # Coarse error indicator preserved.
+        assert servers["srv-stdio"]["has_error"] is True
+        assert servers["srv-http"]["has_error"] is False
+        # Operational fields preserved.
+        assert servers["srv-http"]["tools"] == 1
+        assert servers["srv-http"]["transport"] == "streamable-http"
+        assert servers["srv-stdio"]["circuit_open"] is True
+        # No leaked binary path anywhere in the rendered response.
+        assert "secret-mcp-bin" not in r.text
+
+    def test_status_no_mcp_client_returns_empty_servers(self, node_app_factory) -> None:
+        c = node_app_factory(None)
+        r = c.get("/v1/api/_internal/mcp-status")
+        assert r.status_code == 200
+        assert r.json() == {"servers": {}}

tests/test_mcp_oauth_handlers.py

@@ -27,6 +27,7 @@ from tests.conftest import make_mcp_token_cipher
 from turnstone.core.auth import AuthResult
 from turnstone.core.mcp_crypto import MCPTokenStore
 from turnstone.core.mcp_oauth import (
+    _validate_return_url,
     handle_mcp_oauth_authorize,
     handle_mcp_oauth_callback,
 )
@@ -868,3 +869,73 @@ class TestCallbackErrorPopsPendingState:
         )
         assert replay.status_code == 302
         assert "session+expired" in replay.headers["location"]
+
+
+# ---------------------------------------------------------------------------
+# return_url same-origin pinning (Host-header injection defense)
+# ---------------------------------------------------------------------------
+
+
+class TestValidateReturnUrl:
+    """Direct unit tests for ``_validate_return_url``.
+
+    The validator pins ``return_url`` against the configured
+    ``redirect_base`` rather than the request Host header, so a
+    permissive front proxy cannot turn the OAuth callback into an
+    open redirect via spoofed ``Host``.
+    """
+
+    REDIRECT_BASE = "https://app.example.com"
+
+    def test_empty_returns_none(self) -> None:
+        assert _validate_return_url("", self.REDIRECT_BASE) is None
+
+    def test_relative_path_passes(self) -> None:
+        assert _validate_return_url("/admin/mcp", self.REDIRECT_BASE) == "/admin/mcp"
+
+    def test_non_root_relative_rejected(self) -> None:
+        assert _validate_return_url("admin/mcp", self.REDIRECT_BASE) is None
+
+    def test_same_origin_absolute_passes(self) -> None:
+        same_origin = "https://app.example.com/admin/mcp"
+        assert _validate_return_url(same_origin, self.REDIRECT_BASE) == same_origin
+
+    def test_cross_origin_absolute_rejected(self) -> None:
+        # Even when the request arrives with ``Host: attacker.example``
+        # and ``return_url`` matches that host, pinning to the
+        # configured redirect_base catches the open-redirect attempt.
+        assert _validate_return_url("https://attacker.example/cb", self.REDIRECT_BASE) is None
+
+    def test_scheme_mismatch_rejected(self) -> None:
+        assert _validate_return_url("http://app.example.com/admin", self.REDIRECT_BASE) is None
+
+    def test_backslash_in_path_rejected(self) -> None:
+        # WHATWG-conformant browsers normalise ``\`` to ``/``, so
+        # ``/\evil.example/foo`` becomes the protocol-relative
+        # ``//evil.example/foo`` after the 302 — must be rejected up
+        # front because urlparse leaves the backslash inside ``path``
+        # and the path-only branch would otherwise return it verbatim.
+        assert _validate_return_url("/\\evil.example/foo", self.REDIRECT_BASE) is None
+        # Trailing-position backslash still rejected (defense-in-depth).
+        assert _validate_return_url("/admin\\..", self.REDIRECT_BASE) is None
+
+    def test_protocol_relative_rejected(self) -> None:
+        # ``//evil.example/foo`` would urlparse to netloc=evil.example
+        # and fail the cross-origin check anyway, but the early
+        # ``startswith("//")`` reject is defense-in-depth.
+        assert _validate_return_url("//evil.example/foo", self.REDIRECT_BASE) is None
+
+    def test_same_origin_with_default_port_passes(self) -> None:
+        # Operator may legitimately type ``:443`` even though
+        # ``redirect_base`` was configured without it.
+        url = "https://app.example.com:443/admin"
+        assert _validate_return_url(url, self.REDIRECT_BASE) == url
+
+    def test_same_origin_with_uppercase_host_passes(self) -> None:
+        url = "https://App.Example.COM/admin"
+        assert _validate_return_url(url, self.REDIRECT_BASE) == url
+
+    def test_explicit_port_mismatch_rejected(self) -> None:
+        assert (
+            _validate_return_url("https://app.example.com:8443/admin", self.REDIRECT_BASE) is None
+        )

tests/test_server_lifespan_mcp_crypto.py

@@ -7,6 +7,7 @@ MCP server rows.
 
 from __future__ import annotations
 
+import re
 import types
 
 import pytest
@@ -90,8 +91,12 @@ class TestInitializeMcpCryptoState:
         ):
             initialize_mcp_crypto_state(state, node_id="n1")
         assert exc_info.value.code == 1
-        # Operator-actionable error message names the missing config key.
-        assert any("mcp_token_encryption_key" in record.message for record in caplog.records)
+        # Operator-actionable error message names BOTH supported config-key
+        # forms so an operator using the rotation list (plural) is not
+        # misled into thinking only the singular form is valid.
+        messages = " ".join(record.message for record in caplog.records)
+        assert "mcp_token_encryption_keys" in messages
+        assert re.search(r"mcp_token_encryption_key(?!s)", messages) is not None
 
     def test_startup_aborts_with_invalid_key(
         self, backend, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture

turnstone/core/mcp_crypto.py

@@ -473,7 +473,8 @@ def initialize_mcp_crypto_state(app_state: object, *, node_id: str = "") -> None
     if oauth_user_count > 0 and cipher_cfg is None:
         log.error(
             "mcp.oauth: %d server(s) configured with auth_type='oauth_user' but no "
-            "[security] mcp_token_encryption_key in config.toml. Generate a key with: "
+            "[security] mcp_token_encryption_keys (rotation list) or "
+            "mcp_token_encryption_key (single) in config.toml. Generate a key with: "
             "python -c 'from cryptography.fernet import Fernet; "
             "print(Fernet.generate_key().decode())' "
             "and add it to your config.toml.",

turnstone/core/mcp_oauth.py

@@ -1463,22 +1463,52 @@ async def _register_dynamic_client_if_needed(
         return client_id, None
 
 
-def _validate_return_url(return_url: str, request: Request) -> str | None:
-    """Ensure ``return_url`` is same-origin with the request. Returns sanitised URL or None."""
+def _validate_return_url(return_url: str, redirect_base: str) -> str | None:
+    """Ensure ``return_url`` is same-origin with the configured *redirect_base*.
+
+    Pinning to ``redirect_base`` (rather than ``request.url``) is the
+    same defense as :func:`_resolve_redirect_base`: a permissive front
+    proxy can let an attacker spoof ``Host`` and pass a same-origin
+    check derived from the request, turning the callback into an open
+    redirect. The OIDC module pinned this in PR #476.
+
+    Backslashes and protocol-relative ``//`` prefixes are rejected up
+    front: WHATWG-conformant browsers normalise ``\\`` to ``/``, so a
+    path-only value like ``/\\evil.example/foo`` becomes the
+    protocol-relative ``//evil.example/foo`` after the 302 — slipping
+    past ``urlparse`` (which leaves the backslash inside ``path``) and
+    re-introducing the open redirect.
+    """
     if not return_url:
         return None
+    if "\\" in return_url or return_url.startswith("//"):
+        return None
     parsed = urllib.parse.urlparse(return_url)
     # Allow path-only return URLs.
     if not parsed.scheme and not parsed.netloc:
         if parsed.path.startswith("/"):
             return return_url
         return None
-    request_origin = (request.url.scheme, request.url.netloc)
-    if (parsed.scheme, parsed.netloc) != request_origin:
+    base = urllib.parse.urlparse(redirect_base)
+    if _origin_tuple(parsed) != _origin_tuple(base):
         return None
     return return_url
 
 
+def _origin_tuple(parsed: urllib.parse.ParseResult) -> tuple[str, str, int | None]:
+    """Canonicalise (scheme, host, port) for same-origin comparison.
+
+    Lowercases scheme and hostname, and collapses the scheme's default
+    port — so ``https://Host`` matches ``https://host:443`` instead of
+    silently failing the same-origin check on a cosmetic difference.
+    """
+    scheme = parsed.scheme.lower()
+    host = (parsed.hostname or "").lower()
+    default_port = {"https": 443, "http": 80}.get(scheme)
+    port = parsed.port if parsed.port is not None else default_port
+    return (scheme, host, port)
+
+
 def _apply_security_headers(response: Response) -> Response:
     """Stamp framing protection on OAuth responses.
 
@@ -1531,7 +1561,9 @@ async def _handle_mcp_oauth_authorize_inner(request: Request) -> Response:
     if not server_name:
         return JSONResponse({"error": "Missing 'server' query parameter"}, status_code=400)
 
-    return_url = _validate_return_url(request.query_params.get("return_url", "").strip(), request)
+    return_url = _validate_return_url(
+        request.query_params.get("return_url", "").strip(), redirect_base
+    )
     if return_url is None:
         # Fall back to root — operators often hit /start without a hint.
         return_url = "/"

turnstone/core/oidc.py

@@ -26,7 +26,6 @@ if TYPE_CHECKING:
 
 from turnstone.core.log import get_logger
 from turnstone.core.oauth_ssrf import (
-    KNOWN_TRUSTED_OAUTH_ENDPOINT_HOSTS,
     OAuthSSRFError,
     is_localhost,
 )
@@ -67,11 +66,6 @@ _ALLOWED_ID_TOKEN_ALGS = [
     "PS512",
 ]
 
-# Re-export the shared trusted-host map under the legacy OIDC name so existing
-# callers / tests continue to work without churn.
-_KNOWN_TRUSTED_ENDPOINT_HOSTS = KNOWN_TRUSTED_OAUTH_ENDPOINT_HOSTS
-
-
 # ---------------------------------------------------------------------------
 # Exceptions
 # ---------------------------------------------------------------------------
@@ -328,8 +322,8 @@ def validate_discovered_endpoint(
     hostname. Strict equality is intentional — a hostile or compromised IdP
     must not be able to redirect ``token_endpoint`` to a third-party host where
     ``client_secret`` would leak. Multi-origin IdPs (e.g. Google) are
-    accommodated via :data:`_KNOWN_TRUSTED_ENDPOINT_HOSTS` plus an
-    operator-configurable ``trusted_endpoint_hosts`` list.
+    accommodated via :data:`turnstone.core.oauth_ssrf.KNOWN_TRUSTED_OAUTH_ENDPOINT_HOSTS`
+    plus an operator-configurable ``trusted_endpoint_hosts`` list.
 
     The scheme must match the issuer's scheme, and the *effective* port (with
     scheme defaults applied) must match — so ``https://host`` and

turnstone/server.py

@@ -2855,15 +2855,6 @@ def internal_mcp_reload(request: Request) -> JSONResponse:
     return JSONResponse({"status": "ok", **result})
 
 
-def internal_mcp_status(request: Request) -> JSONResponse:
-    """GET /v1/api/_internal/mcp-status — return MCP server status."""
-    mcp_mgr = getattr(request.app.state, "mcp_client", None)
-    if mcp_mgr is None:
-        return JSONResponse({"servers": {}})
-
-    return JSONResponse({"servers": mcp_mgr.get_all_server_status()})
-
-
 _SERVER_STATUS_PUBLIC_KEYS: tuple[str, ...] = (
     "connected",
     "tools",
@@ -2875,16 +2866,69 @@ _SERVER_STATUS_PUBLIC_KEYS: tuple[str, ...] = (
     "consecutive_failures",
 )
 
+_READ_STATUS_PUBLIC_KEYS: tuple[str, ...] = tuple(
+    k for k in _SERVER_STATUS_PUBLIC_KEYS if k != "error"
+)
+
+
+def _strip_server_status(full: dict[str, Any]) -> dict[str, Any]:
+    """Project a status dict to the approve-scope public-safe key set.
+
+    The full status dict embeds ``command`` (stdio argv) and ``url``
+    (remote MCP endpoint) which are admin-only context. Approve-scoped
+    callers (refresh/reconnect) get the verbose ``error`` text so an
+    operator triaging a failure sees the underlying exception.
+
+    Read-scope callers must use :func:`_strip_server_status_for_read`
+    instead — error strings can carry stdio binary paths
+    (``FileNotFoundError: ... '/usr/local/bin/...'``) or internal MCP
+    URLs (``httpx.ConnectError: ... 'https://internal/...'``) and
+    those are equivalent to leaking ``command``/``url``.
+    """
+    return {k: full[k] for k in _SERVER_STATUS_PUBLIC_KEYS if k in full}
+
+
+def _strip_server_status_for_read(full: dict[str, Any]) -> dict[str, Any]:
+    """Project a status dict for read-scope callers.
+
+    Drops the verbose ``error`` text and replaces it with a coarse
+    ``has_error: bool`` so dashboards can light up a failure indicator
+    without leaking the underlying exception detail.
+    """
+    out = {k: full[k] for k in _READ_STATUS_PUBLIC_KEYS if k in full}
+    out["has_error"] = bool(full.get("error"))
+    return out
+
 
 def _public_server_status(mcp_mgr: Any, name: str) -> dict[str, Any]:
-    """Strip ``command``/``url`` from ``get_server_status`` before returning over the wire.
+    """Strip ``command``/``url`` from ``get_server_status`` before returning over the wire."""
+    return _strip_server_status(mcp_mgr.get_server_status(name))
 
-    The full status dict embeds stdio argv and remote URLs that are
-    admin-only context.  Internal node endpoints surface only the
-    operational fields callers need to reflect to the operator.
+
+def internal_mcp_status(request: Request) -> JSONResponse:
+    """GET /v1/api/_internal/mcp-status — return MCP server status.
+
+    Read-scoped. The full set of configured MCP server names is
+    enumerated to any caller with ``read`` scope: that is the
+    intentional trust boundary so dashboards can render per-server
+    indicators without admin scope. Verbose ``error`` text is dropped
+    in favour of ``has_error`` (see :func:`_strip_server_status_for_read`)
+    because error strings can carry stdio binary paths or internal
+    MCP URLs. Per-server error detail and ``command``/``url`` remain
+    on the approve-scoped ``mcp-refresh``/``mcp-reconnect`` endpoints.
     """
-    full = mcp_mgr.get_server_status(name)
-    return {k: full[k] for k in _SERVER_STATUS_PUBLIC_KEYS if k in full}
+    mcp_mgr = getattr(request.app.state, "mcp_client", None)
+    if mcp_mgr is None:
+        return JSONResponse({"servers": {}})
+
+    return JSONResponse(
+        {
+            "servers": {
+                name: _strip_server_status_for_read(status)
+                for name, status in mcp_mgr.get_all_server_status().items()
+            }
+        }
+    )
 
 
 def internal_mcp_refresh_one(request: Request) -> JSONResponse: