airbyte_ops_mcp.mcp._client_credentials

 1# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
 2"""Opt-in HTTP Basic client-credentials transport auth for the MCP server.
 3
 4The headless bearer path verifies an already-minted, short-lived (~15 min) JWT.
 5That works for MCP clients that run the OAuth flow and refresh tokens
 6automatically, but not for a truly headless agent that can only set a *static*
 7`Authorization` header value and cannot re-mint on a timer.
 8
 9This module bridges that gap, behind an opt-in flag. When enabled, the server
10accepts the long-lived `client_id` / `client_secret` presented on the inbound MCP
11request via standard HTTP Basic auth
12(`Authorization: Basic base64(client_id:client_secret)`, the same credential
13encoding OAuth's `client_secret_basic` uses). The server then runs a
14client-credentials exchange against the Airbyte token endpoint to obtain a
15short-lived access token and rewrites the request to `Authorization: Bearer
16<token>` so the existing `JWTVerifier` validates it unchanged. The agent thus
17presents a durable credential once; the server owns the short-lived-token churn.
18
19The provider-neutral exchange middleware lives in `fastmcp_extensions`
20(`wrap_client_credentials`). This module owns only the Airbyte-specific policy:
21the opt-in env toggle and the Airbyte Cloud token endpoint (overridable for
22self-hosted deployments). It resolves those to plain values and hands them to
23the generic library, so no Airbyte literal or env-var name leaks into the lib.
24
25This mirrors PyAirbyte's `airbyte.mcp._client_credentials`, since the Ops MCP
26server shares Airbyte Cloud's realm and token endpoint.
27"""
28
29from __future__ import annotations
30
31import os
32from typing import TYPE_CHECKING
33
34from fastmcp_extensions import wrap_client_credentials
35
36if TYPE_CHECKING:
37    from collections.abc import Mapping
38
39    from starlette.types import ASGIApp
40
41
42# Opt-in flag. Off by default: accepting long-lived credentials at the transport
43# is a deliberate escalation, so a deployment must explicitly turn it on.
44ALLOW_CLIENT_CREDENTIALS_ENV = "AIRBYTE_MCP_AUTH_ALLOW_CLIENT_CREDENTIALS"
45
46# Airbyte token endpoint that mints an application access token from a
47# `client_id` / `client_secret`. Defaults to Airbyte Cloud; overridable for
48# self-hosted deployments pointing at their own Airbyte instance.
49AIRBYTE_CLOUD_TOKEN_URL = "https://api.airbyte.com/v1/applications/token"
50TOKEN_URL_ENV = "AIRBYTE_MCP_AUTH_CLIENT_CREDENTIALS_TOKEN_URL"
51
52_TRUTHY = frozenset({"1", "true", "t", "yes", "y", "on"})
53
54
55def client_credentials_enabled(env: Mapping[str, str] | None = None) -> bool:
56    """Return whether the opt-in HTTP Basic client-credentials grant is enabled."""
57    source = env if env is not None else os.environ
58    return source.get(ALLOW_CLIENT_CREDENTIALS_ENV, "").strip().lower() in _TRUTHY
59
60
61def _token_url() -> str:
62    """Return the token endpoint, defaulting to Airbyte Cloud.
63
64    A blank or whitespace-only override is treated as unset so the Airbyte Cloud
65    default still applies, rather than POSTing to an invalid URL and failing every
66    Basic-auth request closed.
67    """
68    return os.getenv(TOKEN_URL_ENV, "").strip() or AIRBYTE_CLOUD_TOKEN_URL
69
70
71def wrap_if_enabled(app: ASGIApp) -> ASGIApp:
72    """Wrap `app` with the client-credentials exchange when the flag is set.
73
74    Returns `app` unchanged when the opt-in flag is unset, so the standard
75    bearer/OIDC transport auth is the only path. When enabled, wraps `app` as the
76    outermost ASGI layer (via `fastmcp_extensions.wrap_client_credentials`) so the
77    Basic-to-Bearer rewrite happens before FastMCP's auth verifier runs.
78    """
79    return wrap_client_credentials(
80        app,
81        enabled=client_credentials_enabled(),
82        token_url=_token_url(),
83    )