feat(mcp): wire trusted-execution gate and function-layer guards

airbyte/constants.py

@@ -281,6 +281,17 @@ def _str_to_bool(value: str) -> bool:
 If set, only tools from these domains will be advertised by the MCP server.
 """
 
+MCP_TRUSTED_EXECUTION_ENV_VAR: str = "AIRBYTE_MCP_TRUSTED_EXECUTION"
+"""Environment variable that enables trusted (local) execution for the MCP server.
+
+When set to `1`/`true`/`yes`, the server may use its trusted-machine capabilities: local
+filesystem access, local connector installation/execution, and server-side secret
+resolution. It defaults to *off* on every transport and is permanently unavailable over
+the HTTP transport (a hosted deployment can never enable it). This gate is server-owned
+and is deliberately never read from a request header, because it *widens* the surface and
+so must never be caller-controllable.
+"""
+
 MCP_WORKSPACE_ID_HEADER: str = "X-Airbyte-Workspace-Id"
 """HTTP header key for passing workspace ID to the MCP server.
 

airbyte/exceptions.py

@@ -222,6 +222,34 @@ class PyAirbyteNoStreamsSelectedError(PyAirbyteInputError):
     available_streams: list[str] | None = None
 
 
+# MCP Server Errors
+
+
+@dataclass
+class AirbyteMCPError(PyAirbyteError):
+    """An error occurred in the Airbyte MCP server."""
+
+
+@dataclass
+class AirbyteTrustedExecutionRequiredError(AirbyteMCPError):
+    """A trusted-execution-only capability was invoked while trusted execution is disabled.
+
+    Trusted execution grants the MCP server its trusted-machine capabilities: local
+    filesystem access, local connector installation/execution, and server-side secret
+    resolution. It defaults to *off* on every transport and is permanently unavailable
+    over the HTTP transport, so a backend helper that exposes one of those capabilities
+    hard-fails when the gate is disabled -- independently of whether the corresponding
+    tool was hidden from the tool listing.
+    """
+
+    guidance = (
+        "Set `AIRBYTE_MCP_TRUSTED_EXECUTION=1` on the MCP server process and restart it. "
+        "Trusted execution is only available on the stdio transport; it can never be "
+        "enabled for an HTTP/hosted deployment."
+    )
+    feature: str | None = None
+
+
 # Normalization Errors
 
 

airbyte/mcp/_arg_resolvers.py

@@ -13,10 +13,23 @@
 
 import yaml
 
+from airbyte.constants import SECRETS_HYDRATION_PREFIX
+from airbyte.mcp._guards import raise_if_untrusted_execution_context
 from airbyte.secrets.hydration import deep_update, detect_hardcoded_secrets
 from airbyte.secrets.util import get_secret
 
 
+def _contains_secret_reference(value: object) -> bool:
+    """Return whether `value` contains a `secret_reference::` string at any depth."""
+    if isinstance(value, str):
+        return value.startswith(SECRETS_HYDRATION_PREFIX)
+    if isinstance(value, dict):
+        return any(_contains_secret_reference(item) for item in value.values())
+    if isinstance(value, list):
+        return any(_contains_secret_reference(item) for item in value)
+    return False
+
+
 # Hint: Null result if input is Null
 @overload
 def resolve_list_of_strings(value: None) -> None: ...
