diff --git a/airbyte/mcp/__init__.py b/airbyte/mcp/__init__.py index 61a646927..7065a9503 100644 --- a/airbyte/mcp/__init__.py +++ b/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 diff --git a/airbyte/mcp/_transport_security.py b/airbyte/mcp/_transport_security.py new file mode 100644 index 000000000..7aa62c924 --- /dev/null +++ b/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) diff --git a/airbyte/mcp/http_main.py b/airbyte/mcp/http_main.py index d41776e94..df17aef52 100644 --- a/airbyte/mcp/http_main.py +++ b/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: diff --git a/tests/unit_tests/test_mcp_auth.py b/tests/unit_tests/test_mcp_auth.py index 6fcc639a3..26fb43885 100644 --- a/tests/unit_tests/test_mcp_auth.py +++ b/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( diff --git a/tests/unit_tests/test_mcp_transport_security.py b/tests/unit_tests/test_mcp_transport_security.py new file mode 100644 index 000000000..ceb5b8e84 --- /dev/null +++ b/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",)