fix(mcp): stop deriving the OAuth origin from X-Forwarded headers

deploy/helm/sie-cluster/templates/mcp-edge-deployment.yaml

@@ -44,9 +44,13 @@ spec:
                   name: {{ include "sie-cluster.mcpEdge.secretName" . }}
                   key: connector-secrets
             {{- end }}
-            {{- if .Values.mcpEdge.publicUrl }}
+            {{- $publicUrl := trim (default "" .Values.mcpEdge.publicUrl) }}
+            {{- if and (not $publicUrl) .Values.mcpEdge.ingress.enabled }}
+            {{- $publicUrl = printf "https://%s" (trim (default "" .Values.mcpEdge.ingress.host)) }}
+            {{- end }}
+            {{- if $publicUrl }}
             - name: SIE_MCP_PUBLIC_URL
-              value: {{ .Values.mcpEdge.publicUrl | quote }}
+              value: {{ $publicUrl | quote }}
             {{- end }}
             {{- if not .Values.mcpEdge.oauthEnabled }}
             - name: SIE_MCP_OAUTH_ENABLED

deploy/helm/sie-cluster/values.yaml

@@ -2075,7 +2075,11 @@ mcpEdge:
   connectorSecrets: ""
   existingSecretName: ""
   # SIE_MCP_PUBLIC_URL — the externally reachable https origin (e.g.
-  # https://mcp.example.com). Pin it so OAuth metadata URLs are stable.
+  # https://mcp.example.com) advertised as the OAuth issuer and endpoints.
+  # Empty => https://<mcpEdge.ingress.host> when the edge ingress is enabled
+  # (https even with chart TLS off, which implies upstream TLS termination).
+  # Set it explicitly for any other exposure (LoadBalancer, external proxy):
+  # unpinned, the edge refuses OAuth metadata for non-loopback hosts.
   publicUrl: ""
   # OAuth bridge for claude.ai / desktop connectors (#1312). Disable for
   # connector-secret-only (e.g. Cowork header) deployments.

packages/sie_mcp/README.md

