system_secret_store

Redis-backed store for Stargazer’s own operational API keys.

This module is the single source of truth for the credentials the bot uses to talk to third-party providers on its own behalf – the Gemini embedding key pool, the OpenRouter keys behind embeddings/anamnesis/search-query generation, and the media-tool keys. It exists so that no provider credential is ever written into tracked source.

It is deliberately distinct from tools/manage_api_keys.py:

tools/manage_api_keys.py stores user-supplied keys (per-user hashes, donated global/channel pools) and is driven by chat commands.
This module stores system keys – the operator-owned credentials that keep the service running when no user key is in play. Nothing here is writable from a chat tool.

Storage layout (Redis, db0)

stargazer:system_secrets: HASH mapping secret name -> value, for single-valued secrets.
stargazer:system_secrets:pool:<name>: LIST of values for multi-valued secrets (round-robin key pools). A LIST is used rather than a SET so pool order is stable across reloads, which keeps round-robin distribution deterministic.

Values are encrypted at rest with AES-256-GCM via api_key_encryption using the shared pool key derived from API_KEY_MASTER_KEY. When that master key is absent, values are stored and read as plaintext and a warning is emitted once – Redis replicas and RDB snapshots live on separate hosts, so running without the master key leaves the credentials readable there.

Resolution order

For every secret, in order, first hit wins:

The secret’s environment-variable override (see SECRET_SPECS).
The in-process TTL cache.
Redis.
The secret’s config.yaml fallback attribute, if it declares one.
None (or [] for pools).

There is intentionally no hardcoded fallback anywhere in this module. A missing secret surfaces as None so the caller can raise a clear configuration error rather than silently authenticating with a key baked into git history.

All network access is async-first. Synchronous mirrors exist only for the small number of legacy sync call sites and are documented as such.

system_secret_store.SECRETS_HASH = 'stargazer:system_secrets': Redis HASH holding every single-valued system secret.

system_secret_store.SECRET_POOL_PREFIX = 'stargazer:system_secrets:pool': Redis key prefix for multi-valued (pool) system secrets.

system_secret_store.pool_redis_key(name)

Build the Redis LIST key backing a multi-valued secret.

Pure string formatting with no I/O. Called by get_secret_pool(), set_secret_pool(), get_secret_pool_sync(), and the scripts/system_secrets_cli.py loader so every pool operation for a given secret name addresses the same LIST.

Parameters:: name (str) – The registered pool secret name, e.g. gemini.embed_pool.
Returns:: The fully qualified Redis LIST key.
Return type:: str

class system_secret_store.SecretSpec(name, description, env_var='', config_attr='', is_pool=False, provider='')

Bases: object

Declarative description of one system secret.

Variables:

name – Canonical secret name; the Redis HASH field or pool key suffix.
description – Human-readable purpose, surfaced by the loader CLI and by describe_secrets() so an operator knows what to provision.
env_var – Environment variable consulted before Redis. Kept so a deployment can inject a credential without touching the store.
config_attr – Attribute on the loaded config.Config used as a last-resort fallback, or "" when the secret has no config representation.
is_pool – True when the secret holds an ordered list of interchangeable keys rather than a single value.
provider – Provider slug used to group secrets in operator output.

Parameters:

name (str)
description (str)
env_var (str)
config_attr (str)
is_pool (bool)
provider (str)

name: str

description: str

env_var: str = ''

config_attr: str = ''

is_pool: bool = False

provider: str = ''

