feat(secrets): gsm:// / env:// reference resolution in config (Google Secret Manager)

CHANGELOG.md

@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- **Secret manager references** for config: any string field can now reference a managed secret with a URI scheme instead of plaintext or an env var. Google Secret Manager is the first provider: `token: gsm://notion-api-token` resolves to the secret's latest version, `gsm://my-secret/versions/3` pins a version, and a full `gsm://projects/<p>/secrets/<id>/versions/<n>` resource path is used as-is. References also work **inline** inside a larger string, e.g. `dsn: "postgres://u:${gsm://db-pw}@host/db"`, and multiple `${...}` tokens per value are allowed. A built-in `env://VAR` scheme exposes environment variables the same way (and, unlike the legacy whole-value `BIZON_ENV_` prefix, works inline too) — `BIZON_ENV_` keeps working unchanged. Resolution happens once over the raw config before validation, so **no connector changes are needed** — sources/destinations keep reading plain strings. GSM uses Application Default Credentials (workload identity / ambient creds); set provider defaults under an optional top-level `secrets:` block (e.g. `secrets.gsm.project_id`). New optional dependency: `pip install 'bizon[secretmanager]'`. Validate references before a run with `bizon secrets check <config>`, which dry-runs every reference and reports resolved/failed with masked output.
+
 ### Changed
 - BigQuery destinations (`bigquery`, `bigquery_streaming`, `bigquery_streaming_v2`) now prefix auto-generated table names with `_bizon_` so bizon-managed tables are clearly namespaced in shared datasets. This is **backwards compatible**: when no explicit `destination_id` is set, the destination first checks whether the legacy unprefixed table (`{source}_{stream}`) already exists — if it does, it keeps writing to it untouched, so existing pipelines are never disrupted; only brand-new tables get the `_bizon_` prefix. The lookup result is cached (one `get_table` call per run) and any non-`NotFound` error falls back to the legacy name. An explicit `destination_id` is never prefixed. Temp/staging tables (`_temp` / `_incremental`) inherit the resolved name. The prefix is configurable per destination via the new `table_prefix` field (default `_bizon_`; set to `""` to disable).
 

CLAUDE.md

@@ -312,6 +312,25 @@ See `bizon/connectors/sources/notion/src/source.py` for a complete incremental i
 - `process` - ProcessPoolExecutor (true parallelism)
 - `stream` - Synchronous single-thread
 
+### Secret & Reference Resolution
+
+Keep secrets out of YAML by referencing them with a URI scheme. Resolution runs once over
+the raw config dict before Pydantic validation (`bizon/engine/resolvers/`), so **connectors
+need no changes** — they always read plain strings.
+
+- `gsm://<id>` → Google Secret Manager, latest version (ADC auth). Pin with
+  `gsm://<id>/versions/<N>`, or pass a full `gsm://projects/<p>/secrets/<id>/versions/<N>` path.
+- `env://<VAR>` → environment variable (also works **inline**).
+- Inline form: embed in a larger string with `${...}`, e.g.
+  `dsn: "postgres://u:${gsm://db-pw}@host/db"` (multiple tokens allowed).
+- Optional `secrets:` block holds provider defaults (e.g. `secrets.gsm.project_id`).
+- Legacy whole-value `BIZON_ENV_FOO` references still work unchanged.
+- Install GSM support: `pip install 'bizon[secretmanager]'`.
+- Validate before running: `bizon secrets check <config>` (dry-runs every reference, masked output).
+
+Add a provider by dropping one adapter in `bizon/engine/resolvers/adapters/` and one entry in
+`_SCHEME_FACTORIES` (`bizon/engine/resolvers/resolver.py`).
+
 ### Key Patterns
 
 - **Factory Pattern**: `RunnerFactory`, `QueueFactory`, `BackendFactory`, `DestinationFactory`

bizon/cli/main.py

@@ -1,7 +1,12 @@
 import click
 from dotenv import find_dotenv, load_dotenv
 
-from bizon.engine.engine import RunnerFactory
+from bizon.engine.engine import RunnerFactory, replace_env_variables_in_config
+from bizon.engine.resolvers import (
+    ReferenceResolutionError,
+    ResolverRegistry,
+    collect_references_in_config,
+)
 from bizon.engine.runner.config import LoggerLevel
 from bizon.source.discover import discover_all_sources
 