@@ -397,4 +397,7 @@ For hosted deployments, export the public origin before starting the MCP edge:
 export SIE_MCP_PUBLIC_URL='https://mcp.example.com'
 ```
 
-This ensures that OAuth metadata contains stable public URLs.
+The OAuth metadata advertises this origin as the authorization server, so pin it
+on every exposed deployment. Unpinned, the edge serves OAuth metadata only when the
+request `Host` is loopback or listed in `SIE_MCP_ALLOWED_HOSTS`, and ignores
+`X-Forwarded-Host`.

packages/sie_mcp/plugin/superlinked.md

@@ -90,8 +90,11 @@ as on Cowork).
 
 ### Operator notes
 
-- Pin `SIE_MCP_PUBLIC_URL` to the externally reachable origin so the OAuth metadata URLs
-  are stable (otherwise they are derived per-request from forwarded host/proto headers).
+- Pin `SIE_MCP_PUBLIC_URL` to the externally reachable origin. It is advertised as the
+  OAuth issuer and authorize/token endpoints, which is where users type their connector
+  secret. Unpinned, metadata is served only when the request `Host` is loopback or listed in
+  `SIE_MCP_ALLOWED_HOSTS`, and `X-Forwarded-Host` is never trusted. The Helm chart derives it
+  from `mcpEdge.ingress.host`.
 - Run the edge as a **single worker** (the default `mise run mcp-serve`): the OAuth
   authorization-code store is in-process, so a code issued on one worker cannot be
   redeemed on another.

packages/sie_mcp/src/sie_mcp/app.py

@@ -1,5 +1,7 @@
 """ASGI app for the SIE MCP edge: MCP streamable-HTTP transport + auth + health."""
 
+import logging
+
 from starlette.applications import Starlette
 from starlette.requests import Request
 from starlette.responses import JSONResponse
@@ -10,6 +12,8 @@
 from sie_mcp.oauth import build_oauth_routes
 from sie_mcp.server import build_server
 
+logger = logging.getLogger(__name__)
+
 
 async def _healthz(_request: Request) -> JSONResponse:
     return JSONResponse({"status": "ok"})
@@ -21,6 +25,11 @@ def build_app() -> Starlette:
     app = build_server(config).streamable_http_app()
     app.router.routes.append(Route("/healthz", _healthz, methods=["GET"]))
     if config.oauth_enabled:
+        if not config.public_base_url:
+            logger.warning(
+                "SIE_MCP_PUBLIC_URL is unset: OAuth metadata is served only for loopback or SIE_MCP_ALLOWED_HOSTS "
+                "Host headers. Pin SIE_MCP_PUBLIC_URL to the public https origin for any exposed deployment."
+            )
         # The OAuth bridge lets claude.ai connectors authenticate via the connector
         # secret; the gate below exempts these bootstrap endpoints.
         app.router.routes.extend(build_oauth_routes(config))

packages/sie_mcp/src/sie_mcp/auth.py

@@ -33,17 +33,41 @@
 )
 
 
-def base_url(config: MCPConfig, *, scheme: str, headers: Headers) -> str:
+_LOOPBACK_HOSTNAMES = frozenset({"localhost", "127.0.0.1", "[::1]"})
+
+
+def _hostname(host: str) -> str:
+    if host.startswith("["):
+        return host[: host.find("]") + 1]
+    return host.rsplit(":", 1)[0]
+
+
+def _host_trusted(config: MCPConfig, host: str) -> bool:
+    host = host.lower()
+    if _hostname(host) in _LOOPBACK_HOSTNAMES:
+        return True
+    for allowed in (entry.lower() for entry in config.allowed_hosts):
+        if host == allowed or (allowed.endswith(":*") and host.startswith(allowed[:-1])):
+            return True
+    return False
+
+
+def base_url(config: MCPConfig, *, scheme: str, headers: Headers) -> str | None:
     """Resolve the externally reachable origin for OAuth metadata URLs.
 
-    Prefers the pinned ``SIE_MCP_PUBLIC_URL``; otherwise derives it from forwarded
-    proxy headers (falling back to the request's own scheme/host).
+    Prefers the pinned ``SIE_MCP_PUBLIC_URL``. Unpinned, the request's own scheme and
+    ``Host`` are used only when the host is loopback or listed in
+    ``SIE_MCP_ALLOWED_HOSTS``; otherwise ``None``. This origin names the authorization
+    server clients trust, so caller-controlled ``Host`` and ``X-Forwarded-*`` values are
+    never advertised. A proxy-set scheme is honoured only through uvicorn's
+    ``FORWARDED_ALLOW_IPS`` trust list, which rewrites ``scheme`` upstream.
     """
     if config.public_base_url:
         return config.public_base_url
-    proto = headers.get("x-forwarded-proto") or scheme
-    host = headers.get("x-forwarded-host") or headers.get("host") or ""
-    return f"{proto}://{host}"
+    host = headers.get("host") or ""
+    if not host or not _host_trusted(config, host):
+        return None
+    return f"{scheme}://{host}"
 
 
 def bearer_token(authorization: str | None) -> str | None:
@@ -107,5 +131,7 @@ def _challenge_headers(self, scope: Scope, headers: Headers) -> dict[str, str]:
         if not self._config.oauth_enabled:
             return {}
         origin = base_url(self._config, scheme=scope.get("scheme", "http"), headers=headers)
+        if origin is None:
+            return {}
         metadata = f"{origin}/.well-known/oauth-protected-resource"
         return {"WWW-Authenticate": f'Bearer resource_metadata="{metadata}"'}

packages/sie_mcp/src/sie_mcp/config.py

@@ -201,7 +201,7 @@ class MCPConfig:
     # claude.ai connectors are OAuth-only (no pasteable Bearer); the edge bridges
     # the OAuth handshake onto the connector-secret shim. `public_base_url`
     # pins the externally reachable origin used to build OAuth metadata URLs; when
-    # unset it is derived per-request from forwarded host/proto headers.
+    # unset it is derived per-request from a loopback or SIE_MCP_ALLOWED_HOSTS Host header.
     oauth_enabled: bool
     public_base_url: str | None
     oauth_redirect_uris: tuple[str, ...]

packages/sie_mcp/src/sie_mcp/oauth.py

@@ -229,13 +229,23 @@ def _parse_form(body: bytes) -> dict[str, str]:
     return {key: values[0] for key, values in parse_qs(decoded).items() if values}
 
 
+_ORIGIN_UNAVAILABLE = {
+    "error": "server_error",
+    "error_description": "OAuth origin is not configured; set SIE_MCP_PUBLIC_URL or SIE_MCP_ALLOWED_HOSTS.",
+}
+
+
 async def _protected_resource(request: Request, config: MCPConfig) -> JSONResponse:
     origin = base_url(config, scheme=request.url.scheme, headers=request.headers)
+    if origin is None:
+        return JSONResponse(_ORIGIN_UNAVAILABLE, status_code=503)
     return JSONResponse(protected_resource_metadata(origin))
 
 
 async def _authorization_server(request: Request, config: MCPConfig) -> JSONResponse:
     origin = base_url(config, scheme=request.url.scheme, headers=request.headers)
+    if origin is None:
+        return JSONResponse(_ORIGIN_UNAVAILABLE, status_code=503)
     return JSONResponse(authorization_server_metadata(origin))
 
 

packages/sie_mcp/tests/test_auth.py

@@ -119,21 +119,67 @@ def test_base_url_prefers_pinned_public_url() -> None:
     assert base_url(cfg, scheme="http", headers=headers) == "https://mcp.example.com"
 
 
-def test_base_url_derives_from_forwarded_headers() -> None:
-    cfg = _cfg(public_base_url=None)
-    headers = Headers({"host": "internal:8088", "x-forwarded-proto": "https", "x-forwarded-host": "mcp.example.com"})
-    assert base_url(cfg, scheme="http", headers=headers) == "https://mcp.example.com"
+def test_base_url_derives_from_request_scheme_and_allowed_host() -> None:
+    cfg = _cfg(public_base_url=None, allowed_hosts=["mcp.example.com"])
+    headers = Headers({"host": "mcp.example.com"})
+    assert base_url(cfg, scheme="https", headers=headers) == "https://mcp.example.com"
+
+
+def test_base_url_ignores_forwarded_headers() -> None:
+    cfg = _cfg(public_base_url=None, allowed_hosts=["mcp.example.com"])
+    headers = Headers(
+        {"host": "mcp.example.com", "x-forwarded-proto": "http", "x-forwarded-host": "evil.attacker.example"}
+    )
+    assert base_url(cfg, scheme="https", headers=headers) == "https://mcp.example.com"
+
+
+@pytest.mark.parametrize("host", ["evil.attacker.example", "mcp.example.com.evil.example", "", "notlocalhost:8088"])
+def test_base_url_refuses_untrusted_host(host: str) -> None:
+    cfg = _cfg(public_base_url=None, allowed_hosts=["mcp.example.com", "edge.example.com:*"])
+    assert base_url(cfg, scheme="https", headers=Headers({"host": host})) is None
+
+
+@pytest.mark.parametrize(
+    ("host", "expected"),
+    [
+        ("localhost:8088", "http://localhost:8088"),
+        ("127.0.0.1:8088", "http://127.0.0.1:8088"),
+        ("[::1]:8088", "http://[::1]:8088"),
+        ("LOCALHOST:8088", "http://LOCALHOST:8088"),
+        ("edge.example.com:8443", "http://edge.example.com:8443"),
+        ("Edge.Example.com:8443", "http://Edge.Example.com:8443"),
+        ("mcp.example.com", "http://mcp.example.com"),
+    ],
+)
+def test_base_url_trusts_loopback_and_wildcard_port_hosts(host: str, expected: str) -> None:
+    cfg = _cfg(public_base_url=None, allowed_hosts=["EDGE.example.com:*", "MCP.Example.com"])
+    assert base_url(cfg, scheme="http", headers=Headers({"host": host})) == expected
 
 
 async def test_unauthorized_returns_www_authenticate_challenge() -> None:
-    cfg = _cfg(connector_secrets={"s3cret": "user-1"})
+    cfg = _cfg(connector_secrets={"s3cret": "user-1"}, allowed_hosts=["mcp.example.com"])
     reached, start = await _run_middleware(cfg, path="/mcp", headers=[(b"host", b"mcp.example.com")])
     assert reached["app"] is False
     assert start["status"] == 401
     challenge = Headers(raw=start["headers"])["www-authenticate"]
     assert 'resource_metadata="https://mcp.example.com/.well-known/oauth-protected-resource"' in challenge
 
 
+async def test_unauthorized_challenge_ignores_forwarded_host() -> None:
+    cfg = _cfg(connector_secrets={"s3cret": "user-1"}, allowed_hosts=["mcp.example.com"])
+    headers = [(b"host", b"mcp.example.com"), (b"x-forwarded-host", b"evil.attacker.example")]
+    _reached, start = await _run_middleware(cfg, path="/mcp", headers=headers)
+    challenge = Headers(raw=start["headers"])["www-authenticate"]
+    assert 'resource_metadata="https://mcp.example.com/.well-known/oauth-protected-resource"' in challenge
+
+
+async def test_unauthorized_omits_challenge_for_untrusted_host() -> None:
+    cfg = _cfg(connector_secrets={"s3cret": "user-1"})
+    _reached, start = await _run_middleware(cfg, path="/mcp", headers=[(b"host", b"evil.attacker.example")])
+    assert start["status"] == 401
+    assert "www-authenticate" not in Headers(raw=start["headers"])
+
+
 async def test_unauthorized_omits_challenge_when_oauth_disabled() -> None:
     cfg = _cfg(connector_secrets={"s3cret": "user-1"}, oauth_enabled=False)
     _reached, start = await _run_middleware(cfg, path="/mcp", headers=[(b"host", b"mcp.example.com")])

packages/sie_mcp/tests/test_oauth.py

@@ -219,9 +219,42 @@ def test_metadata_documents_use_origin() -> None:
     assert server["code_challenge_methods_supported"] == ["S256"]
 
 
-def _client(cfg: MCPConfig) -> TestClient:
+def _client(cfg: MCPConfig, *, base_url: str = "https://mcp.example.com") -> TestClient:
     app = Starlette(routes=build_oauth_routes(cfg))
-    return TestClient(app, base_url="https://mcp.example.com")
+    return TestClient(app, base_url=base_url)
+
+
+@pytest.mark.parametrize("path", ["/.well-known/oauth-authorization-server", "/.well-known/oauth-protected-resource"])
+def test_metadata_origin_ignores_forwarded_headers(path: str) -> None:
+    body = (
+        _client(_cfg(allowed_hosts=["mcp.example.com"]))
+        .get(path, headers={"X-Forwarded-Host": "evil.attacker.example", "X-Forwarded-Proto": "http"})
+        .json()
+    )
+    assert "evil.attacker.example" not in str(body)
+    assert "http://" not in str(body)
+    if "issuer" in body:
+        assert body["issuer"] == "https://mcp.example.com"
+        assert body["token_endpoint"] == "https://mcp.example.com/token"  # noqa: S105
+    else:
+        assert body["authorization_servers"] == ["https://mcp.example.com"]
+
+
+@pytest.mark.parametrize("path", ["/.well-known/oauth-authorization-server", "/.well-known/oauth-protected-resource"])
+def test_metadata_refused_for_untrusted_host_when_unpinned(path: str) -> None:
+    resp = _client(_cfg()).get(path)
+    assert resp.status_code == 503
+    assert "mcp.example.com" not in resp.text
+
+
+def test_metadata_served_for_loopback_host_when_unpinned() -> None:
+    body = _client(_cfg(), base_url="http://localhost:8088").get("/.well-known/oauth-authorization-server").json()
+    assert body["issuer"] == "http://localhost:8088"
+
+
+def test_metadata_uses_pinned_public_url_for_any_host() -> None:
+    client = _client(_cfg(public_base_url="https://mcp.example.com"), base_url="https://evil.attacker.example")
+    assert client.get("/.well-known/oauth-authorization-server").json()["issuer"] == "https://mcp.example.com"
 
 
 def _authorize_params(verifier: str) -> dict[str, str]:
@@ -244,7 +277,7 @@ def _obtain_code(client: TestClient, verifier: str) -> str:
 
 
 def test_metadata_endpoints_use_request_origin() -> None:
-    client = _client(_cfg())
+    client = _client(_cfg(allowed_hosts=["mcp.example.com"]))
     resource = client.get("/.well-known/oauth-protected-resource").json()
     assert resource["resource"] == "https://mcp.example.com/mcp"
     server = client.get("/.well-known/oauth-authorization-server").json()
@@ -376,7 +409,7 @@ def test_token_rejects_client_id_mismatch_over_http() -> None:
 
 def test_protected_resource_metadata_path_suffixed() -> None:
     # RFC 9728 path-suffixed variant resolves the same metadata.
-    client = _client(_cfg())
+    client = _client(_cfg(allowed_hosts=["mcp.example.com"]))
     resp = client.get("/.well-known/oauth-protected-resource/mcp")
     assert resp.status_code == 200
     assert resp.json()["resource"] == "https://mcp.example.com/mcp"