system_secret_store.SECRET_SPECS: dict[str, SecretSpec] = {'gemini.embed_pool': SecretSpec(name='gemini.embed_pool', description='Round-robin pool of free-tier Gemini API keys used for embedding rate-limit distribution (gemini_embed_pool.py).', env_var='GEMINI_EMBED_KEY_POOL', config_attr='', is_pool=True, provider='gemini'), 'gemini.paid': SecretSpec(name='gemini.paid', description='Paid-tier Gemini key: last-resort embedding fallback and the default key for the image/video/music generation tools.', env_var='GEMINI_EMBED_PAID_KEY', config_attr='gemini_api_key', is_pool=False, provider='gemini'), 'openrouter.anamnesis': SecretSpec(name='openrouter.anamnesis', description='Round-robin OpenRouter keys for anamnesis_engine.py memory extraction calls.', env_var='', config_attr='', is_pool=True, provider='openrouter'), 'openrouter.embed': SecretSpec(name='openrouter.embed', description='OpenRouter key for the /embeddings fallback tier in gemini_embed_pool.py.', env_var='OPENROUTER_API_KEY', config_attr='api_key', is_pool=False, provider='openrouter'), 'openrouter.search_query': SecretSpec(name='openrouter.search_query', description='OpenRouter key for search_query_generator.py web-search query generation.', env_var='', config_attr='', is_pool=False, provider='openrouter')}: Every system secret this codebase reads, keyed by canonical name.

system_secret_store.describe_secrets()

Return every registered secret spec, grouped by provider then name.

Pure in-memory sort over SECRET_SPECS with no I/O. Called by scripts/system_secrets_cli.py to render the provisioning checklist an operator works through when standing up or rotating the store.

Returns:: The specs in stable display order.
Return type:: list[SecretSpec]

exception system_secret_store.UnknownSecretError

Bases: KeyError

Raised when a caller references a secret name absent from the registry.

Guards against typos silently resolving to None and then to a confusing auth failure at the provider. Raised by _spec().

system_secret_store.CACHE_TTL_SECONDS = 300.0

Seconds a resolved secret stays cached in-process before Redis is re-read.

Kept short enough that a rotation propagates on its own within minutes, and long enough that hot paths (every embedding batch) do not add a Redis round-trip. Override with SYSTEM_SECRET_CACHE_TTL; 0 disables caching entirely.

system_secret_store.invalidate_cache(name=None)

Drop cached secret resolutions so the next read hits Redis.

Called by set_secret(), set_secret_pool(), and delete_secret() after a write, and available to operators (via the loader CLI) to force an immediate refresh after an out-of-band rotation.

Parameters:: name (str | None) – Secret to forget, or None to clear every entry.
Return type:: None
Returns:: None

system_secret_store.set_redis_client(client)

Inject the process-wide async Redis client used for secret reads.

Lets a service hand over the connection it already owns instead of this module opening a second one. Mirrors gemini_embed_pool.init_quota_tracking. Call during service startup, before the first secret read; when it is never called, _get_async_client() lazily builds a Sentinel-aware client from config.Config.

An injected client is never discarded by the loop-affinity check in _get_async_client(), because its lifecycle belongs to the caller.

Parameters:: client (Any) – An redis.asyncio client with decode_responses=True.
Return type:: None
Returns:: None

async system_secret_store.get_secret(name, *, redis_client=None)

Resolve a single-valued system secret.

Applies the module’s documented resolution order: environment override, in-process cache, Redis, config.yaml fallback, then None. There is no hardcoded default – callers must handle None by raising a configuration error rather than proceeding unauthenticated.

Called by gemini_embed_pool.get_openrouter_api_key / get_paid_fallback_key, search_query_generator.generate_search_queries, and the tools/ media handlers.

Parameters:

name (str) – A canonical name from SECRET_SPECS.
redis_client (Any | None) – Client to use instead of the module-level one.

Returns:

The credential, or None when it is not provisioned.

Return type:

str | None

Raises:

UnknownSecretError – If name is not registered.

async system_secret_store.get_secret_pool(name, *, redis_client=None)

Resolve a multi-valued system secret as an ordered key list.

Order is preserved because the backing structure is a Redis LIST, which keeps round-robin distribution across a key pool deterministic between reloads. Duplicates are removed while preserving first-seen order so a double-loaded key does not receive twice the traffic.

Called by gemini_embed_pool._get_key_pool and anamnesis_engine._next_key (via its pool refresh).

Parameters:

name (str) – A canonical pool name from SECRET_SPECS.
redis_client (Any | None) – Client to use instead of the module-level one.

Returns:

The credentials in pool order; empty when unprovisioned.

Return type:

list[str]

Raises:

