feat(mcp): always-on HTTP auth via typed build_mcp_auth with Airbyte Cloud defaults

airbyte/mcp/__init__.py

@@ -151,32 +151,36 @@
 credentials that matter are your Airbyte Cloud creds in the dotenv file.
 
 When the server is instead exposed over **HTTP** (`airbyte-mcp-http` /
-`poe mcp-serve-http`), it verifies an `Authorization: Bearer <token>` on every
-request. Two client shapes are supported on the same deployment (combined
-automatically when both are configured):
+`poe mcp-serve-http`), transport auth is **always on** and targets Airbyte Cloud
+out of the box — it verifies an `Authorization: Bearer <token>` on every request
+against Airbyte Cloud's public Keycloak realms with no auth config required. Two
+client shapes are supported on the same deployment (combined automatically):
 
 ### Humans → interactive OIDC
 
-Set `OIDC_CONFIG_URL`, `OIDC_CLIENT_ID`, and `OIDC_CLIENT_SECRET`. Interactive
-clients open a browser (Keycloak Authorization Code + PKCE) and the resulting
-token is verified by the server. No bearer token to manage by hand.
+Supply `AIRBYTE_MCP_OIDC_CLIENT_ID` and `AIRBYTE_MCP_OIDC_CLIENT_SECRET` (the
+OIDC discovery URL defaults to Airbyte Cloud). Interactive clients open a browser
+(Keycloak Authorization Code + PKCE) and the resulting token is verified by the
+server. No bearer token to manage by hand.
 
 ### Machines / agents → headless bearer token
 
-There is **no** transport mode that accepts a raw `client_id` + `client_secret`
-in a header. A headless agent **mints its own short-lived bearer token** and
-sends it as `Authorization: Bearer <token>`; the server verifies the signature
-(no browser, no stored/rotating refresh token).
+By default there is **no** transport mode that accepts a raw `client_id` +
+`client_secret` in a header (see the opt-in exception below). A headless agent
+**mints its own short-lived bearer token** and sends it as
+`Authorization: Bearer <token>`; the server verifies the signature (no browser,
+no stored/rotating refresh token).
 
-For Airbyte Cloud, set `MCP_AUTH_AIRBYTE_CLOUD=true` on the server so it verifies
-tokens against Airbyte Cloud's application-client realm without hand-configuring
-URLs. The agent mints an Airbyte Cloud access token from its
+By default the server verifies against Airbyte Cloud's application-client realm,
+so an agent just mints an Airbyte Cloud access token from its
 `AIRBYTE_CLOUD_CLIENT_ID` / `AIRBYTE_CLOUD_CLIENT_SECRET` (the
 `https://api.airbyte.com/v1/applications/token` endpoint) and sends it as the
 bearer. That single token both authenticates transport (verified by the server)
 and authorizes downstream Cloud API calls, because an Airbyte-Cloud-issued token
 is itself a valid Cloud API bearer. Tokens are short-lived (~15 min), so
-re-mint on expiry / on a `401` rather than pinning a static token.
+re-mint on expiry / on a `401` rather than pinning a static token. A self-hosted
+deployment pointing at its own Airbyte instance overrides the realm via the
+`AIRBYTE_MCP_AUTH_*` vars below.
 
 Clients that support HTTP transports can pass the token via a `headers` block in
 their MCP config:
@@ -194,24 +198,58 @@
 }
 ```
 
+### Opt-in: static client credentials via HTTP Basic
+
+The headless bearer path assumes the client can re-mint a token every ~15 min.
+An agent that can only set a **static** `Authorization` value (no refresh hook)
+can't do that. For this case the server can optionally accept the long-lived
+`client_id` / `client_secret` directly, via standard HTTP Basic
+(`client_secret_basic`):
+
+```http
+Authorization: Basic base64(client_id:client_secret)
+```
+
+When `AIRBYTE_MCP_AUTH_ALLOW_CLIENT_CREDENTIALS` is truthy, the server exchanges
+those credentials for a short-lived token server-side (caching it until shortly
+before expiry) and verifies it via the same headless path, so the agent presents
+one durable credential and the server owns the token churn. This is **off by
+default**, because accepting long-lived credentials at the transport is a
+deliberate escalation; a `Bearer` request is always preferred when the client
+can manage it.
+
 ### Server environment variables (HTTP mode)
 
+All auth settings default to Airbyte Cloud's public (non-secret) Keycloak realms,
+so the hosted Airbyte Cloud deployment needs none of them except the interactive
+OIDC client credentials. Each is an override for a self-hosted deployment that
+points at its own Airbyte instance:
+
 - `MCP_SERVER_URL` — public base URL of the server (also used for OIDC redirect
   callbacks); defaults to `http://localhost:8080`.