@@ -88,13 +101,24 @@ def resolve_connector_config(  # noqa: PLR0912
         ValueError: If JSON parsing fails or a provided input is invalid
 
     We reject hardcoded secrets in a config dict if we detect them.
+
+    The trusted-machine inputs are gated by trusted execution
+    (`airbyte.mcp._guards.raise_if_untrusted_execution_context`): reading a local `config_file`,
+    resolving a server-side `config_secret_name`, and hydrating inline
+    `secret_reference::` values each hard-fail when trusted execution is disabled. Passing
+    an already-resolved inline `config` dict remains available to untrusted callers (for
+    example hosted cloud tools), so only the paths that touch the server's filesystem or
+    secret store are restricted.
     """
     config_dict: dict[str, Any] = {}
 
     if config is None and config_file is None and config_secret_name is None:
         return {}
 
     if config_file is not None:
+        raise_if_untrusted_execution_context(
+            "Reading connector config from a local file (`config_file`)"
+        )
         if isinstance(config_file, str):
             config_file = Path(config_file)
 
@@ -137,6 +161,11 @@ def _raise_invalid_type(file_config: object) -> None:
         else:
             raise ValueError(f"Config must be a dict or JSON string, got: {type(config).__name__}")
 
+    if _contains_secret_reference(config_dict):
+        raise_if_untrusted_execution_context(
+            "Resolving inline secret references (`secret_reference::`) in connector config"
+        )
+
     if config_dict and config_spec_jsonschema is not None:
         hardcoded_secrets: list[list[str]] = detect_hardcoded_secrets(
             config=config_dict,
@@ -156,6 +185,9 @@ def _raise_invalid_type(file_config: object) -> None:
             raise ValueError(error_msg)
 
     if config_secret_name is not None:
+        raise_if_untrusted_execution_context(
+            "Resolving connector config from a server-side secret (`config_secret_name`)"
+        )
         # Assume this is a secret name that points to a JSON/YAML config.
         secret_config = yaml.safe_load(str(get_secret(config_secret_name)))
         if not isinstance(secret_config, dict):

airbyte/mcp/_config.py

@@ -9,6 +9,7 @@
 import dotenv
 
 from airbyte._util.meta import is_interactive
+from airbyte.mcp._guards import is_trusted_execution_enabled
 from airbyte.secrets import (
     DotenvSecretManager,
     GoogleGSMSecretManager,
@@ -39,25 +40,39 @@ def load_secrets_to_env_vars() -> None:
     from the loaded environment variables.
 
     Note: Later secret manager registrations have higher priority than earlier ones.
+
+    When trusted execution is disabled (`airbyte.mcp._guards.is_trusted_execution_enabled`),
+    the dotenv and Google Secret Manager backends are *not* registered as secret sources.
+    This prevents an untrusted (for example hosted HTTP) deployment from resolving
+    server-side secrets into connector config via `config_secret_name` /
+    `secret_reference::`. Ordinary environment variables are still loaded from any
+    referenced dotenv file so that safe, non-secret-manager configuration continues to
+    work.
     """
+    trusted = is_trusted_execution_enabled()
+
     # Load the .env file from the current working directory.
     envrc_path = Path.cwd() / ".envrc"
     if envrc_path.exists():
-        envrc_secret_mgr = DotenvSecretManager(envrc_path)
         _load_dotenv_file(envrc_path)
-        register_secret_manager(
-            envrc_secret_mgr,
-        )
+        if trusted:
+            register_secret_manager(
+                DotenvSecretManager(envrc_path),
+            )
 
     if AIRBYTE_MCP_DOTENV_PATH_ENVVAR in os.environ:
         dotenv_path = Path(os.environ[AIRBYTE_MCP_DOTENV_PATH_ENVVAR]).absolute()
-        custom_dotenv_secret_mgr = DotenvSecretManager(dotenv_path)
         _load_dotenv_file(dotenv_path)
-        register_secret_manager(
-            custom_dotenv_secret_mgr,
-        )
+        if trusted:
+            register_secret_manager(
+                DotenvSecretManager(dotenv_path),
+            )
 
-    if is_secret_available("GCP_GSM_CREDENTIALS") and is_secret_available("GCP_GSM_PROJECT_ID"):
+    if (
+        trusted
+        and is_secret_available("GCP_GSM_CREDENTIALS")
+        and is_secret_available("GCP_GSM_PROJECT_ID")
+    ):
         # Initialize the GoogleGSMSecretManager if the credentials and project are set.
         register_secret_manager(
             GoogleGSMSecretManager(

airbyte/mcp/_guards.py

@@ -0,0 +1,49 @@
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+"""Trusted-execution guards for Airbyte MCP backend helpers.
+
+Trusted execution is the master gate for the MCP server's *trusted-machine* capabilities:
+local filesystem access, local connector installation/execution, and server-side secret
+resolution. It is controlled solely by the `AIRBYTE_MCP_TRUSTED_EXECUTION` server
+environment variable (`airbyte.constants.MCP_TRUSTED_EXECUTION_ENV_VAR`) and defaults to
+*off* on every transport.
+
+`fastmcp_extensions` already hides trusted-machine tools from the tool listing when the
+gate is off, but that is a *visibility* control. The guards here are an independent
+*function-layer* control: backend helpers call `raise_if_untrusted_execution_context` so a direct
+call hard-fails when the gate is disabled, even if a future registration mistake left the
+tool visible. Because the two layers are independent, a mistake in either one alone cannot
+expose a trusted-machine capability to an untrusted (for example hosted HTTP) caller.
+"""
+
+from __future__ import annotations
+
+import os
+
+from airbyte.constants import MCP_TRUSTED_EXECUTION_ENV_VAR
+from airbyte.exceptions import AirbyteTrustedExecutionRequiredError
+
+
+_TRUTHY_VALUES = frozenset({"1", "true", "yes"})
+
+
+def is_trusted_execution_enabled() -> bool:
+    """Return whether trusted execution is enabled for the MCP server.
+
+    Reads `AIRBYTE_MCP_TRUSTED_EXECUTION` from the server environment only. A value of
+    `1`/`true`/`yes` (case-insensitive) enables it; anything else -- including unset --
+    leaves it disabled.
+    """
+    return os.environ.get(MCP_TRUSTED_EXECUTION_ENV_VAR, "0").strip().lower() in _TRUTHY_VALUES
+
+
+def raise_if_untrusted_execution_context(feature: str) -> None:
+    """Hard-fail when `feature` is invoked while trusted execution is disabled.
+
+    Call this from any backend helper that exposes a trusted-machine capability (local
+    filesystem access, connector installation/execution, or server-side secret
+    resolution). It raises `airbyte.exceptions.AirbyteTrustedExecutionRequiredError`
+    when the gate is off, independently of whether the corresponding tool was hidden from
+    the listing.
+    """
+    if not is_trusted_execution_enabled():
+        raise AirbyteTrustedExecutionRequiredError(feature=feature)

airbyte/mcp/_tool_utils.py

@@ -27,6 +27,7 @@
 from fastmcp_extensions.tool_filters import (
     ANNOTATION_MCP_MODULE,
     ANNOTATION_READ_ONLY_HINT,
+    CONFIG_TRUSTED_EXECUTION,
     get_annotation,
 )
 
@@ -52,8 +53,10 @@
     MCP_DOMAINS_DISABLED_ENV_VAR,
     MCP_DOMAINS_ENV_VAR,
     MCP_READONLY_MODE_ENV_VAR,
+    MCP_TRUSTED_EXECUTION_ENV_VAR,
     MCP_WORKSPACE_ID_HEADER,
 )
+from airbyte.exceptions import PyAirbyteInputError
 
 
 if TYPE_CHECKING:
@@ -155,6 +158,20 @@ def check_guid_created_in_session(guid: str) -> None:
 )
 """Config arg for legacy AIRBYTE_MCP_DOMAINS env var."""
 
+TRUSTED_EXECUTION_CONFIG_ARG = MCPServerConfigArg(
+    name=CONFIG_TRUSTED_EXECUTION,
+    env_var=MCP_TRUSTED_EXECUTION_ENV_VAR,
+    default="0",
+    required=False,
+)
+"""Config arg mapping the generic `trusted_execution` gate to `AIRBYTE_MCP_TRUSTED_EXECUTION`.
+
+Registering this lets `fastmcp_extensions` resolve the trusted-execution gate from the
+Airbyte-specific env var while keeping the generic library Airbyte-agnostic. It
+deliberately has no `http_header_key`: the gate *widens* the tool surface, so it must never
+be caller-controllable.
+"""
+
 WORKSPACE_ID_CONFIG_ARG = MCPServerConfigArg(
     name=MCP_CONFIG_WORKSPACE_ID,
     http_header_key=MCP_WORKSPACE_ID_HEADER,
@@ -418,6 +435,77 @@ def airbyte_module_filter(tool: Tool, app: FastMCP) -> bool:
     return True
 
 
+def _known_mcp_modules() -> set[str]:
+    """Return the set of Airbyte MCP domain (module) names that have registered tools."""
+    modules: set[str] = set()
+    for _func, tool_annotations in _REGISTERED_TOOLS:
+        module = tool_annotations.get(ANNOTATION_MCP_MODULE)
+        if module:
+            modules.add(_normalize_mcp_module(str(module)))
+    for _provider_factory, tool_annotations in _REGISTERED_PROVIDERS:
+        module = tool_annotations.get(ANNOTATION_MCP_MODULE)
+        if module:
+            modules.add(_normalize_mcp_module(str(module)))
+    return modules
+
+
+def validate_airbyte_domains(app: FastMCP) -> None:
+    """Validate the `AIRBYTE_MCP_DOMAINS` / `AIRBYTE_MCP_DOMAINS_DISABLED` selection.
+
+    Hard-fails at startup instead of silently dropping a domain that was explicitly
+    requested. Two incompatibilities are rejected:
+
+    1. Setting both `AIRBYTE_MCP_DOMAINS` (include) and `AIRBYTE_MCP_DOMAINS_DISABLED`
+       (exclude), which are mutually exclusive -- `airbyte_module_filter` would otherwise
+       silently honor only the exclude list and drop the requested includes.
+    2. Naming a domain that has no registered tools (typically a typo), which would
+       otherwise silently expose or hide nothing for that name.
+
+    Call this once at startup, after all tools are registered.
+
+    Args:
+        app: The FastMCP app instance.
+
+    Raises:
+        PyAirbyteInputError: If the domain configuration is incompatible.
+    """
+    exclude_modules = _parse_csv_config(get_mcp_config(app, MCP_CONFIG_EXCLUDE_MODULES) or "")
+    include_modules = _parse_csv_config(get_mcp_config(app, MCP_CONFIG_INCLUDE_MODULES) or "")
+
+    if include_modules and exclude_modules:
+        raise PyAirbyteInputError(
+            message=(
+                "AIRBYTE_MCP_DOMAINS and AIRBYTE_MCP_DOMAINS_DISABLED are mutually exclusive."
+            ),
+            guidance=(
+                "Clear one of `AIRBYTE_MCP_DOMAINS` (include) or `AIRBYTE_MCP_DOMAINS_DISABLED` "
+                "(exclude) so only one is set, then restart the MCP server process."
+            ),
+            context={
+                "include_domains": include_modules,
+                "exclude_domains": exclude_modules,
+            },
+        )
+
+    known_modules = _known_mcp_modules()
+    unknown_modules = sorted(
+        {module for module in (*include_modules, *exclude_modules) if module not in known_modules}
+    )
+    if unknown_modules:
+        raise PyAirbyteInputError(
+            message="One or more requested MCP domains are not recognized.",
+            guidance=(
+                "Correct the unknown domain name(s) in `AIRBYTE_MCP_DOMAINS` / "
+                "`AIRBYTE_MCP_DOMAINS_DISABLED` (or clear the variable), then restart the MCP "
+                "server process."
+            ),
+            context={
+                "unknown_domains": unknown_modules,
+                "known_domains": sorted(known_modules),
+            },
+        )
+
+
 def airbyte_ui_support_filter(tool: Tool, _app: FastMCP) -> bool:
     """Filter tools that require MCP Apps UI support."""
     if not get_annotation(tool, INTERACTIVE_UI_ANNOTATION, default=False):

airbyte/mcp/http_main.py

@@ -37,7 +37,10 @@
 import os
 from urllib.parse import urlparse
 
-from fastmcp_extensions import register_landing_page
+from fastmcp_extensions import (
+    assert_http_trusted_execution_disabled,
+    register_landing_page,
+)
 
 from airbyte.mcp.server import (
     DEFAULT_HTTP_HOST,
@@ -103,6 +106,12 @@ def main() -> None:
         mcp_path,
     )
 
+    # Trusted execution grants local filesystem, connector-execution, and
+    # server-side secret-resolution capability, which must never be reachable
+    # over HTTP. Hard-fail startup if it was explicitly enabled on this hosted
+    # entrypoint (a permanent gate; the per-request filter also forces it off).
+    assert_http_trusted_execution_disabled(app)
+
     app.run(
         transport="streamable-http",
         host=DEFAULT_HTTP_HOST,