airbyte_ops_mcp.cli.secrets

  1# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
  2"""CLI commands for connector integration-test secrets.
  3
  4Commands:
  5    airbyte-ops secrets fetch   - Download connector secrets from GSM into `./secrets/`
  6    airbyte-ops secrets list    - List connector secrets visible in GSM
  7    airbyte-ops secrets ci-mask - Emit GitHub Actions `::add-mask::` lines
  8
  9These commands operate on the connector integration-test secrets stored in
 10Google Secret Manager (project `dataline-integration-testing` by default).
 11They are intended to replace the `airbyte-cdk secrets` group in a subsequent
 12phase; the CDK commands remain functional until that migration lands.
 13
 14> **Note:** `airbyte-ops devin secret-*` is a different subsystem for
 15> Devin-internal 1Password on-demand secrets and is unrelated to this group.
 16
 17## CLI reference
 18
 19The commands below are regenerated by `poe docs-generate` via cyclopts's
 20programmatic docs API; see `docs/generate_cli.py`.
 21
 22.. include:: ../../../docs/generated/cli/secrets.md
 23   :start-line: 2
 24"""
 25
 26from __future__ import annotations
 27
 28__all__: list[str] = []
 29
 30from typing import Annotated
 31
 32from airbyte_cdk.utils.connector_paths import resolve_connector_name_and_directory
 33from cyclopts import Parameter
 34from rich.console import Console
 35from rich.table import Table
 36
 37from airbyte_ops_mcp.cli._base import App, app
 38from airbyte_ops_mcp.cli._shared import error_console
 39from airbyte_ops_mcp.connector_secrets import (
 40    DEFAULT_GCP_PROJECT_ID,
 41    ConnectorSecretWithNoValidVersionsError,
 42    extract_gcp_secret_name,
 43    fetch_secret_handles,
 44    get_gcp_secret_url,
 45    get_gsm_secrets_client,
 46    get_secret_filepath,
 47    get_secrets_dir,
 48    write_secret_file,
 49)
 50from airbyte_ops_mcp.connector_secrets.ci_masks import (
 51    is_running_in_ci,
 52    print_ci_secrets_masks,
 53)
 54
 55secrets_app = App(
 56    name="secrets", help="Manage connector integration-test secrets in GSM."
 57)
 58app.command(secrets_app)
 59
 60_CONNECTOR_HELP = (
 61    "Connector name (e.g. 'source-pokeapi') or path to a connector directory. "
 62    "If omitted, the current working directory is used and must be a connector "
 63    "directory."
 64)
 65_GCP_PROJECT_ID_HELP = (
 66    "GCP project ID for retrieving integration-test credentials. Defaults to "
 67    "the `GCP_PROJECT_ID` environment variable, or 'dataline-integration-testing' "
 68    "when unset."
 69)
 70
 71
 72@secrets_app.command(name="fetch")
 73def fetch(
 74    connector: Annotated[str | None, Parameter(help=_CONNECTOR_HELP)] = None,
 75    *,
 76    gcp_project_id: Annotated[
 77        str, Parameter(help=_GCP_PROJECT_ID_HELP)
 78    ] = DEFAULT_GCP_PROJECT_ID,
 79    print_ci_secrets_masks_flag: Annotated[
 80        bool | None,
 81        Parameter(
 82            name=["--print-ci-secrets-masks"],
 83            negative=["--no-print-ci-secrets-masks"],
 84            help=(
 85                "Emit GitHub Actions `::add-mask::` lines for fetched secrets. "
 86                "Ignored outside CI (requires the `CI` env var). Defaults to "
 87                "auto-detection from `CI`."
 88            ),
 89        ),
 90    ] = None,
 91) -> None:
 92    """Fetch connector secrets from GSM into the connector's `secrets/` directory.
 93
 94    Equivalent to the old `airbyte-cdk secrets fetch`. Runs in three steps:
 95
 96    1. Resolve the connector (from `[CONNECTOR]`, or the cwd).
 97    2. List GSM secrets labeled `connector=<name>` in `--gcp-project-id`.
 98    3. Write each secret's latest enabled version to `secrets/<filename>.json`
 99       (mode `0o600`). The filename comes from the secret's `filename` label,
100       or defaults to `config.json`.
101
102    When `--print-ci-secrets-masks` is set (or auto-detected in CI), emits
103    GitHub Actions masking commands after writing.
104    """
105    error_console.print("Fetching secrets...")
106
107    client = get_gsm_secrets_client()
108    connector_name, connector_directory = resolve_connector_name_and_directory(
109        connector
110    )
111    secrets_dir = get_secrets_dir(
112        connector_directory=connector_directory,
113        ensure_exists=True,
114    )
115    secrets = fetch_secret_handles(
116        connector_name=connector_name,
117        gcp_project_id=gcp_project_id,
118        client=client,
119    )
120
121    secret_count = 0
122    exceptions: list[ConnectorSecretWithNoValidVersionsError] = []
123
124    for secret in secrets:
125        secret_file_path = get_secret_filepath(secrets_dir=secrets_dir, secret=secret)
126        try:
127            write_secret_file(
128                secret=secret,
129                client=client,
130                file_path=secret_file_path,
131                connector_name=connector_name,
132                gcp_project_id=gcp_project_id,
133            )
134        except ConnectorSecretWithNoValidVersionsError as err:
135            exceptions.append(err)
136            error_console.print(
137                f"Failed to retrieve secret '{err.secret_name}': No enabled version found"
138            )
139            continue
140
141        error_console.print(f"Secret written to: {secret_file_path.absolute()!s}")
142        secret_count += 1
143
144    if secret_count == 0 and not exceptions:
145        error_console.print(f"No secrets found for connector: '{connector_name}'")
146
147    if exceptions:
148        error_console.print(
149            f"[red]Failed to retrieve {len(exceptions)} secret(s)[/red]"
150        )
151        if secret_count == 0:
152            raise exceptions[0]
153
154    emit_masks = print_ci_secrets_masks_flag
155    if emit_masks is None:
156        emit_masks = is_running_in_ci()
157
158    if emit_masks:
159        print_ci_secrets_masks(
160            secrets_dir=secrets_dir,
161            strict_ci_env_check=True,
162        )
163
164
165@secrets_app.command(name="list")
166def list_(
167    connector: Annotated[str | None, Parameter(help=_CONNECTOR_HELP)] = None,
168    *,
169    gcp_project_id: Annotated[
170        str, Parameter(help=_GCP_PROJECT_ID_HELP)
171    ] = DEFAULT_GCP_PROJECT_ID,
172) -> None:
173    """List connector secrets visible in GSM without downloading any values.
174
175    Prints a Rich table with the secret name (linked to its GCP console page),
176    labels, and creation time. Only metadata is read — no secret payloads are
177    fetched.
178    """
179    error_console.print("Scanning secrets...")
180
181    if connector and "/" not in connector and "\\" not in connector:
182        connector_name = connector
183    else:
184        connector_name, _ = resolve_connector_name_and_directory(connector)
185
186    secrets = fetch_secret_handles(
187        connector_name=connector_name,
188        gcp_project_id=gcp_project_id,
189    )
190    if not secrets:
191        error_console.print(f"No secrets found for connector: '{connector_name}'")
192        return
193
194    console = Console()
195    console.print(
196        f"[green]Secrets for connector '{connector_name}' in project "
197        f"'{gcp_project_id}':[/green]"
198    )
199    table = Table(title=f"'{connector_name}' Secrets")
200    table.add_column("Name", justify="left", style="cyan", overflow="fold")
201    table.add_column("Labels", justify="left", style="magenta", overflow="fold")
202    table.add_column("Created", justify="left", style="blue", overflow="fold")
203    for secret in secrets:
204        secret_name = extract_gcp_secret_name(secret.name)
205        secret_url = get_gcp_secret_url(secret_name, gcp_project_id)
206        table.add_row(
207            f"[link={secret_url}]{secret_name}[/link]",
208            "\n".join(f"{k}={v}" for k, v in secret.labels.items()),
209            str(secret.create_time),
210        )
211    console.print(table)
212
213
214@secrets_app.command(name="ci-mask")
215def ci_mask(
216    connector: Annotated[str | None, Parameter(help=_CONNECTOR_HELP)] = None,
217) -> None:
218    """Emit GitHub Actions `::add-mask::` lines for a connector's fetched secrets.
219
220    Reads every `*.json` file in the connector's local `secrets/` directory and
221    prints a mask command for each secret-valued property (matched against the
222    canonical
223    [`specs_secrets_mask.yaml`](https://connectors.airbyte.com/files/registries/v0/specs_secrets_mask.yaml)
224    list). Intended to run inside a GitHub Actions job; refuses to emit output
225    when `CI` is unset to prevent accidentally printing secret values locally.
226
227    This subcommand is equivalent to running `airbyte-ops secrets fetch
228    --print-ci-secrets-masks` without re-downloading secrets.
229    """
230    _, connector_directory = resolve_connector_name_and_directory(connector)
231    secrets_dir = get_secrets_dir(
232        connector_directory=connector_directory,
233        ensure_exists=False,
234    )
235    if not secrets_dir.is_dir():
236        error_console.print(
237            f"No secrets directory found at {secrets_dir!s}; nothing to mask."
238        )
239        return
240
241    print_ci_secrets_masks(
242        secrets_dir=secrets_dir,
243        strict_ci_env_check=True,
244    )