إضافات مصادر الأسرار

Secret sources resolve provider credentials from an external secret manager (a vault, a password manager, an OS keystore, a custom script) into environment variables at process startup — after ~/.hermes/.env loads, before Hermes reads credentials. Bitwarden, 1Password, and a generic command-helper source ship in-tree; every other backend is a plugin. This guide covers building one.

load_hermes_dotenv() often runs at import time before plugins register. Hermes then re-pulls secrets after plugin discovery when any enabled plugin secret source is configured. Enablement uses the source's is_enabled(cfg) contract; the standard form is secrets.<name>.enabled: true, while custom activation remains supported. That closes the "replace Bitwarden with my vault" first-process gap (#64177).

• Re-pull is idempotent and fail-open (never blocks startup).
• Sources only supply env vars through the orchestrator; there is no plugin API to dump other plugins' or the user's entire secret store beyond what your source's own config allows.
• Reading os.environ after load is possible for any in-process code — the trust boundary remains "enabled plugins run with agent privilege".

The orchestrator (agent.secret_sources.registry.apply_all) owns everything security- and precedence-sensitive, so a backend cannot get it wrong:

Implement agent.secret_sources.base.SecretSource. One method is required:

from pathlib import Path

from agent.secret_sources.base import (
    ErrorKind,
    FetchResult,
    SecretSource,
    run_secret_cli,
)


class MyVaultSource(SecretSource):
    name = "myvault"          # config section key: secrets.myvault
    label = "My Vault"        # used in startup lines + provenance labels
    shape = "mapped"          # "mapped" (explicit VAR→ref map) or "bulk" (project dump)
    scheme = "mv"             # optional: unique URI scheme you own (mv://...)

    def fetch(self, cfg: dict, home_path: Path) -> FetchResult:
        """Resolve secrets. MUST NOT raise. MUST NOT prompt."""
        result = FetchResult()
        token = os.environ.get("MYVAULT_TOKEN", "").strip()
        if not token:
            result.error = "secrets.myvault.enabled is true but MYVAULT_TOKEN is not set."
            result.error_kind = ErrorKind.NOT_CONFIGURED
            return result

        try:
            proc = run_secret_cli(
                ["myvault-cli", "export", "--json"],
                allow_env=["MYVAULT_TOKEN"],   # ONLY your auth vars — never full os.environ
                timeout=30,
            )
        except RuntimeError as exc:           # spawn failure / timeout
            result.error = str(exc)
            result.error_kind = ErrorKind.BINARY_MISSING
            return result

        if proc.returncode != 0:
            result.error = f"myvault-cli exited {proc.returncode}: {proc.stderr[:200]}"
            result.error_kind = ErrorKind.AUTH_FAILED
            return result

        result.secrets = parse_your_output(proc.stdout)  # {ENV_VAR: value}
        return result

    def protected_env_vars(self, cfg: dict):
        # Your bootstrap token — no source (including yours) may ever overwrite it.
        return frozenset({"MYVAULT_TOKEN"})

Contract rules (enforced, not suggestions)

• fetch() never raises. Errors go in result.error + result.error_kind. A raising fetch is contained by the orchestrator and reported as INTERNAL — a contract violation, not a feature.
• fetch() never prompts. Startup runs in non-TTY contexts (gateway, cron, Docker). run_secret_cli() closes stdin so a prompting helper fails fast. Interactive auth belongs in your CLI setup flow, never on the startup path.
• Sync, within budget. The orchestrator enforces a wall-clock timeout (default 120s, user-tunable via secrets.<name>.timeout_seconds). Exceeding it reports TIMEOUT and your result is discarded.
• You fetch; the orchestrator applies. Return the mapping you would contribute. Never write os.environ yourself — you'd bypass precedence, conflict detection, and provenance.
• API versioning. SecretSource.api_version defaults to the current SECRET_SOURCE_API_VERSION. The registry skips (with a warning) sources built against a different version instead of crashing startup.

Choosing your shape

• mapped — the user explicitly binds env-var names to references in config (like 1Password's env: map). Strongest intent: mapped claims beat bulk claims on contested vars.
• bulk — you inject a whole project/folder of secrets implicitly (like Bitwarden BSM). Yields to mapped sources.

Optional hooks

If your backend shells out to a CLI, use the shared helper instead of subprocess.run directly. It gives you the audited posture for free: argv-only (no shell=True), a minimal allowlisted child environment (by the time sources run, os.environ holds every credential Hermes knows — never hand that to a child process), NO_COLOR + ANSI-scrubbed stderr, stdin closed, timeout → clean RuntimeError. Pass user-supplied reference strings after a -- terminator in your argv so they can never parse as flags.

# __init__.py
def register(ctx):
    ctx.register_secret_source(MyVaultSource())

Registration is rejected (with a log warning, never a crash) for: non-SecretSource instances, invalid/duplicate names, a scheme another source owns, wrong api_version, or a shape outside mapped/bulk.

secrets:
  sources: [myvault, bitwarden]   # optional ordering
  myvault:
    enabled: true
    # ... your config_schema keys

Multi-source precedence, conflict warnings, and (from My Vault) provenance labels all work automatically — see the user-facing secrets docs ↗ for the precedence ladder.

Subclass the kit from the Hermes repo (tests/secret_sources/conformance.py) in your plugin's tests:

from tests.secret_sources.conformance import SecretSourceConformance

class TestMyVaultConformance(SecretSourceConformance):
    @pytest.fixture
    def source(self):
        return MyVaultSource()

It checks the rules that break other people when violated: never-raises on malformed config, machine-readable error kinds, disabled-by-default, positive timeouts, valid protected-var names, and a full apply_all() round trip. Green conformance is the review bar for calling a backend contract-compliant.