-- `OIDC_CONFIG_URL`, `OIDC_CLIENT_ID`, `OIDC_CLIENT_SECRET` — enable interactive
-  OIDC (all three required).
-- `MCP_AUTH_AIRBYTE_CLOUD` — set truthy to verify headless tokens against Airbyte
-  Cloud's realm (fills JWKS URI / issuer / audience / algorithm with Airbyte
-  Cloud defaults; each stays overridable).
-- `MCP_AUTH_JWKS_URI` / `MCP_AUTH_JWT_PUBLIC_KEY` — JWKS URL or static public key
-  for verifying headless tokens.
-- `MCP_AUTH_ISSUER` / `MCP_AUTH_AUDIENCE` / `MCP_AUTH_ALGORITHM` — expected `iss`
-  / `aud` claims and signing algorithm (recommended for headless).
-
-If no transport auth variables are set, the HTTP server starts
-**unauthenticated** and logs a warning — acceptable only behind a trusted
-network boundary. The verifier assembly is provided by
-[`fastmcp-extensions`](https://github.com/airbytehq/fastmcp-extensions).
+- `AIRBYTE_MCP_OIDC_CLIENT_ID` / `AIRBYTE_MCP_OIDC_CLIENT_SECRET` — OAuth client
+  credentials that activate the interactive OIDC path (no default; deployment
+  supplies them).
+- `AIRBYTE_MCP_OIDC_CONFIG_URL` — OIDC discovery URL; defaults to Airbyte Cloud's
+  `airbyte` realm.
+- `AIRBYTE_MCP_AUTH_JWKS_URI` — JWKS URL for verifying headless bearer tokens;
+  defaults to Airbyte Cloud's application-client realm certs.
+- `AIRBYTE_MCP_AUTH_JWT_PUBLIC_KEY` — static public key for verifying headless
+  bearer tokens; an alternative to `AIRBYTE_MCP_AUTH_JWKS_URI` for self-hosted
+  realms without a JWKS endpoint (no default).
+- `AIRBYTE_MCP_AUTH_ISSUER` / `AIRBYTE_MCP_AUTH_AUDIENCE` /
+  `AIRBYTE_MCP_AUTH_ALGORITHM` — expected `iss` / `aud` claims and signing
+  algorithm; default to Airbyte Cloud (`.../realms/_airbyte-application-clients`,
+  `account`, `RS256`).
+- `AIRBYTE_MCP_AUTH_ALLOW_CLIENT_CREDENTIALS` — opt-in flag (off by default) that
+  enables the HTTP Basic `client_secret_basic` path described above.
+- `AIRBYTE_MCP_AUTH_CLIENT_CREDENTIALS_TOKEN_URL` — token endpoint used for that
+  exchange; defaults to Airbyte Cloud's `https://api.airbyte.com/v1/applications/token`.
+
+This module owns those Airbyte-branded names and translates them to the generic
+names consumed by
+[`fastmcp-extensions`](https://github.com/airbytehq/fastmcp-extensions), which
+assembles the verifiers and keeps the extensions library provider-agnostic.
 
 ## Troubleshooting
 

airbyte/mcp/_client_credentials.py

@@ -0,0 +1,232 @@
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+"""Opt-in HTTP Basic client-credentials transport auth for the MCP server.
+
+The headless bearer path verifies an already-minted, short-lived (~15 min) JWT.
+That works for MCP clients that run the OAuth flow and refresh tokens
+automatically, but not for a truly headless agent that can only set a *static*
+`Authorization` header value and cannot re-mint on a timer.
+
+This module bridges that gap, behind an opt-in flag. When enabled, the server
+accepts the long-lived `client_id` / `client_secret` presented via standard HTTP
+Basic auth (`Authorization: Basic base64(client_id:client_secret)`, the
+`client_secret_basic` token-endpoint auth method), exchanges them for a
+short-lived access token at the Airbyte token endpoint, and rewrites the request
+to `Authorization: Bearer <token>` so the existing `JWTVerifier` validates it
+unchanged. The agent thus presents a durable credential once; the server owns
+the short-lived-token churn.
+
+The exchange runs as the outermost ASGI layer (ahead of FastMCP's auth
+middleware) so the rewritten bearer header is what the verifier sees. Minted
+tokens are cached per credential until shortly before expiry to avoid minting on
+every request. A `Bearer` request, or any request when the flag is unset, passes
+through untouched.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import binascii
+import hashlib
+import logging
+import os
+import time
+from typing import TYPE_CHECKING
+
+import httpx
+
+
+if TYPE_CHECKING:
+    from starlette.types import ASGIApp, Receive, Scope, Send
+
+
+logger = logging.getLogger(__name__)
+
+# Opt-in flag. Off by default: accepting long-lived credentials at the transport
+# is a deliberate escalation, so a deployment must explicitly turn it on.
+ALLOW_CLIENT_CREDENTIALS_ENV = "AIRBYTE_MCP_AUTH_ALLOW_CLIENT_CREDENTIALS"
+
+# Airbyte token endpoint that mints an application access token from a
+# `client_id` / `client_secret`. Defaults to Airbyte Cloud; overridable for
+# self-hosted deployments pointing at their own Airbyte instance.
+AIRBYTE_CLOUD_TOKEN_URL = "https://api.airbyte.com/v1/applications/token"
+TOKEN_URL_ENV = "AIRBYTE_MCP_AUTH_CLIENT_CREDENTIALS_TOKEN_URL"
+
+# Re-mint this many seconds before the cached token actually expires, so a token
+# never lapses mid-request due to clock skew or in-flight latency.
+_EXPIRY_SAFETY_MARGIN_SECONDS = 60
+
+_TRUTHY = frozenset({"1", "true", "t", "yes", "y", "on"})
+
+
+def client_credentials_enabled(env: os._Environ[str] | None = None) -> bool:
+    """Return whether the opt-in HTTP Basic client-credentials grant is enabled."""
+    source = env if env is not None else os.environ
+    return source.get(ALLOW_CLIENT_CREDENTIALS_ENV, "").strip().lower() in _TRUTHY
+
+
+def _token_url() -> str:
+    """Return the token endpoint, defaulting to Airbyte Cloud."""
+    return os.getenv(TOKEN_URL_ENV, AIRBYTE_CLOUD_TOKEN_URL)
+
+
+class ClientCredentialsExchangeMiddleware:
+    """ASGI middleware that exchanges HTTP Basic client credentials for a bearer.
+
+    Runs as the outermost layer so the rewritten `Authorization: Bearer` header
+    reaches FastMCP's auth verifier. Only `Basic` requests are touched; `Bearer`
+    and unauthenticated requests pass through unchanged (the latter are then
+    rejected by the verifier, preserving fail-closed behavior).
+    """
+
+    def __init__(
+        self,
+        app: ASGIApp,
+        *,
+        token_url: str,
+        expiry_margin_seconds: int = _EXPIRY_SAFETY_MARGIN_SECONDS,
+    ) -> None:
+        self._app = app
+        self._token_url = token_url
+        self._expiry_margin_seconds = expiry_margin_seconds
+        # Maps a per-credential cache key to `(access_token, expiry_epoch)`.
+        self._token_cache: dict[str, tuple[str, float]] = {}
+        self._lock = asyncio.Lock()
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        """Rewrite a Basic-auth HTTP request to Bearer, then delegate downstream."""
+        if scope["type"] != "http":
+            await self._app(scope, receive, send)
+            return
+
+        credentials = _parse_basic_credentials(scope)
+        if credentials is None:
+            await self._app(scope, receive, send)
+            return
+
+        token = await self._token_for(credentials)
+        if token is None:
+            # Exchange failed (e.g. bad credentials): pass the request through
+            # unmodified so the verifier rejects it with a 401, rather than
+            # masking the failure here.
+            await self._app(scope, receive, send)
+            return
+
+        await self._app(_with_bearer(scope, token), receive, send)
+
+    async def _token_for(self, credentials: tuple[str, str]) -> str | None:
+        """Return a valid access token for the credentials, minting if needed."""
+        client_id, client_secret = credentials
+        cache_key = _cache_key(client_id, client_secret)
+        now = time.monotonic()
+
+        async with self._lock:
+            cached = self._token_cache.get(cache_key)
+            if cached is not None and cached[1] > now:
+                return cached[0]
+
+            minted = await self._mint_token(client_id, client_secret)
+            if minted is None:
+                return None
+
+            token, expires_in = minted
+            expiry = now + max(expires_in - self._expiry_margin_seconds, 0)
+            self._token_cache[cache_key] = (token, expiry)
+            return token
+
+    async def _mint_token(self, client_id: str, client_secret: str) -> tuple[str, float] | None:
+        """Exchange client credentials for `(access_token, expires_in)` or `None`.
+
+        Returns `None` when the token endpoint rejects the credentials or omits
+        an access token; never logs the credentials or the minted token.
+        """
+        async with httpx.AsyncClient(timeout=10.0) as client:
+            response = await client.post(
+                self._token_url,
+                json={"client_id": client_id, "client_secret": client_secret},
+                headers={"Content-Type": "application/json"},
+            )
+
+        if response.status_code != httpx.codes.OK:
+            logger.warning(
+                "Client-credentials token exchange failed (HTTP %d); passing "
+                "request through for the verifier to reject.",
+                response.status_code,
+            )
+            return None
+
+        payload = response.json()
+        access_token = payload.get("access_token")
+        if not access_token:
+            logger.warning(
+                "Client-credentials token exchange returned no access token; "
+                "passing request through for the verifier to reject."
+            )
+            return None
+
+        # `expires_in` is advisory; fall back to a conservative default so a
+        # missing field can't produce a non-expiring cache entry.
+        expires_in = payload.get("expires_in")
+        return access_token, float(expires_in) if expires_in else 0.0
+
+
+def _parse_basic_credentials(scope: Scope) -> tuple[str, str] | None:
+    """Return `(client_id, client_secret)` from a Basic `Authorization` header.
+
+    Returns `None` when the header is absent, uses a non-Basic scheme, or is
+    malformed, so the request passes through for the verifier to handle.
+    """
+    for name, value in scope.get("headers", []):
+        if name == b"authorization":
+            return _decode_basic(value)
+    return None
+
+
+def _decode_basic(header_value: bytes) -> tuple[str, str] | None:
+    """Decode a `Basic <base64(client_id:client_secret)>` header value."""
+    scheme, _, encoded = header_value.partition(b" ")
+    if scheme.lower() != b"basic" or not encoded:
+        return None
+    try:
+        decoded = base64.b64decode(encoded, validate=True).decode("utf-8")
+    except (binascii.Error, UnicodeDecodeError):
+        return None
+    client_id, sep, client_secret = decoded.partition(":")
+    if not sep:
+        return None
+    return client_id, client_secret
+
+
+def _cache_key(client_id: str, client_secret: str) -> str:
+    """Return a stable, non-reversible cache key for a credential pair.
+
+    Hashes the secret so plaintext credentials never sit in the cache dict.
+    """
+    return hashlib.sha256(f"{client_id}:{client_secret}".encode()).hexdigest()
+
+
+def _with_bearer(scope: Scope, token: str) -> Scope:
+    """Return a shallow copy of `scope` with `Authorization` set to `Bearer`."""
+    headers = [(name, value) for name, value in scope["headers"] if name != b"authorization"]
+    headers.append((b"authorization", b"Bearer " + token.encode("ascii")))
+    new_scope = dict(scope)
+    new_scope["headers"] = headers
+    return new_scope
+
+
+def wrap_if_enabled(app: ASGIApp) -> ASGIApp:
+    """Wrap `app` with the client-credentials exchange when the flag is set.
+
+    Returns `app` unchanged when the opt-in flag is unset, so the standard
+    bearer/OIDC transport auth is the only path. When enabled, returns `app`
+    wrapped as the outermost ASGI layer so the Basic-to-Bearer rewrite happens
+    before FastMCP's auth verifier runs.
+    """
+    if not client_credentials_enabled():
+        return app
+    logger.info(
+        "HTTP Basic client-credentials transport auth is enabled (%s); the "
+        "server will exchange presented client credentials for bearer tokens.",
+        ALLOW_CLIENT_CREDENTIALS_ENV,
+    )
+    return ClientCredentialsExchangeMiddleware(app, token_url=_token_url())