fix(mcp): validate Host and Origin headers on the HTTP transport

airbyte/mcp/__init__.py

@@ -211,6 +211,12 @@
 
 - `MCP_SERVER_URL` — public base URL of the server (also used for OIDC redirect
   callbacks); defaults to `http://localhost:8080`.
+- `AIRBYTE_MCP_ALLOWED_HOSTS` — comma-separated allowed hostnames or `fnmatch`
+  patterns for HTTP `Host` and `Origin` validation. Ports are ignored: an entry
+  may carry one for readability, but matching is on hostname only, so
+  `example.com:8443` also allows `example.com` on any port.
+- `AIRBYTE_MCP_HTTP_HOST` — host interface to bind for the HTTP server (defaults
+  to `0.0.0.0`).
 - `AIRBYTE_MCP_OIDC_CLIENT_ID`, `AIRBYTE_MCP_OIDC_CLIENT_SECRET` — enable
   interactive OIDC (both required).
 - `AIRBYTE_MCP_OIDC_CONFIG_URL` — OIDC discovery URL (required when the client

airbyte/mcp/_transport_security.py

@@ -0,0 +1,129 @@
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+"""ASGI transport security for the HTTP MCP server."""
+
+from __future__ import annotations
+
+import fnmatch
+import ipaddress
+import os
+from typing import TYPE_CHECKING
+from urllib.parse import urlparse
+
+from starlette.responses import Response
+
+
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+
+    from starlette.types import ASGIApp, Receive, Scope, Send
+
+
+ALLOWED_HOSTS_ENV = "AIRBYTE_MCP_ALLOWED_HOSTS"
+HTTP_HOST_ENV = "AIRBYTE_MCP_HTTP_HOST"
+
+_DEFAULT_ALLOWED_HOSTS = ("127.0.0.1", "localhost", "::1")
+
+
+def _strip_host_port(value: str) -> str:
+    """Normalize a host or host pattern for comparison."""
+    normalized = value.strip().lower()
+    if normalized.startswith("["):
+        closing_bracket = normalized.find("]")
+        if closing_bracket != -1:
+            return normalized[1:closing_bracket]
+    if normalized.count(":") == 1:
+        return normalized.rsplit(":", 1)[0]
+    return normalized
+
+
+def _is_unspecified_address(hostname: str) -> bool:
+    try:
+        return ipaddress.ip_address(hostname).is_unspecified
+    except ValueError:
+        return False
+
+
+def resolve_allowed_hosts(server_url: str) -> tuple[str, ...]:
+    """Resolve allowed hostnames from defaults, the server URL, and the environment."""
+    resolved: list[str] = []
+    seen: set[str] = set()
+
+    def add(host: str) -> None:
+        cleaned = host.strip()
+        if not cleaned:
+            return
+        key = _strip_host_port(cleaned)
+        if key and key not in seen:
+            resolved.append(cleaned)
+            seen.add(key)
+
+    for host in _DEFAULT_ALLOWED_HOSTS:
+        add(host)
+
+    hostname = urlparse(server_url).hostname
+    if hostname and not _is_unspecified_address(hostname):
+        add(hostname)
+
+    for host in os.getenv(ALLOWED_HOSTS_ENV, "").split(","):
+        add(host)
+
+    return tuple(resolved)
+
+
+def _request_host(scope: Scope) -> str | None:
+    hosts = [
+        value.decode("latin-1")
+        for name, value in scope.get("headers", [])
+        if name.lower() == b"host"
+    ]
+    if len(hosts) != 1:
+        return None
+    return _strip_host_port(hosts[0]) or None
+
+
+def _origin_hosts(scope: Scope) -> tuple[str | None, ...]:
+    origins: list[str | None] = []
+    for name, value in scope.get("headers", []):
+        if name.lower() == b"origin":
+            origin = value.decode("latin-1").strip()
+            try:
+                hostname = urlparse(origin).hostname
+            except ValueError:
+                origins.append(None)
+            else:
+                origins.append(_strip_host_port(hostname) if hostname else None)
+    return tuple(origins)
+
+
+class HostOriginGuardMiddleware:
+    """Reject HTTP requests whose host or origin is outside the allowlist."""
+
+    def __init__(self, app: ASGIApp, allowed_hosts: Sequence[str]) -> None:
+        self.app = app
+        self.allowed_hosts = tuple(
+            normalized
+            for normalized in (_strip_host_port(host) for host in allowed_hosts)
+            if normalized
+        )
+
+    def _is_allowed(self, hostname: str | None) -> bool:
+        return hostname is not None and any(
+            fnmatch.fnmatchcase(hostname, allowed_host) for allowed_host in self.allowed_hosts
+        )
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+
+        if not self._is_allowed(_request_host(scope)):
+            await Response("Misdirected Request", status_code=421)(scope, receive, send)
+            return
+
+        origin_hosts = _origin_hosts(scope)
+        # Both headers are attacker-controlled; do not fall back to request Host.
+        if any(not self._is_allowed(origin_host) for origin_host in origin_hosts):
+            await Response("Forbidden Origin", status_code=403)(scope, receive, send)
+            return
+
+        await self.app(scope, receive, send)

airbyte/mcp/http_main.py

@@ -33,6 +33,12 @@
 - `MCP_SERVER_URL`: Public base URL. Used for OIDC redirect callbacks and to
   derive the MCP endpoint mount path (serves at `/` when the URL has a path
   prefix, otherwise defaults to `/mcp`).
+- `AIRBYTE_MCP_ALLOWED_HOSTS`: Comma-separated allowed hostnames or `fnmatch`
+  patterns for HTTP `Host` and `Origin` validation. Ports are ignored: an entry
+  may carry one for readability, but matching is on hostname only, so
+  `example.com:8443` also allows `example.com` on any port.
+- `AIRBYTE_MCP_HTTP_HOST`: Host interface to bind for the HTTP server. Defaults
+  to `0.0.0.0`.
 
 Interactive OIDC (Keycloak Authorization Code + PKCE), enabled when the client
 credentials are set:
@@ -86,6 +92,11 @@
     client_credentials_enabled,
     wrap_if_enabled,
 )
