Skip to content

Secrets

zeroth.platform.secrets

Secret provider, resolution, and redaction primitives.

SecretProviderConfigError

Bases: RuntimeError

Secret-backend configuration is incomplete or invalid.

EnvSecretProvider 

EnvSecretProvider(
    environment: Mapping[str, str] | None = None,
)

Resolve secret references by reading process or injected environment variables.

Tenant-scoped lookups follow the convention ZEROTH_SECRET__{TENANT}__{NAME} and fall back to the bare name, so a single-tenant deployment can key secrets off deployment.tenant_id while dev setups keep using plain env vars.

resolve 

resolve(
    secret_ref: str, *, tenant_id: str | None = None
) -> str | None

Resolve a concrete ref from the environment; the tenant-scoped key wins.

resolve_many 

resolve_many(
    refs: list[str], *, tenant_id: str | None = None
) -> dict[str, str]

Resolve several refs; refs that do not resolve are omitted from the result.

resolve_secret 

resolve_secret(
    logical_name: str,
    *,
    tenant_id: str | None = None,
    deployment_ref: str | None = None,
) -> str | None

Resolve a logical name; the tenant-scoped token wins over the bare env token.

resolve_async async 

resolve_async(
    secret_ref: str, *, tenant_id: str | None = None
) -> str | None

Async variant of :meth:resolve (pure dict read; no thread hop needed).

resolve_many_async async 

resolve_many_async(
    refs: list[str], *, tenant_id: str | None = None
) -> dict[str, str]

Async variant of :meth:resolve_many.

resolve_secret_async async 

resolve_secret_async(
    logical_name: str,
    *,
    tenant_id: str | None = None,
    deployment_ref: str | None = None,
) -> str | None

Async variant of :meth:resolve_secret.

SecretProvider

Bases: Protocol

Interface for resolving secret references to concrete values.

The optional tenant_id keyword scopes resolution to a single tenant (one value per deployment under the single-tenant model). It defaults to None so existing ref-based callers (http client, execution units) keep working unchanged.

resolve 

resolve(
    secret_ref: str, *, tenant_id: str | None = None
) -> str | None

Resolve a single secret reference.

resolve_many 

resolve_many(
    refs: list[str], *, tenant_id: str | None = None
) -> dict[str, str]

Resolve multiple secret references at once.

resolve_secret 

resolve_secret(
    logical_name: str,
    *,
    tenant_id: str | None = None,
    deployment_ref: str | None = None,
) -> str | None

Resolve a logical secret name (e.g. 'llm.openai') to a value.

Unlike :meth:resolve, which takes a concrete reference, this maps a stable logical key onto whatever concrete secret the backend holds for the given tenant. Returns None when unresolved; the caller decides whether to fail closed.

SecretResolutionError

Bases: RuntimeError

A required secret could not be resolved and no fallback is permitted.

Raised on a fail-closed path (e.g. allow_env_fallback=False) so a missing key surfaces loudly at call time instead of silently falling through to a process-global environment variable. Carries only the logical name — NEVER a concrete secret value.

SecretResolver 

SecretResolver(provider: SecretProvider)

Resolve environment-variable models into a concrete runtime environment.

resolve_environment_variables 

resolve_environment_variables(
    variables: list[EnvironmentVariable],
) -> dict[str, str]

Resolve variable models into an env dict; a missing secret ref raises KeyError.

resolve_environment_variables_async async 

resolve_environment_variables_async(
    variables: list[EnvironmentVariable],
) -> dict[str, str]

Async variant for event-loop callers (execution-unit dispatch).

known_secrets 

known_secrets() -> dict[str, str]

Snapshot every secret value resolved so far, keyed by secret ref.

redactor 

redactor() -> SecretRedactor

Build a :class:SecretRedactor seeded with every resolved secret value.

SecretRedactor 

SecretRedactor(
    known_secrets: Mapping[str, str] | None = None,
)

Replace known secret values with stable redaction markers.

Build from named mappings or a sequence of (reference, value) seeds.

String references are normalized to the conventional uppercase marker; tenant-qualified tuple references always use the opaque SECRET marker. Equal values are intentionally reduced to one deterministic marker.

redact 

redact(value: Any) -> Any

Recursively redact strings, dicts, and lists that contain known secrets.

TenantScopedVaultDriver 

TenantScopedVaultDriver(
    provider: VaultSecretProvider, *, tenant_id: str
)

Bind a shared Vault provider to one tenant before exposing persistence.

resource_definition property 

resource_definition: ResourceScopeDefinition

Return the immutable logical-resource contract.

operations property 

operations: MappingProxyType[
    ResourceOperation, ScopedOperation
]