UnknownSecretError – If name is not registered.
ValueError – If name is registered as a single-valued secret.

system_secret_store.get_secret_sync(name)

Blocking mirror of get_secret() for legacy synchronous call sites.

This performs a blocking Redis round-trip and must never be called from the event loop; async code should await get_secret() instead. Retained only for module-import-time and thread-pool paths that cannot await.

Parameters:: name (str) – A canonical name from SECRET_SPECS.
Returns:: The credential, or None when it is not provisioned.
Return type:: str | None

system_secret_store.get_secret_pool_sync(name)

Blocking mirror of get_secret_pool() for legacy sync call sites.

Performs a blocking Redis round-trip; see get_secret_sync() for the event-loop caveat.

Parameters:: name (str) – A canonical pool name from SECRET_SPECS.
Returns:: The credentials in pool order; empty when unprovisioned.
Return type:: list[str]
Raises:: ValueError – If name is registered as a single-valued secret.

async system_secret_store.set_secret(name, value, *, redis_client=None)

Store (or rotate) a single-valued system secret in Redis.

Encrypts the value when a master key is configured and invalidates the in-process cache so the new credential takes effect immediately in this process; other processes pick it up within CACHE_TTL_SECONDS.

Called by scripts/system_secrets_cli.py; not reachable from chat tools.

Parameters:

name (str) – A canonical name from SECRET_SPECS.
value (str) – The plaintext credential.
redis_client (Any | None) – Client to use instead of the module-level one.

Return type:

None

Returns:

None

Raises:

UnknownSecretError – If name is not registered.
ValueError – If name is a pool secret or value is blank.
RuntimeError – If Redis is unreachable, so a failed rotation is loud.

async system_secret_store.set_secret_pool(name, values, *, redis_client=None)

Replace a multi-valued system secret with a new ordered key list.

Writes the replacement LIST and deletes the old one in a single transaction so a concurrent reader never observes a half-populated pool.

Called by scripts/system_secrets_cli.py; not reachable from chat tools.

Parameters:

name (str) – A canonical pool name from SECRET_SPECS.
values (list[str]) – Plaintext credentials in the desired pool order.
redis_client (Any | None) – Client to use instead of the module-level one.

Return type:

None

Returns:

None

Raises:

UnknownSecretError – If name is not registered.
ValueError – If name is single-valued or values is empty.
RuntimeError – If Redis is unreachable.

async system_secret_store.delete_secret(name, *, redis_client=None)

Remove a system secret from Redis entirely.

Used to retire a credential that is being revoked rather than rotated, so a compromised key cannot be served from the store while the replacement is still being provisioned.

Parameters:

name (str) – A canonical name from SECRET_SPECS.
redis_client (Any | None) – Client to use instead of the module-level one.

Return type:

None

Returns:

None

Raises:

UnknownSecretError – If name is not registered.
RuntimeError – If Redis is unreachable.

system_secret_store.mask(value)

Redact a credential down to a recognisable, non-usable preview.

Shows at most the leading four and trailing four characters so an operator can tell two keys apart in status output without the full value reaching a log, a terminal scrollback, or a screenshot.

Parameters:: value (str) – The plaintext credential.
Returns:: The masked preview, or "(unset)" when value is empty.
Return type:: str

async system_secret_store.status(*, redis_client=None)

Report provisioning state for every registered secret, values masked.

Resolves each secret through the normal order and records where the value came from, which is what an operator needs to confirm a migration actually moved a credential into Redis rather than still reading a leftover env var or config.yaml entry.

Called by scripts/system_secrets_cli.py status.

Parameters:: redis_client (Any | None) – Client to use instead of the module-level one.
Return type:: list[dict[str, Any]]

A value that is present in Redis but fails to decrypt is reported as undecryptable, never as missing. Those two states call for opposite responses – provision the secret, versus fix the master key – and conflating them once produced a confident, wholly wrong “nothing is provisioned” report against a fully populated store.

Returns:: One row per secret with name, provider, is_pool, source, count, and preview fields.
Return type:: list[dict[str, Any]]
Parameters:: redis_client (Any | None)