+from airbyte.mcp._transport_security import (
+    HTTP_HOST_ENV,
+    HostOriginGuardMiddleware,
+    resolve_allowed_hosts,
+)
 from airbyte.mcp.server import (
     DEFAULT_HTTP_HOST,
     DEFAULT_HTTP_PORT,
@@ -99,6 +110,7 @@
 
 if TYPE_CHECKING:
     from fastmcp.server.auth import AuthProvider
+    from starlette.types import ASGIApp
 
 logger = logging.getLogger(__name__)
 
@@ -234,9 +246,13 @@ def main() -> None:
 
     _log_auth_status()
 
+    http_host = _env_or_default(HTTP_HOST_ENV, DEFAULT_HTTP_HOST)
+    allowed_hosts = resolve_allowed_hosts(server_url)
+    logger.info("Resolved HTTP transport allowed hosts: %s", allowed_hosts)
+
     logger.info(
         "Starting Airbyte MCP HTTP server on %s:%d (mcp_path=%r)",
-        DEFAULT_HTTP_HOST,
+        http_host,
         DEFAULT_HTTP_PORT,
         mcp_path,
     )
@@ -247,14 +263,20 @@ def main() -> None:
     # entrypoint (a permanent gate; the per-request filter also forces it off).
     assert_http_trusted_execution_disabled(app)
 
+    def wrap_http_app(http_app: ASGIApp) -> ASGIApp:
+        return HostOriginGuardMiddleware(
+            wrap_if_enabled(http_app),
+            allowed_hosts,
+        )
+
     try:
         run_mcp_http_server(
             app,
             path=mcp_path,
             transport="streamable-http",
             stateless_http=True,
-            wrapper=wrap_if_enabled,
-            host=DEFAULT_HTTP_HOST,
+            wrapper=wrap_http_app,
+            host=http_host,
             port=DEFAULT_HTTP_PORT,
         )
     except KeyboardInterrupt:

tests/unit_tests/test_mcp_auth.py

@@ -23,6 +23,7 @@
 from airbyte.mcp import _client_credentials as client_credentials
 from airbyte.mcp import http_main
 from airbyte.mcp import server
+from airbyte.mcp._transport_security import HostOriginGuardMiddleware
 
 
 if TYPE_CHECKING:
@@ -167,14 +168,12 @@ def capture_run(app: object, **kwargs: object) -> None:
 
     http_main.main()
 
-    assert config == {
-        "path": "/mcp",
-        "transport": "streamable-http",
-        "stateless_http": True,
-        "wrapper": http_main.wrap_if_enabled,
-        "host": http_main.DEFAULT_HTTP_HOST,
-        "port": http_main.DEFAULT_HTTP_PORT,
-    }
+    assert config["path"] == "/mcp"
+    assert config["transport"] == "streamable-http"
+    assert config["stateless_http"] is True
+    assert config["host"] == http_main.DEFAULT_HTTP_HOST
+    assert config["port"] == http_main.DEFAULT_HTTP_PORT
+    assert isinstance(config["wrapper"](object()), HostOriginGuardMiddleware)
 
 
 @pytest.mark.parametrize(

tests/unit_tests/test_mcp_transport_security.py

@@ -0,0 +1,124 @@
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+"""Unit tests for HTTP MCP host and origin validation."""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+from starlette.responses import PlainTextResponse
+from starlette.testclient import TestClient
+from starlette.types import Receive, Scope, Send
+
+from airbyte.mcp._transport_security import (
+    ALLOWED_HOSTS_ENV,
+    HostOriginGuardMiddleware,
+    resolve_allowed_hosts,
+)
+
+
+async def _app(scope: Scope, receive: Receive, send: Send) -> None:
+    if scope["type"] == "lifespan":
+        while True:
+            message = await receive()
+            if message["type"] == "lifespan.startup":
+                await send({"type": "lifespan.startup.complete"})
+            elif message["type"] == "lifespan.shutdown":
+                await send({"type": "lifespan.shutdown.complete"})
+                return
+    await PlainTextResponse("ok")(scope, receive, send)
+
+
+def _client(allowed_hosts: tuple[str, ...]) -> TestClient:
+    return TestClient(HostOriginGuardMiddleware(_app, allowed_hosts))
+
+
+@pytest.mark.parametrize("host", ["localhost", "127.0.0.1"])
+def test_allowed_loopback_hosts_pass_through(host: str) -> None:
+    with _client(("localhost", "127.0.0.1")) as client:
+        response = client.get("/", headers={"host": host})
+
+    assert response.status_code == 200
+    assert response.text == "ok"
+
+
+def test_unallowed_host_is_rejected() -> None:
+    with _client(("localhost",)) as client:
+        response = client.get("/", headers={"host": "attacker.example:8080"})
+
+    assert response.status_code == 421
+    assert response.text == "Misdirected Request"
+
+
+def test_unallowed_origin_is_rejected_even_with_allowed_host() -> None:
+    with _client(("localhost",)) as client:
+        response = client.get(
+            "/",
+            headers={
+                "host": "localhost",
+                "origin": "http://attacker.example:8080",
+            },
+        )
+
+    assert response.status_code == 403
+    assert response.text == "Forbidden Origin"
+
+
+def test_missing_origin_is_allowed() -> None:
+    with _client(("localhost",)) as client:
+        response = client.get("/", headers={"host": "localhost"})
+
+    assert response.status_code == 200
+
+
+def test_missing_host_is_rejected() -> None:
+    with _client(("localhost",)) as client:
+        response = client.get("/", headers={"host": ""})
+
+    assert response.status_code == 421
+
+
+def test_duplicate_hosts_are_rejected() -> None:
+    messages: list[dict[str, object]] = []
+
+    async def receive() -> dict[str, object]:
+        return {"type": "http.request", "body": b""}
+
+    async def send(message: dict[str, object]) -> None:
+        messages.append(message)
+
+    scope: Scope = {
+        "type": "http",
+        "method": "GET",
+        "path": "/",
+        "raw_path": b"/",
+        "query_string": b"",
+        "headers": [
+            (b"host", b"localhost"),
+            (b"host", b"attacker.example"),
+        ],
+    }
+
+    async def call_middleware() -> None:
+        await HostOriginGuardMiddleware(_app, ("localhost",))(scope, receive, send)
+
+    asyncio.run(call_middleware())
+
+    assert messages[0]["status"] == 421
+
+
+def test_configured_hosts_are_allowed(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv(ALLOWED_HOSTS_ENV, "custom.example:8443,*.run.app")
+
+    with _client(resolve_allowed_hosts("")) as client:
+        custom_response = client.get("/", headers={"host": "custom.example"})
+        pattern_response = client.get("/", headers={"host": "service.run.app"})
+
+    assert custom_response.status_code == 200
+    assert pattern_response.status_code == 200
+
+
+def test_server_url_hostname_is_allowed() -> None:
+    allowed_hosts = resolve_allowed_hosts("https://mcp.example.com:443/mcp")
+
+    assert allowed_hosts[-1:] == ("mcp.example.com",)