@@ -73,6 +78,59 @@ def destination():
     pass
 
 
+# Create a 'secrets' group under 'bizon'
+@cli.group()
+def secrets():
+    """Subcommands for handling secret/reference resolution."""
+    pass
+
+
+@secrets.command()
+@click.argument("filename", type=click.Path(exists=True))
+@click.option(
+    "--env-file",
+    required=False,
+    type=click.Path(exists=True),
+    help="Path to .env file to load environment variables from.",
+)
+def check(filename: str, env_file: str):
+    """Dry-run all gsm:// / env:// references in a config and report (masked) results."""
+
+    # Load environment variables from .env file (same as `run`)
+    if env_file:
+        load_dotenv(env_file)
+    else:
+        load_dotenv(find_dotenv(".env"))
+
+    config = parse_from_yaml(filename)
+    # Resolve legacy BIZON_ENV_ whole-value references first, like the real run does
+    config = replace_env_variables_in_config(config=config)
+
+    references = collect_references_in_config(config)
+    if not references:
+        click.echo("No gsm:// / env:// references found in config.")
+        return
+
+    registry = ResolverRegistry(settings=config.get("secrets") or {})
+
+    path_width = max(len(path) for path, _ in references)
+    ref_width = max(len(reference) for _, reference in references)
+    failures = 0
+
+    for path, reference in references:
+        try:
+            value = registry.resolve_reference(reference)
+            status = click.style(f"✓  ({len(value)} chars)", fg="green")
+        except ReferenceResolutionError as error:
+            failures += 1
+            status = click.style(f"✗  {error}", fg="red")
+        click.echo(f"{path.ljust(path_width)}   {reference.ljust(ref_width)}   {status}")
+
+    if failures:
+        raise click.exceptions.ClickException(f"{failures} reference(s) failed to resolve.")
+    click.secho(f"All {len(references)} reference(s) resolved.", fg="green")
+
+
 @cli.command()
 @click.argument("filename", type=click.Path(exists=True))
 @click.option(

bizon/common/models.py

@@ -16,6 +16,7 @@
 from bizon.connectors.destinations.file.src.config import FileDestinationConfig
 from bizon.connectors.destinations.logger.src.config import LoggerConfig
 from bizon.engine.config import EngineConfig
+from bizon.engine.resolvers.config import SecretsConfig
 from bizon.monitoring.config import MonitoringConfig
 from bizon.source.config import SourceConfig, SourceSyncModes
 from bizon.transform.config import TransformModel
@@ -132,6 +133,11 @@ class BizonConfig(BaseModel):
         default=None,
     )
 
