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