Return immutable canonical operations bound to this tenant scope.

resolve async 

resolve(logical_name: str) -> str | None

Resolve one logical secret through the bound production provider.

resolve_many async 

resolve_many(logical_names: list[str]) -> dict[str, str]

Resolve several logical secrets within the bound tenant.

VaultSecretProvider 

VaultSecretProvider(
    *,
    addr: str,
    mount: str = "secret",
    token: str | None = None,
    role_id: str | None = None,
    secret_id: str | None = None,
    cache_ttl: float = 300.0,
    transport: BaseTransport | None = None,
    async_transport: AsyncBaseTransport | None = None,
    timeout: float = 10.0,
)

Resolve secrets from a Vault KV v2 mount with an in-memory TTL cache.

Parameters

addr: Vault base address, e.g. https://vault.internal:8200. mount: KV v2 mount point (default secret). token: A Vault token. Either this or an AppRole (role_id + secret_id) is required — :func:build_secret_provider enforces that at startup. role_id / secret_id: AppRole credentials; exchanged for a token lazily on first use. cache_ttl: Seconds a resolved value stays cached before a re-fetch. transport: Optional httpx transport (tests inject an httpx.MockTransport).

resolve 

resolve(
    secret_ref: str, *, tenant_id: str | None = None
) -> str | None

Resolve a concrete ref, treated as a logical name under the tenant path.

resolve_many 

resolve_many(
    refs: list[str], *, tenant_id: str | None = None
) -> dict[str, str]

Resolve several refs; refs that do not resolve are omitted from the result.

resolve_secret 

resolve_secret(
    logical_name: str,
    *,
    tenant_id: str | None = None,
    deployment_ref: str | None = None,
) -> str | None

Resolve a logical name via the TTL cache, doing one sync Vault GET on a miss.

resolve_async async 

resolve_async(
    secret_ref: str, *, tenant_id: str | None = None
) -> str | None

Async variant of :meth:resolve, served by the pooled client.

resolve_many_async async 

resolve_many_async(
    refs: list[str], *, tenant_id: str | None = None
) -> dict[str, str]

Async variant of :meth:resolve_many; refs that do not resolve are omitted.

resolve_secret_async async 

resolve_secret_async(
    logical_name: str,
    *,
    tenant_id: str | None = None,
    deployment_ref: str | None = None,
) -> str | None

Async :meth:resolve_secret: cache first, then one single-flight fetch per key.

aclose async 

aclose() -> None

Close the pooled async client. Safe to call more than once.

warm async 

warm(entries: list[tuple[str, str]] | None = None) -> None

Best-effort async prefetch of (tenant_id, logical_name) pairs.

Failures here are swallowed (logged, redacted) — resolution still works lazily via :meth:resolve_secret. entries defaults to nothing, because Vault KV does not cheaply enumerate arbitrary paths; callers that know their key set can pass it to pre-populate the cache.

build_secret_provider 

build_secret_provider(
    settings: SecretsSettings,
) -> SecretProvider

Return the single secret provider selected by settings.backend.

  • env (dev default) — reads process environment variables, with the ZEROTH_SECRET__{TENANT}__{NAME} tenant-scoping convention.
  • vault — reads a HashiCorp Vault KV-v2 mount. Requires vault_addr plus either a vault_token or a full AppRole pair (vault_role_id + vault_secret_id); otherwise raises.

normalize_secret_name 

normalize_secret_name(name: str) -> str

Map a logical secret name to its ENV-VAR token form.

'llm.openai' -> 'LLM_OPENAI'; 'signing.deployment' -> 'SIGNING_DEPLOYMENT'. Dots and dashes become underscores and the whole string is upper-cased so it composes into the ZEROTH_SECRET__{TENANT}__{NAME} convention.

resolve_async async 

resolve_async(
    provider: SecretProvider,
    secret_ref: str,
    *,
    tenant_id: str | None = None,
) -> str | None

Resolve one secret ref without blocking the event loop.

Awaits the provider's native resolve_async when it has one; sync-only providers (which may perform blocking I/O, e.g. Vault over HTTP) run via asyncio.to_thread.

resolve_many_async async 

resolve_many_async(
    provider: SecretProvider,
    refs: list[str],
    *,
    tenant_id: str | None = None,
) -> dict[str, str]

Resolve several secret refs without blocking the event loop.

resolve_secret_async async 

resolve_secret_async(
    provider: SecretProvider,
    logical_name: str,
    *,
    tenant_id: str | None = None,
    deployment_ref: str | None = None,
) -> str | None

Resolve a logical secret name without blocking the event loop.