+    secrets: Optional[SecretsConfig] = Field(
+        default=None,
+        description="Provider defaults for gsm:// / env:// reference resolution",
+    )
+
     streams: Optional[list[StreamConfig]] = Field(
         None,
         description="Stream routing configuration (opt-in for multi-table streaming). "

bizon/engine/engine.py

@@ -4,6 +4,7 @@
 from bizon.common.models import BizonConfig
 
 from .config import RunnerTypes
+from .resolvers import resolve_references_in_config
 from .runner.runner import AbstractRunner
 
 
@@ -21,9 +22,12 @@ def replace_env_variables_in_config(config: dict) -> dict:
 class RunnerFactory:
     @staticmethod
     def create_from_config_dict(config: dict) -> AbstractRunner:
-        # Replace env variables in config
+        # Replace legacy BIZON_ENV_ whole-value env references (kept for backwards compat)
         config = replace_env_variables_in_config(config=config)
 
+        # Resolve gsm:// / env:// references (whole-value and inline ${...})
+        config = resolve_references_in_config(config=config)
+
         bizon_config = BizonConfig.model_validate(obj=config)
 
         if bizon_config.engine.runner.type == RunnerTypes.THREAD:

bizon/engine/resolvers/__init__.py

@@ -0,0 +1,18 @@
+from .config import GSMSettings, SecretsConfig
+from .resolver import (
+    AbstractReferenceResolver,
+    ReferenceResolutionError,
+    ResolverRegistry,
+    collect_references_in_config,
+    resolve_references_in_config,
+)
+
+__all__ = [
+    "AbstractReferenceResolver",
+    "GSMSettings",
+    "ReferenceResolutionError",
+    "ResolverRegistry",
+    "SecretsConfig",
+    "collect_references_in_config",
+    "resolve_references_in_config",
+]

bizon/engine/resolvers/adapters/env.py

@@ -0,0 +1,19 @@
+import os
+
+from ..resolver import AbstractReferenceResolver, ReferenceResolutionError
+
+
+class EnvResolver(AbstractReferenceResolver):
+    """Resolves ``env://VAR_NAME`` references to environment variables.
+
+    Unlike the legacy whole-value ``BIZON_ENV_`` prefix, ``env://`` also works inline,
+    e.g. ``dsn: "postgres://u:${env://PG_PASSWORD}@host/db"``.
+    """
+
+    scheme = "env"
+
+    def resolve(self, path: str) -> str:
+        var_name = path.strip()
+        if var_name not in os.environ:
+            raise ReferenceResolutionError(f"environment variable '{var_name}' is not set")
+        return os.environ[var_name]

bizon/engine/resolvers/adapters/gcp/gsm.py

@@ -0,0 +1,65 @@
+from typing import Optional
+
+from ...resolver import AbstractReferenceResolver, ReferenceResolutionError
+
+
+class GSMResolver(AbstractReferenceResolver):
+    """Resolves ``gsm://`` references against Google Secret Manager using ADC.
+
+    Path forms:
+    - ``<id>``                         -> projects/<project>/secrets/<id>/versions/latest
+    - ``<id>/versions/<N>``            -> projects/<project>/secrets/<id>/versions/<N>
+    - ``projects/.../versions/<N>``    -> used as-is (full resource name, pins everything)
+
+    Authentication uses Application Default Credentials (workload identity / ambient creds).
+    """
+
+    scheme = "gsm"
+
+    def __init__(self, project_id: Optional[str] = None):
+        try:
+            from google.cloud import secretmanager
+        except ImportError as error:
+            raise ReferenceResolutionError(
+                "gsm:// references require the Google Secret Manager client. "
+                "Install it with: pip install 'bizon[secretmanager]'."
+            ) from error
+
+        self._client = secretmanager.SecretManagerServiceClient()
+        self._project_id = project_id or self._default_project()
+
+    @staticmethod
+    def _default_project() -> Optional[str]:
+        try:
+            import google.auth
+
+            _, project = google.auth.default()
+            return project
+        except Exception:
+            return None
+
+    def _resource_name(self, path: str) -> str:
+        path = path.strip()
+        if path.startswith("projects/"):
+            return path
+
+        if "/versions/" in path:
+            secret_id, version = path.split("/versions/", 1)
+        else:
+            secret_id, version = path, "latest"
+
+        if not self._project_id:
+            raise ReferenceResolutionError(
+                f"cannot resolve gsm://{path}: no GCP project. Set 'secrets.gsm.project_id' "
+                "in the config, the GOOGLE_CLOUD_PROJECT env var, or use a full "
+                "'gsm://projects/<project>/secrets/<id>/versions/<n>' path."
+            )
+        return f"projects/{self._project_id}/secrets/{secret_id}/versions/{version}"
+
+    def resolve(self, path: str) -> str:
+        name = self._resource_name(path)
+        try:
+            response = self._client.access_secret_version(name=name)
+        except Exception as error:
+            raise ReferenceResolutionError(f"could not access '{name}': {error}") from error
+        return response.payload.data.decode("UTF-8")

bizon/engine/resolvers/config.py

@@ -0,0 +1,31 @@
+from typing import Optional
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class GSMSettings(BaseModel):
+    """Provider defaults for Google Secret Manager (gsm://) references."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    project_id: Optional[str] = Field(
+        default=None,
+        description="GCP project hosting the secrets. Optional: falls back to the "
+        "Application Default Credentials project / GOOGLE_CLOUD_PROJECT when omitted.",
+    )
+
+
+class SecretsConfig(BaseModel):
+    """Optional provider defaults for reference resolution.
+
+    The reference scheme (gsm://, env://, ...) identifies the provider, so this block
+    only carries optional per-provider settings and is itself optional.
+    """
+
+    model_config = ConfigDict(extra="forbid")
+
+    gsm: Optional[GSMSettings] = Field(
+        default=None,
+        description="Defaults for Google Secret Manager (gsm://) references.",
+    )
+    # awssm: Optional[AWSSMSettings] = None  # future