feat(secrets): add 1Password (op://) secret source

Resolve provider credentials from 1Password op://vault/item/field references
at startup via the official `op` CLI, alongside the existing Bitwarden source.
Users map env-var names to references in secrets.onepassword.env; after .env
loads, each is resolved with `op read` and injected into os.environ. Auth is
whatever `op` already uses (service-account token or desktop/interactive
session) — Hermes never authenticates or installs `op` itself.

Startup-safe and fail-open: a missing binary, expired auth, a bad reference,
or an empty value each warn and fall back to existing credentials, never
blocking startup. Successful, complete pulls are cached in-process and on disk
(<hermes_home>/cache/op_cache.json, 0600) via the shared DiskCache; only
secret values are stored, never the token (auth is fingerprinted into the
key). Adds `hermes secrets onepassword {setup,status,set,remove,sync,disable}`
(aliases op/1password), config defaults, the cli-config example, docs, and
hermetic tests.

Hardening applied across both backends in env_loader: each source runs in its
own guard, config sections are coerced to dict, and cache_ttl_seconds is
coerced defensively — so a malformed secrets: section can't abort startup.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Taylor H. Perkins 2026-06-01 09:28:53 -07:00 committed by Teknium
parent db495b0fba
commit 5c4c0e9d9b
10 changed files with 1728 additions and 21 deletions

View file

@ -9,21 +9,25 @@ runs the enabled sources (ordering, mapped-beats-bulk precedence,
first-claim-wins conflicts, ``override_existing`` semantics, provenance)
is :func:`agent.secret_sources.registry.apply_all`. Multiple sources
can be enabled at once see the registry module docstring for the
precedence ladder.
precedence ladder. The atomic-write / 0600 / TTL disk-cache substrate
is shared across backends in ``agent.secret_sources._cache`` so the
security-sensitive bits live in exactly one place.
Currently bundled:
- ``bitwarden`` Bitwarden Secrets Manager (`bws` CLI). See
``agent.secret_sources.bitwarden`` for the integration and
``hermes_cli.secrets_cli`` for the user-facing setup wizard.
- ``onepassword`` 1Password ``op://`` secret references (`op` CLI).
See ``agent.secret_sources.onepassword`` for the integration and
``hermes_cli.onepassword_secrets_cli`` for the user-facing commands.
The bundled set is deliberately closed (policy mirrors memory
providers): new third-party secret managers ship as standalone plugin
repos that subclass ``SecretSource`` and register through
``PluginContext.register_secret_source()`` they are NOT added to this
package. Exceptions (planned): 1Password, and possibly a generic
``command`` source; OS keystores (Keychain/DPAPI/libsecret) are under
discussion.
package. A generic ``command`` source is a possible future exception;
OS keystores (Keychain/DPAPI/libsecret) are under discussion.
"""
from agent.secret_sources.base import ( # noqa: F401

View file

@ -0,0 +1,492 @@
"""1Password (`op` CLI) secret source.
Resolve provider credentials from 1Password ``op://vault/item/field``
references at process startup so they don't have to live in plaintext in
``~/.hermes/.env``.
Design summary
--------------
* Users map environment-variable names to official 1Password secret
references in ``secrets.onepassword.env``::
secrets:
onepassword:
enabled: true
env:
OPENAI_API_KEY: "op://Private/OpenAI/api key"
ANTHROPIC_API_KEY: "op://Private/Anthropic/credential"
* After ``.env`` loads, each reference is resolved with a single
``op read -- <reference>`` call and injected into ``os.environ`` (the
same point in startup as the Bitwarden source).
* Authentication is whatever the user's ``op`` CLI already uses — a
service-account token (``OP_SERVICE_ACCOUNT_TOKEN``) for headless boxes,
or a desktop/interactive session (``OP_SESSION_*``). Hermes never
authenticates on the user's behalf; it shells out to an already-trusted,
already-authenticated CLI.
* Failures NEVER block startup. A missing ``op`` binary, expired auth, a
bad reference, or a permission error each surface a one-line warning and
Hermes continues with whatever credentials ``.env`` already had.
The atomic-write / ``0600`` / TTL cache mechanics are shared with the other
backends via :mod:`agent.secret_sources._cache` successful, complete pulls
are cached in-process and on disk under ``<hermes_home>/cache/op_cache.json``
so back-to-back short-lived ``hermes`` invocations don't re-shell ``op`` for
every reference. The disk file holds only resolved secret *values*; auth
material is fingerprinted, never stored.
"""
from __future__ import annotations
import hashlib
import logging
import os
import re
import shutil
import subprocess
import time
from pathlib import Path
from typing import Dict, List, Optional, Tuple
from agent.secret_sources._cache import (
CachedFetch,
DiskCache,
FetchResult,
is_valid_env_name,
)
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Configuration constants
# ---------------------------------------------------------------------------
# How long to wait for a single `op read`, in seconds.
_OP_RUN_TIMEOUT = 30
# Default env var the official `op` CLI reads for service-account auth. Users
# can point `service_account_token_env` at a different name; we always export
# the value to the child as OP_SERVICE_ACCOUNT_TOKEN, which is what `op` itself
# looks for.
_DEFAULT_TOKEN_ENV = "OP_SERVICE_ACCOUNT_TOKEN"
# Strip whole ANSI CSI sequences (colour, cursor moves, line erases) from any
# `op` diagnostic we surface — not just the lone ESC byte — so a control
# sequence can't reposition the cursor or hide text after a redaction marker.
_ANSI_CSI_RE = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]")
# Env vars the `op` child actually needs. We build a minimal allowlisted env
# rather than copying all of os.environ (which, post-dotenv, holds every
# provider credential) into the child — tighter blast radius if `op` or
# anything it execs ever misbehaves. OP_SESSION_* and the token are added
# dynamically in _op_child_env().
_OP_ENV_ALLOWLIST = (
"PATH",
"HOME",
"USERPROFILE",
"APPDATA",
"LOCALAPPDATA",
"SystemRoot",
"TMPDIR",
"TMP",
"TEMP",
"XDG_CONFIG_HOME",
"XDG_RUNTIME_DIR",
"OP_ACCOUNT",
"OP_CONNECT_HOST",
"OP_CONNECT_TOKEN",
)
# ---------------------------------------------------------------------------
# Cache
# ---------------------------------------------------------------------------
# In-process cache. The key folds in str(home_path) so a HERMES_HOME switch
# inside one long-lived process (e.g. the gateway) can't return another
# profile's secrets from L1. The disk layer omits home from its serialized
# key because the file already lives under the home dir (see _disk_key_str).
_CacheKey = Tuple[str, str, str, str] # (auth_fp, account, home, refs_fp)
_CACHE: Dict[_CacheKey, CachedFetch] = {}
_DISK_CACHE_BASENAME = "op_cache.json"
def _disk_key_str(cache_key: _CacheKey) -> str:
"""Serialize a cache key for on-disk storage, omitting home_path.
The disk file is already partitioned by home (it lives under
``<home>/cache/``), so the path provides the home dimension; folding it
into the key string too would be redundant.
"""
auth_fp, account, _home, refs_fp = cache_key
return f"{auth_fp}|{account}|{refs_fp}"
_DISK_CACHE: DiskCache = DiskCache(
_DISK_CACHE_BASENAME, key_serializer=_disk_key_str
)
def _disk_cache_path(home_path: Optional[Path] = None) -> Path:
"""Path to the on-disk cache (exposed for tests and direct callers)."""
return _DISK_CACHE.path(home_path)
# ---------------------------------------------------------------------------
# Reference validation + fingerprinting
# ---------------------------------------------------------------------------
def _validate_references(
references: Optional[Dict[str, str]],
) -> Tuple[Dict[str, str], List[str]]:
"""Return ``(valid_refs, warnings)`` from an ``env`` mapping.
A reference is kept only if its target env-var name is a valid POSIX
name and the value is a stripped ``op://`` reference string. Everything
else produces a warning and is dropped (never fatal).
"""
valid: Dict[str, str] = {}
warnings: List[str] = []
for name, ref in (references or {}).items():
if not is_valid_env_name(name):
warnings.append(f"Skipping {name!r}: not a valid env-var name")
continue
if not isinstance(ref, str):
warnings.append(f"Skipping {name!r}: reference is not a string")
continue
cleaned = ref.strip()
if not cleaned.startswith("op://"):
warnings.append(
f"Skipping {name!r}: {ref!r} is not an op:// secret reference"
)
continue
valid[name] = cleaned
return valid, warnings
def _auth_fingerprint(token_env: str) -> str:
"""SHA-256 prefix over the auth material `op` would use.
Folds in the service-account token, ``OP_ACCOUNT``, and *all*
``OP_SESSION_*`` vars (the names `op` actually exports for interactive
sessions ``OP_SESSION_<account_shorthand>``). Signing out and into a
different identity therefore changes the cache key, so a value cached under
a previous identity is never served under a new one. Never logged or
displayed; the raw token never leaves this hash.
"""
parts: List[str] = [
f"token={os.environ.get(token_env, '')}",
f"account={os.environ.get('OP_ACCOUNT', '')}",
]
for key in sorted(os.environ):
if key.startswith("OP_SESSION_"):
parts.append(f"{key}={os.environ[key]}")
material = "\n".join(parts)
return hashlib.sha256(material.encode("utf-8")).hexdigest()[:16]
def _refs_fingerprint(references: Dict[str, str]) -> str:
"""SHA-256 prefix over the configured name→reference mapping."""
material = "\n".join(f"{name}={references[name]}" for name in sorted(references))
return hashlib.sha256(material.encode("utf-8")).hexdigest()[:16]
# ---------------------------------------------------------------------------
# Binary discovery
# ---------------------------------------------------------------------------
def find_op(binary_path: str = "") -> Optional[Path]:
"""Resolve a usable ``op`` binary, or None.
When ``binary_path`` is set it is used verbatim and PATH is NOT consulted
pinning an absolute path is a way to avoid trusting whatever ``op`` shows
up first on ``PATH``. A pinned-but-missing path returns None (the caller
surfaces a clear error) rather than silently falling back.
"""
if binary_path:
pinned = Path(binary_path)
if pinned.exists() and os.access(pinned, os.X_OK):
return pinned
return None
found = shutil.which("op")
return Path(found) if found else None
# ---------------------------------------------------------------------------
# `op read` invocation
# ---------------------------------------------------------------------------
def _scrub(text: str) -> str:
"""Remove ANSI control sequences and trim, for safe message surfacing."""
return _ANSI_CSI_RE.sub("", text).replace("\x1b", "").strip()
def _op_child_env(token_value: str) -> Dict[str, str]:
"""Build a minimal allowlisted environment for the ``op`` child process."""
env: Dict[str, str] = {}
for key in _OP_ENV_ALLOWLIST:
val = os.environ.get(key)
if val is not None:
env[key] = val
# Desktop / interactive session credentials.
for key, val in os.environ.items():
if key.startswith("OP_SESSION_"):
env[key] = val
# `op` reads OP_SERVICE_ACCOUNT_TOKEN regardless of which env var the user
# configured Hermes to source it from, so normalize to that name here.
if token_value:
env["OP_SERVICE_ACCOUNT_TOKEN"] = token_value
env["NO_COLOR"] = "1"
return env
def _run_op_read(
op: Path,
reference: str,
*,
account: str = "",
token_value: str = "",
) -> str:
"""Resolve a single ``op://`` reference to its value.
Raises :class:`RuntimeError` on any failure including a ``returncode 0``
with empty output, which would otherwise silently clobber a good
``.env``/shell credential with ``""``.
"""
cmd: List[str] = [str(op), "read"]
if account:
cmd += ["--account", account]
# `--` terminates option parsing so a reference can never be mis-parsed as
# an `op` flag even if validation is ever loosened.
cmd += ["--", reference]
try:
proc = subprocess.run( # noqa: S603 — op path is user-trusted, argv list
cmd,
env=_op_child_env(token_value),
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
timeout=_OP_RUN_TIMEOUT,
)
except subprocess.TimeoutExpired as exc:
raise RuntimeError(
f"op read timed out after {_OP_RUN_TIMEOUT}s for {reference!r}"
) from exc
except OSError as exc:
raise RuntimeError(f"failed to invoke op: {exc}") from exc
if proc.returncode != 0:
err = _scrub(proc.stderr or "")[:200]
if err:
raise RuntimeError(f"op read failed for {reference!r}: {err}")
raise RuntimeError(
f"op read exited {proc.returncode} for {reference!r}"
)
# `op` appends a trailing newline; strip only that so a value with
# intentional internal/edge spaces survives. But a value that is empty or
# whitespace-only is treated as empty: applying it would silently clobber a
# good .env/shell credential with effectively nothing.
value = (proc.stdout or "").rstrip("\r\n")
if not value.strip():
raise RuntimeError(f"op read returned an empty value for {reference!r}")
return value
# ---------------------------------------------------------------------------
# Fetch
# ---------------------------------------------------------------------------
def fetch_onepassword_secrets(
*,
references: Dict[str, str],
account: str = "",
token_env: str = _DEFAULT_TOKEN_ENV,
binary: Optional[Path] = None,
binary_path: str = "",
use_cache: bool = True,
cache_ttl_seconds: float = 300,
home_path: Optional[Path] = None,
) -> Tuple[Dict[str, str], List[str]]:
"""Resolve ``references`` (name → ``op://…``) to ``(secrets, warnings)``.
Raises :class:`RuntimeError` only when no ``op`` binary is available a
fatal "can't fetch anything" condition. Per-reference failures (expired
auth, bad reference, empty value) are collected as warnings and the
reference is dropped, so one bad entry never sinks the rest.
Only a complete, error-free pull is cached, so a transient auth failure
isn't frozen in for the whole TTL window.
"""
valid, warnings = _validate_references(references)
if not valid:
return {}, warnings
token_value = os.environ.get(token_env, "").strip()
cache_key: _CacheKey = (
_auth_fingerprint(token_env),
account or "",
str(home_path) if home_path is not None else "",
_refs_fingerprint(valid),
)
if use_cache:
cached = _CACHE.get(cache_key)
if cached and cached.is_fresh(cache_ttl_seconds):
return dict(cached.secrets), warnings
disk_cached = _DISK_CACHE.read(cache_key, cache_ttl_seconds, home_path)
if disk_cached is not None:
# Promote into L1 so later fetches in this process skip the disk read.
_CACHE[cache_key] = disk_cached
return dict(disk_cached.secrets), warnings
op = binary or find_op(binary_path)
if op is None:
raise RuntimeError(
"op CLI not found. Install the 1Password CLI "
"(https://developer.1password.com/docs/cli/get-started/) or set "
"secrets.onepassword.binary_path to its absolute location."
)
secrets: Dict[str, str] = {}
read_errors = 0
for name in sorted(valid):
try:
secrets[name] = _run_op_read(
op, valid[name], account=account, token_value=token_value
)
except RuntimeError as exc:
warnings.append(str(exc))
read_errors += 1
if use_cache and not read_errors and secrets:
entry = CachedFetch(secrets=dict(secrets), fetched_at=time.time())
_CACHE[cache_key] = entry
_DISK_CACHE.write(cache_key, entry, cache_ttl_seconds, home_path)
return secrets, warnings
# ---------------------------------------------------------------------------
# Public entry point — called from hermes_cli.env_loader
# ---------------------------------------------------------------------------
def apply_onepassword_secrets(
*,
enabled: bool,
env: Optional[Dict[str, str]] = None,
account: str = "",
service_account_token_env: str = _DEFAULT_TOKEN_ENV,
binary_path: str = "",
override_existing: bool = True,
cache_ttl_seconds: float = 300,
home_path: Optional[Path] = None,
) -> FetchResult:
"""Resolve configured ``op://`` references and set them on ``os.environ``.
Called by ``load_hermes_dotenv()`` after the .env files have loaded.
Intentionally defensive any failure returns a :class:`FetchResult` with
``error`` set (or surfaces warnings); it never raises.
Parameters mirror the ``secrets.onepassword.*`` config keys so the caller
can splat the dict in. References that are already satisfied by the
current environment (when ``override_existing`` is false) are skipped
*before* fetching, so ``op`` is never invoked for a value that would be
discarded.
"""
result = FetchResult()
if not enabled:
return result
valid, warnings = _validate_references(env)
result.warnings.extend(warnings)
# Skip-before-fetch: never resolve a reference we'd only throw away.
refs_to_fetch: Dict[str, str] = {}
for name, ref in valid.items():
if name == service_account_token_env:
# Never let a resolved secret clobber the very token used to auth.
result.skipped.append(name)
continue
if not override_existing and os.environ.get(name):
result.skipped.append(name)
continue
refs_to_fetch[name] = ref
if not refs_to_fetch:
return result
binary = find_op(binary_path)
result.binary_path = binary
if binary is None:
if binary_path:
result.error = (
f"secrets.onepassword.binary_path ({binary_path!r}) is not an "
"executable op binary."
)
else:
result.error = (
"secrets.onepassword.enabled is true but the op CLI was not "
"found on PATH. Install it "
"(https://developer.1password.com/docs/cli/get-started/) or set "
"secrets.onepassword.binary_path."
)
return result
try:
secrets, fetch_warnings = fetch_onepassword_secrets(
references=refs_to_fetch,
account=account,
token_env=service_account_token_env,
binary=binary,
cache_ttl_seconds=cache_ttl_seconds,
home_path=home_path,
)
except RuntimeError as exc:
result.error = str(exc)
return result
result.secrets = secrets
result.warnings.extend(fetch_warnings)
for name, value in secrets.items():
# The token-var and override guards already filtered refs_to_fetch, but
# re-check defensively in case the fetch layer ever returns extras.
if name == service_account_token_env:
if name not in result.skipped:
result.skipped.append(name)
continue
if not override_existing and os.environ.get(name):
if name not in result.skipped:
result.skipped.append(name)
continue
os.environ[name] = value
result.applied.append(name)
return result
# ---------------------------------------------------------------------------
# Test hook — used by hermetic tests to flush the cache between cases.
# ---------------------------------------------------------------------------
def _reset_cache_for_tests(home_path: Optional[Path] = None) -> None:
"""Clear in-process AND disk caches.
Tests can pass ``home_path`` to scope the disk cleanup to a tmpdir.
Without it we fall back to the same default resolution as the writer.
"""
_CACHE.clear()
_DISK_CACHE.clear(home_path)

View file

@ -1386,10 +1386,14 @@ updates:
# =============================================================================
# Pull provider credentials from external secret managers at process startup
# instead of storing them in ~/.hermes/.env. Only the manager's bootstrap
# token (e.g. BWS_ACCESS_TOKEN) lives in .env; everything else rotates
# centrally in the vault. Multiple sources can be enabled at once:
# - "mapped" sources (explicit VAR -> ref bindings) beat "bulk" sources
# (whole-project dumps like Bitwarden BSM)
# credential (e.g. BWS_ACCESS_TOKEN / OP_SERVICE_ACCOUNT_TOKEN) lives in .env
# (or your shell / desktop session); everything else rotates centrally.
# Failures never block startup — Hermes warns once and continues with
# whatever .env already had.
#
# Multiple sources can be enabled at once:
# - "mapped" sources (explicit VAR -> ref bindings, e.g. 1Password's env:
# map) beat "bulk" sources (whole-project dumps like Bitwarden BSM)
# - within a shape, the first source to claim a var wins; later claims
# are skipped with a startup warning (never a silent clobber)
# - a source's override_existing lets it beat .env/shell values, but
@ -1398,12 +1402,28 @@ updates:
#
# secrets:
# # Optional explicit ordering of enabled sources.
# # sources: [bitwarden]
# # sources: [onepassword, bitwarden]
#
# # ---- Bitwarden Secrets Manager (bws CLI) --------------------------------
# bitwarden:
# enabled: false
# project_id: "" # BSM project UUID
# access_token_env: BWS_ACCESS_TOKEN
# cache_ttl_seconds: 300 # 0 disables memory+disk caching
# override_existing: true # BSM wins over .env so rotation works
# auto_install: true # auto-download the pinned bws binary
# server_url: "" # e.g. https://vault.bitwarden.eu (EU cloud)
# access_token_env: BWS_ACCESS_TOKEN # bootstrap token, sourced from .env
# project_id: "" # UUID of the BSM project to sync
# server_url: "" # "" = US Cloud; EU/self-hosted URL otherwise
# cache_ttl_seconds: 300 # 0 disables caching
# override_existing: true # BSM values win over existing env
# auto_install: true # lazy-download bws into ~/.hermes/bin
#
# # ---- 1Password (op CLI) -------------------------------------------------
# onepassword:
# enabled: false
# # Map env-var names to op:// secret references. Each is resolved with a
# # single `op read` at startup.
# env:
# OPENAI_API_KEY: "op://Private/OpenAI/api key"
# ANTHROPIC_API_KEY: "op://Private/Anthropic/credential"
# account: "" # op --account shorthand; "" = default
# service_account_token_env: OP_SERVICE_ACCOUNT_TOKEN # headless auth; unset = desktop session
# binary_path: "" # "" = resolve op via PATH; else absolute path
# cache_ttl_seconds: 300 # 0 disables BOTH cache layers
# override_existing: true # resolved values win over existing env

View file

@ -3113,6 +3113,34 @@ DEFAULT_CONFIG = {
# `hermes secrets bitwarden setup`.
"server_url": "",
},
"onepassword": {
# Master switch. When false, the op CLI is never invoked —
# same as not having this section at all.
"enabled": False,
# Mapping of env-var name → 1Password secret reference
# (op://vault/item/field). Each entry is resolved with a
# single `op read` at startup.
"env": {},
# Optional account shorthand / sign-in address passed as
# `op read --account <account>`. Empty = op's default account.
"account": "",
# Name of the env var holding a 1Password service-account token
# for headless auth. Sourced from ~/.hermes/.env (or the shell)
# and exported to the op child as OP_SERVICE_ACCOUNT_TOKEN.
# Leave the var unset to use an interactive/desktop op session.
"service_account_token_env": "OP_SERVICE_ACCOUNT_TOKEN",
# Optional absolute path to the op binary. When set it is used
# verbatim (PATH is not consulted) — pin this to avoid trusting
# whatever `op` appears first on PATH. Empty = resolve via PATH.
"binary_path": "",
# Seconds to cache resolved values in-process and on disk. 0
# disables BOTH cache layers (no values are written to disk).
"cache_ttl_seconds": 300,
# When True (default), resolved values overwrite existing env
# vars so rotating a secret in 1Password takes effect on next
# start. Flip to false to let .env / shell exports win locally.
"override_existing": True,
},
},
# Paste collapse thresholds (TUI + CLI).

View file

@ -12787,16 +12787,16 @@ def main():
fallback_parser.set_defaults(func=cmd_fallback)
# =========================================================================
# secrets command — external secret managers (currently: Bitwarden)
# secrets command — external secret managers (Bitwarden, 1Password)
# =========================================================================
secrets_parser = subparsers.add_parser(
"secrets",
help="Manage external secret sources (Bitwarden Secrets Manager)",
help="Manage external secret sources (Bitwarden, 1Password)",
description=(
"Pull API keys from an external secret manager at process startup "
"instead of storing them in ~/.hermes/.env. Currently supports "
"Bitwarden Secrets Manager. See: "
"https://hermes-agent.nousresearch.com/docs/user-guide/secrets/bitwarden"
"instead of storing them in ~/.hermes/.env. Supports Bitwarden "
"Secrets Manager and 1Password. See: "
"https://hermes-agent.nousresearch.com/docs/user-guide/secrets/"
),
)
secrets_subparsers = secrets_parser.add_subparsers(dest="secrets_command")
@ -12807,16 +12807,27 @@ def main():
help="Bitwarden Secrets Manager integration",
)
secrets_op = secrets_subparsers.add_parser(
"onepassword",
aliases=["op", "1password"],
help="1Password (op:// references) integration",
)
# Lazy import — only pays for itself when this subcommand is actually used.
from hermes_cli import secrets_cli as _secrets_cli
from hermes_cli import onepassword_secrets_cli as _op_secrets_cli
_secrets_cli.register_cli(secrets_bw)
_op_secrets_cli.register_cli(secrets_op)
def _dispatch_secrets(args): # noqa: ANN001
sub = getattr(args, "secrets_command", None)
bw_sub = getattr(args, "secrets_bw_command", None)
op_sub = getattr(args, "secrets_op_command", None)
if sub in ("bitwarden", "bw") and bw_sub is not None:
return args.func(args)
if sub in ("onepassword", "op", "1password") and op_sub is not None:
return args.func(args)
secrets_parser.print_help()
return 0

View file

@ -0,0 +1,433 @@
"""CLI handlers for ``hermes secrets onepassword ...``.
Subcommands:
setup verify the op CLI, set account / token env var, enable
status show config + op binary + auth + configured references
set map an env var to an ``op://`` reference
remove drop a mapping
sync resolve references now and show what would be applied (dry-run)
disable flip ``secrets.onepassword.enabled`` to False
Unlike Bitwarden, the ``op`` binary is NOT auto-installed: 1Password publishes
the CLI through OS package managers and signed installers, so Hermes expects
an already-installed, already-authenticated ``op`` and never downloads one.
"""
from __future__ import annotations
import argparse
import os
import subprocess
from pathlib import Path
from typing import Optional
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from agent.secret_sources import onepassword as op_src
from hermes_cli.config import (
get_env_path,
load_config,
save_config,
save_env_value,
)
from hermes_cli.secret_prompt import masked_secret_prompt
_DEFAULT_TOKEN_ENV = "OP_SERVICE_ACCOUNT_TOKEN"
_DOCS_URL = "https://developer.1password.com/docs/cli/get-started/"
# ---------------------------------------------------------------------------
# Argparse wiring — called from hermes_cli.main
# ---------------------------------------------------------------------------
def register_cli(parent_parser: argparse.ArgumentParser) -> None:
"""Attach the ``onepassword`` subcommand tree to a parent parser."""
sub = parent_parser.add_subparsers(dest="secrets_op_command")
setup = sub.add_parser(
"setup",
help="Verify the op CLI, set account / token env var, and enable",
)
setup.add_argument(
"--account",
help="1Password account shorthand or sign-in address (op --account)",
)
setup.add_argument(
"--token-env",
help=f"Env var holding a service-account token (default {_DEFAULT_TOKEN_ENV})",
)
setup.add_argument(
"--token",
help="Service-account token to store in .env non-interactively",
)
setup.add_argument(
"--binary-path",
help="Absolute path to the op binary (skips PATH lookup)",
)
setup.set_defaults(func=cmd_setup)
status = sub.add_parser("status", help="Show config + op binary + references")
status.set_defaults(func=cmd_status)
set_p = sub.add_parser("set", help="Map an env var to an op:// reference")
set_p.add_argument("env_var", help="Environment variable name, e.g. OPENAI_API_KEY")
set_p.add_argument("reference", help="1Password reference, e.g. op://Private/OpenAI/api key")
set_p.set_defaults(func=cmd_set)
remove = sub.add_parser("remove", help="Remove an env-var → reference mapping")
remove.add_argument("env_var", help="Environment variable name to unmap")
remove.set_defaults(func=cmd_remove)
sync = sub.add_parser("sync", help="Resolve references now and report what changed")
sync.add_argument(
"--apply",
action="store_true",
help="Actually export resolved values into the current shell (default: dry-run)",
)
sync.set_defaults(func=cmd_sync)
disable = sub.add_parser("disable", help="Turn off the 1Password integration")
disable.set_defaults(func=cmd_disable)
# ---------------------------------------------------------------------------
# Handlers
# ---------------------------------------------------------------------------
def cmd_setup(args: argparse.Namespace) -> int:
console = Console()
console.print(
Panel.fit(
"[bold]1Password secret source setup[/bold]\n\n"
"Hermes resolves [cyan]op://vault/item/field[/cyan] references through your\n"
"already-installed, already-authenticated 1Password CLI (`op`).\n\n"
f"Don't have it yet? Install + sign in: [cyan]{_DOCS_URL}[/cyan]",
border_style="cyan",
)
)
cfg = load_config()
op_cfg = cfg.setdefault("secrets", {}).setdefault("onepassword", {})
# ------------------------------------------------------------------ binary
console.print()
console.print("[bold]Step 1[/bold] Locate the op CLI")
binary_path = (args.binary_path or op_cfg.get("binary_path", "") or "").strip()
binary = op_src.find_op(binary_path)
if binary is None:
if binary_path:
console.print(f" [red]✗ {binary_path} is not an executable op binary.[/red]")
else:
console.print(" [red]✗ op not found on PATH.[/red]")
console.print(f" Install the 1Password CLI: {_DOCS_URL}")
return 1
console.print(f" [green]✓[/green] {binary} ({_op_version(binary)})")
if binary_path:
op_cfg["binary_path"] = binary_path
# ----------------------------------------------------------------- account
if args.account and args.account.strip():
op_cfg["account"] = args.account.strip()
console.print(f" Account: [cyan]{op_cfg['account']}[/cyan]")
# ------------------------------------------------------------------- token
console.print()
console.print("[bold]Step 2[/bold] Authentication")
token_env = (args.token_env or op_cfg.get("service_account_token_env")
or _DEFAULT_TOKEN_ENV).strip()
op_cfg["service_account_token_env"] = token_env
token = (args.token or "").strip()
if token:
save_env_value(token_env, token)
os.environ[token_env] = token
console.print(f" [green]✓[/green] service-account token stored in "
f"{get_env_path()} as {token_env}")
elif os.environ.get(token_env):
console.print(f" [green]✓[/green] using service-account token from {token_env}")
else:
who = _op_whoami(binary, op_cfg.get("account", ""))
if who:
console.print(f" [green]✓[/green] using existing op session ({who})")
else:
console.print(
" [yellow]No service-account token and no active op session "
"detected.[/yellow]\n"
" Either run [cyan]op signin[/cyan] (desktop/interactive) or set a "
f"service-account token in {token_env}, then re-run status."
)
# ----------------------------------------------------------------- enable
op_cfg["enabled"] = True
op_cfg.setdefault("env", {})
op_cfg.setdefault("cache_ttl_seconds", 300)
op_cfg.setdefault("override_existing", True)
save_config(cfg)
console.print()
console.print("[green]✓ 1Password secret source is enabled.[/green]")
console.print(
" Map credentials: [cyan]hermes secrets onepassword set OPENAI_API_KEY "
"\"op://Private/OpenAI/api key\"[/cyan]\n"
" Preview: [cyan]hermes secrets onepassword sync[/cyan]\n"
" Status: [cyan]hermes secrets onepassword status[/cyan]"
)
return 0
def cmd_status(args: argparse.Namespace) -> int:
console = Console()
cfg = load_config()
op_cfg = (cfg.get("secrets") or {}).get("onepassword") or {}
enabled = bool(op_cfg.get("enabled"))
account = str(op_cfg.get("account", "") or "").strip()
token_env = op_cfg.get("service_account_token_env", _DEFAULT_TOKEN_ENV)
binary_path = str(op_cfg.get("binary_path", "") or "").strip()
references = op_cfg.get("env") if isinstance(op_cfg.get("env"), dict) else {}
token_set = bool(os.environ.get(token_env))
binary = op_src.find_op(binary_path)
table = Table(show_header=False, box=None, padding=(0, 2))
table.add_column("", style="bold")
table.add_column("")
table.add_row("Enabled", _yn(enabled))
table.add_row("Account", account or "[dim]default[/dim]")
table.add_row("Token env var", token_env)
table.add_row("Token in env", _yn(token_set))
table.add_row("Override existing", _yn(bool(op_cfg.get("override_existing", True))))
table.add_row("Cache TTL (s)", str(op_cfg.get("cache_ttl_seconds", 300)))
if binary:
table.add_row("op binary", f"{binary} ({_op_version(binary)})")
else:
table.add_row("op binary", "[yellow]not found[/yellow]")
table.add_row("References", str(len(references)))
console.print(Panel(table, title="1Password secret source", border_style="cyan"))
if references:
ref_table = Table(show_header=True, header_style="bold")
ref_table.add_column("Env var", style="cyan")
ref_table.add_column("Reference")
for name in sorted(references):
ref_table.add_row(name, str(references[name]))
console.print(ref_table)
if not enabled:
console.print("\n Run [cyan]hermes secrets onepassword setup[/cyan] to enable.")
return 0
if binary and not token_set:
who = _op_whoami(binary, account)
if who:
console.print(f"\n [green]Active op session:[/green] {who}")
else:
console.print(
f"\n [yellow]No active op session and {token_env} is unset — "
"Hermes will warn and skip 1Password on next startup.[/yellow]"
)
if not references:
console.print(
"\n [yellow]No references mapped yet.[/yellow] Add one: "
"[cyan]hermes secrets onepassword set ENV_VAR \"op://…\"[/cyan]"
)
return 0
def cmd_set(args: argparse.Namespace) -> int:
console = Console()
# Reuse the backend validator so the CLI and startup paths agree on what a
# valid reference is — and store the *validated/stripped* value, not the
# raw arg (so trailing whitespace never lands in config.yaml).
valid, warnings = op_src._validate_references({args.env_var: args.reference})
if args.env_var not in valid:
for w in warnings:
console.print(f"[red]{w}[/red]")
return 1
cfg = load_config()
op_cfg = cfg.setdefault("secrets", {}).setdefault("onepassword", {})
env_map = op_cfg.get("env")
if not isinstance(env_map, dict):
env_map = {}
op_cfg["env"] = env_map
env_map[args.env_var] = valid[args.env_var]
save_config(cfg)
console.print(
f"[green]✓[/green] mapped [cyan]{args.env_var}[/cyan] → "
f"{valid[args.env_var]}"
)
if not op_cfg.get("enabled"):
console.print(
" [yellow]Note: the integration is disabled — run "
"[cyan]hermes secrets onepassword setup[/cyan] to turn it on.[/yellow]"
)
return 0
def cmd_remove(args: argparse.Namespace) -> int:
console = Console()
cfg = load_config()
op_cfg = cfg.setdefault("secrets", {}).setdefault("onepassword", {})
env_map = op_cfg.get("env")
if not isinstance(env_map, dict) or args.env_var not in env_map:
console.print(f"[yellow]{args.env_var} is not mapped.[/yellow]")
return 1
del env_map[args.env_var]
save_config(cfg)
console.print(f"[green]✓[/green] removed mapping for [cyan]{args.env_var}[/cyan]")
return 0
def cmd_sync(args: argparse.Namespace) -> int:
console = Console()
cfg = load_config()
op_cfg = (cfg.get("secrets") or {}).get("onepassword") or {}
if not op_cfg.get("enabled"):
console.print(
"[yellow]1Password integration is disabled. Run "
"`hermes secrets onepassword setup` first.[/yellow]"
)
return 1
references = op_cfg.get("env") if isinstance(op_cfg.get("env"), dict) else {}
if not references:
console.print(
"[yellow]No op:// references configured. Add one with "
"`hermes secrets onepassword set ENV_VAR \"op://…\"`.[/yellow]"
)
return 0
account = str(op_cfg.get("account", "") or "").strip()
token_env = op_cfg.get("service_account_token_env", _DEFAULT_TOKEN_ENV)
binary_path = str(op_cfg.get("binary_path", "") or "").strip()
# --apply delegates to the same code path startup uses, so the skip /
# override / token-guard policy lives in exactly one place.
if args.apply:
result = op_src.apply_onepassword_secrets(
enabled=True,
env=references,
account=account,
service_account_token_env=token_env,
binary_path=binary_path,
override_existing=bool(op_cfg.get("override_existing", True)),
cache_ttl_seconds=0, # an explicit sync always resolves fresh
)
if result.error:
console.print(f"[red]{result.error}[/red]")
return 1
table = Table(show_header=True, header_style="bold")
table.add_column("Env var", style="cyan")
table.add_column("Action")
for name in sorted(result.applied):
table.add_row(name, "[green]exported[/green]")
for name in sorted(result.skipped):
table.add_row(name, "[dim]skipped (already set / token var)[/dim]")
console.print(table)
for w in result.warnings:
console.print(f"[yellow]warning:[/yellow] {w}")
console.print(
f"\n [green]Exported {len(result.applied)} secret(s) into current "
"process.[/green]"
)
return 0
# Dry-run: resolve fresh (no cache) and preview, mutating nothing.
try:
secrets, warnings = op_src.fetch_onepassword_secrets(
references=references,
account=account,
token_env=token_env,
binary_path=binary_path,
use_cache=False,
)
except RuntimeError as exc:
console.print(f"[red]{exc}[/red]")
return 1
override = bool(op_cfg.get("override_existing", True))
table = Table(show_header=True, header_style="bold")
table.add_column("Env var", style="cyan")
table.add_column("Action")
for name in sorted(references):
if name == token_env:
table.add_row(name, "[dim]skip (token var)[/dim]")
elif name not in secrets:
table.add_row(name, "[red]unresolved (see warnings)[/red]")
elif os.environ.get(name) and not override:
table.add_row(name, "[dim]skip (already set)[/dim]")
else:
already = bool(os.environ.get(name))
table.add_row(
name,
"[green]would export[/green]" + (" (overrides)" if already else ""),
)
console.print(table)
for w in warnings:
console.print(f"[yellow]warning:[/yellow] {w}")
console.print(
"\n This was a dry-run — references resolve automatically on the next "
"[cyan]hermes[/cyan] invocation. Re-run with [cyan]--apply[/cyan] to export "
"into the current shell instead."
)
return 0
def cmd_disable(args: argparse.Namespace) -> int:
console = Console()
cfg = load_config()
op_cfg = cfg.setdefault("secrets", {}).setdefault("onepassword", {})
op_cfg["enabled"] = False
save_config(cfg)
console.print(
"[green]Disabled.[/green] 1Password references will NOT be resolved on the "
"next Hermes invocation.\n"
" Your reference mappings are left in config.yaml — remove them with "
"[cyan]hermes secrets onepassword remove ENV_VAR[/cyan] if you no longer "
"need them."
)
return 0
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _yn(b: bool) -> str:
return "[green]yes[/green]" if b else "[dim]no[/dim]"
def _op_version(binary: Path) -> str:
try:
res = subprocess.run(
[str(binary), "--version"],
capture_output=True,
text=True,
timeout=5,
)
if res.returncode == 0:
return (res.stdout or res.stderr).strip().splitlines()[0]
except (OSError, subprocess.TimeoutExpired):
pass
return "version unknown"
def _op_whoami(binary: Path, account: str) -> Optional[str]:
"""Return a short identity string if op is authenticated, else None."""
cmd = [str(binary), "whoami"]
if account:
cmd += ["--account", account]
try:
res = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
except (OSError, subprocess.TimeoutExpired):
return None
if res.returncode != 0:
return None
out = (res.stdout or "").strip()
return out.replace("\n", " ")[:120] or "authenticated"

View file

@ -63,6 +63,14 @@ def test_format_secret_source_suffix_generic_label_for_future_sources():
)
def test_format_secret_source_suffix_onepassword_uses_proper_name():
env_loader._SECRET_SOURCES["OPENAI_API_KEY"] = "onepassword"
assert (
env_loader.format_secret_source_suffix("OPENAI_API_KEY")
== " (from 1Password)"
)
def test_apply_external_secret_sources_records_bitwarden_origin(tmp_path, monkeypatch):
"""End-to-end: when the Bitwarden source fetches keys, applied vars
end up in ``_SECRET_SOURCES`` so the UI can label them."""
@ -174,3 +182,89 @@ def test_apply_external_secret_sources_dedupes_within_process(tmp_path, monkeypa
env_loader.reset_secret_source_cache()
env_loader._apply_external_secret_sources(tmp_path)
assert call_count["n"] == 2
def test_apply_external_secret_sources_records_onepassword_origin(tmp_path, monkeypatch):
"""When ``apply_onepassword_secrets`` returns applied keys, they end up in
``_SECRET_SOURCES`` labeled ``onepassword``."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
(tmp_path / "config.yaml").write_text(
"secrets:\n"
" onepassword:\n"
" enabled: true\n"
" env:\n"
" ANTHROPIC_API_KEY: 'op://Private/Anthropic/credential'\n",
encoding="utf-8",
)
from agent.secret_sources.onepassword import FetchResult
def _fake_apply(**_kwargs):
return FetchResult(
secrets={"ANTHROPIC_API_KEY": "sk-ant-test"},
applied=["ANTHROPIC_API_KEY"],
)
import agent.secret_sources.onepassword as op_module
monkeypatch.setattr(op_module, "apply_onepassword_secrets", _fake_apply)
env_loader._apply_external_secret_sources(tmp_path)
assert env_loader.get_secret_source("ANTHROPIC_API_KEY") == "onepassword"
assert (
env_loader.format_secret_source_suffix("ANTHROPIC_API_KEY")
== " (from 1Password)"
)
def test_apply_external_secret_sources_survives_non_dict_section(tmp_path, monkeypatch):
"""A malformed `secrets:` section must not abort startup (fail-open).
Both `onepassword: true` (non-dict) and a bad bitwarden section must be
coerced to empty config instead of raising AttributeError up through
load_hermes_dotenv().
"""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
(tmp_path / "config.yaml").write_text(
"secrets:\n"
" bitwarden: true\n"
" onepassword: true\n",
encoding="utf-8",
)
# Must not raise and must not record anything.
env_loader._apply_external_secret_sources(tmp_path)
assert env_loader.get_secret_source("ANYTHING") is None
def test_apply_external_secret_sources_bad_ttl_does_not_crash(tmp_path, monkeypatch):
"""A non-numeric cache_ttl_seconds must be coerced, not crash startup."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
(tmp_path / "config.yaml").write_text(
"secrets:\n"
" onepassword:\n"
" enabled: true\n"
" cache_ttl_seconds: not-a-number\n"
" env:\n"
" K: 'op://V/I/F'\n",
encoding="utf-8",
)
captured = {}
from agent.secret_sources.onepassword import FetchResult
def _fake_apply(**kwargs):
captured.update(kwargs)
return FetchResult()
import agent.secret_sources.onepassword as op_module
monkeypatch.setattr(op_module, "apply_onepassword_secrets", _fake_apply)
env_loader._apply_external_secret_sources(tmp_path)
# Coerced to the 300s default rather than raising ValueError.
assert captured["cache_ttl_seconds"] == 300

View file

@ -0,0 +1,484 @@
"""Hermetic tests for the 1Password (`op` CLI) secret source.
We never invoke the real ``op`` binary: ``subprocess.run`` is mocked so the
suite stays fast and offline-safe. A live resolve is exercised manually via
``hermes secrets onepassword sync`` outside of pytest.
"""
from __future__ import annotations
import json
import os
import subprocess
import sys
import time
from pathlib import Path
from unittest import mock
import pytest
# Make the worktree importable without depending on the installed wheel.
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from agent.secret_sources import onepassword as op # noqa: E402
@pytest.fixture(autouse=True)
def _reset_caches():
op._reset_cache_for_tests()
yield
op._reset_cache_for_tests()
@pytest.fixture(autouse=True)
def _clean_op_env(monkeypatch):
"""Start every test from a known 1Password auth state."""
for key in list(os.environ):
if key.startswith("OP_SESSION_"):
monkeypatch.delenv(key, raising=False)
monkeypatch.delenv("OP_SERVICE_ACCOUNT_TOKEN", raising=False)
monkeypatch.delenv("OP_ACCOUNT", raising=False)
yield
def _ok(value: str):
return mock.Mock(returncode=0, stdout=value, stderr="")
def _err(code: int, stderr: str):
return mock.Mock(returncode=code, stdout="", stderr=stderr)
# ---------------------------------------------------------------------------
# Reference validation
# ---------------------------------------------------------------------------
def test_validate_references_filters_bad_names_and_refs():
refs = {
"OPENAI_API_KEY": "op://Private/OpenAI/api key",
"1BAD_NAME": "op://Private/x/y", # bad env name
"HAS SPACE": "op://Private/x/y", # bad env name
"NOT_A_REF": "https://example.com", # not op://
"WHITESPACE": " op://Private/z/field ", # stripped + kept
}
valid, warnings = op._validate_references(refs)
assert valid == {
"OPENAI_API_KEY": "op://Private/OpenAI/api key",
"WHITESPACE": "op://Private/z/field",
}
assert len(warnings) == 3
# ---------------------------------------------------------------------------
# fetch_onepassword_secrets
# ---------------------------------------------------------------------------
def test_fetch_happy_path(monkeypatch, tmp_path):
fake_op = tmp_path / "op"
fake_op.write_text("")
values = {
"op://Private/OpenAI/api key": "sk-abc\n",
"op://Private/Anthropic/credential": "sk-ant-xyz",
}
def fake_run(cmd, **kwargs):
# argv list, never shell=True; reference passed after `--`.
assert "--" in cmd
ref = cmd[cmd.index("--") + 1]
return _ok(values[ref])
monkeypatch.setattr(op.subprocess, "run", fake_run)
secrets, warnings = op.fetch_onepassword_secrets(
references={
"OPENAI_API_KEY": "op://Private/OpenAI/api key",
"ANTHROPIC_API_KEY": "op://Private/Anthropic/credential",
},
binary=fake_op,
use_cache=False,
)
assert secrets == {"OPENAI_API_KEY": "sk-abc", "ANTHROPIC_API_KEY": "sk-ant-xyz"}
assert warnings == []
def test_fetch_uses_option_terminator_and_account(monkeypatch, tmp_path):
fake_op = tmp_path / "op"
fake_op.write_text("")
captured = {}
def fake_run(cmd, **kwargs):
captured["cmd"] = cmd
return _ok("value")
monkeypatch.setattr(op.subprocess, "run", fake_run)
op.fetch_onepassword_secrets(
references={"K": "op://V/I/F"},
account="my.1password.com",
binary=fake_op,
use_cache=False,
)
cmd = captured["cmd"]
assert cmd[:2] == [str(fake_op), "read"]
assert "--account" in cmd and "my.1password.com" in cmd
# `--` must precede the positional reference.
assert cmd[-2:] == ["--", "op://V/I/F"]
def test_fetch_empty_rc0_does_not_clobber(monkeypatch, tmp_path):
"""returncode 0 with empty stdout must surface as a warning, not a value."""
fake_op = tmp_path / "op"
fake_op.write_text("")
monkeypatch.setattr(op.subprocess, "run", lambda *a, **k: _ok(" \n"))
secrets, warnings = op.fetch_onepassword_secrets(
references={"K": "op://V/I/F"}, binary=fake_op, use_cache=False
)
assert secrets == {}
assert any("empty value" in w for w in warnings)
def test_fetch_read_failure_becomes_warning(monkeypatch, tmp_path):
fake_op = tmp_path / "op"
fake_op.write_text("")
monkeypatch.setattr(
op.subprocess, "run", lambda *a, **k: _err(1, "\x1b[31m[ERROR] not signed in\x1b[0m")
)
secrets, warnings = op.fetch_onepassword_secrets(
references={"K": "op://V/I/F"}, binary=fake_op, use_cache=False
)
assert secrets == {}
assert len(warnings) == 1
# ANSI control sequences are fully scrubbed from the surfaced message.
assert "\x1b" not in warnings[0]
assert "[31m" not in warnings[0]
assert "not signed in" in warnings[0]
def test_fetch_one_bad_one_good(monkeypatch, tmp_path):
fake_op = tmp_path / "op"
fake_op.write_text("")
def fake_run(cmd, **kwargs):
ref = cmd[cmd.index("--") + 1]
if ref == "op://V/good/f":
return _ok("good-value")
return _err(1, "no access")
monkeypatch.setattr(op.subprocess, "run", fake_run)
secrets, warnings = op.fetch_onepassword_secrets(
references={"GOOD": "op://V/good/f", "BAD": "op://V/bad/f"},
binary=fake_op,
use_cache=False,
)
assert secrets == {"GOOD": "good-value"}
assert len(warnings) == 1
def test_fetch_missing_binary_raises(monkeypatch):
monkeypatch.setattr(op, "find_op", lambda binary_path="": None)
with pytest.raises(RuntimeError, match="op CLI not found"):
op.fetch_onepassword_secrets(
references={"K": "op://V/I/F"}, use_cache=False
)
def test_fetch_child_env_is_allowlisted(monkeypatch, tmp_path):
"""The op child must NOT inherit unrelated provider credentials."""
fake_op = tmp_path / "op"
fake_op.write_text("")
monkeypatch.setenv("OPENAI_API_KEY", "leak-me")
monkeypatch.setenv("OP_SERVICE_ACCOUNT_TOKEN", "ops_tok")
monkeypatch.setenv("OP_SESSION_myacct", "sess123")
captured = {}
def fake_run(cmd, **kwargs):
captured["env"] = kwargs["env"]
return _ok("v")
monkeypatch.setattr(op.subprocess, "run", fake_run)
op.fetch_onepassword_secrets(
references={"K": "op://V/I/F"}, binary=fake_op, use_cache=False
)
env = captured["env"]
assert "OPENAI_API_KEY" not in env # not inherited
assert env["OP_SERVICE_ACCOUNT_TOKEN"] == "ops_tok"
assert env["OP_SESSION_myacct"] == "sess123"
assert env.get("NO_COLOR") == "1"
# ---------------------------------------------------------------------------
# Caching
# ---------------------------------------------------------------------------
def test_inprocess_cache_hit(monkeypatch, tmp_path):
fake_op = tmp_path / "op"
fake_op.write_text("")
calls = {"n": 0}
def fake_run(*a, **k):
calls["n"] += 1
return _ok("v")
monkeypatch.setattr(op.subprocess, "run", fake_run)
op._reset_cache_for_tests(tmp_path)
for _ in range(2):
op.fetch_onepassword_secrets(
references={"K": "op://V/I/F"}, cache_ttl_seconds=60,
binary=fake_op, home_path=tmp_path,
)
assert calls["n"] == 1 # second call served from L1 cache
def test_disk_cache_roundtrip_and_no_token_on_disk(monkeypatch, tmp_path):
fake_op = tmp_path / "op"
fake_op.write_text("")
monkeypatch.setenv("OP_SERVICE_ACCOUNT_TOKEN", "ops_supersecret")
calls = {"n": 0}
def fake_run(*a, **k):
calls["n"] += 1
return _ok("resolved")
monkeypatch.setattr(op.subprocess, "run", fake_run)
op._reset_cache_for_tests(tmp_path)
op.fetch_onepassword_secrets(
references={"K": "op://V/I/F"}, cache_ttl_seconds=300,
binary=fake_op, home_path=tmp_path,
)
assert calls["n"] == 1
cache_path = op._disk_cache_path(tmp_path)
assert cache_path.exists()
assert (os.stat(cache_path).st_mode & 0o777) == 0o600
text = cache_path.read_text()
assert "ops_supersecret" not in text # token never on disk
payload = json.loads(text)
assert payload["secrets"] == {"K": "resolved"}
# Simulate a fresh process: clear only the in-process cache.
op._CACHE.clear()
op.fetch_onepassword_secrets(
references={"K": "op://V/I/F"}, cache_ttl_seconds=300,
binary=fake_op, home_path=tmp_path,
)
assert calls["n"] == 1 # served from disk, op not re-invoked
def test_ttl_zero_disables_both_layers(monkeypatch, tmp_path):
fake_op = tmp_path / "op"
fake_op.write_text("")
calls = {"n": 0}
def fake_run(*a, **k):
calls["n"] += 1
return _ok("v")
monkeypatch.setattr(op.subprocess, "run", fake_run)
op._reset_cache_for_tests(tmp_path)
op.fetch_onepassword_secrets(
references={"K": "op://V/I/F"}, cache_ttl_seconds=0,
binary=fake_op, home_path=tmp_path,
)
# No disk file written when TTL is 0.
assert not op._disk_cache_path(tmp_path).exists()
op._CACHE.clear()
op.fetch_onepassword_secrets(
references={"K": "op://V/I/F"}, cache_ttl_seconds=0,
binary=fake_op, home_path=tmp_path,
)
assert calls["n"] == 2 # never cached
def test_session_change_invalidates_cache(monkeypatch, tmp_path):
"""A different OP_SESSION_* identity must not reuse a cached value."""
fake_op = tmp_path / "op"
fake_op.write_text("")
calls = {"n": 0}
def fake_run(*a, **k):
calls["n"] += 1
return _ok("v")
monkeypatch.setattr(op.subprocess, "run", fake_run)
op._reset_cache_for_tests(tmp_path)
monkeypatch.setenv("OP_SESSION_acctA", "sessA")
op.fetch_onepassword_secrets(
references={"K": "op://V/I/F"}, cache_ttl_seconds=300,
binary=fake_op, home_path=tmp_path,
)
# Switch identity.
monkeypatch.delenv("OP_SESSION_acctA", raising=False)
monkeypatch.setenv("OP_SESSION_acctB", "sessB")
op._CACHE.clear()
op.fetch_onepassword_secrets(
references={"K": "op://V/I/F"}, cache_ttl_seconds=300,
binary=fake_op, home_path=tmp_path,
)
assert calls["n"] == 2 # cache key changed → refetch
def test_partial_failure_not_cached(monkeypatch, tmp_path):
fake_op = tmp_path / "op"
fake_op.write_text("")
def fake_run(cmd, **kwargs):
ref = cmd[cmd.index("--") + 1]
return _ok("v") if ref == "op://V/good/f" else _err(1, "fail")
monkeypatch.setattr(op.subprocess, "run", fake_run)
op._reset_cache_for_tests(tmp_path)
op.fetch_onepassword_secrets(
references={"G": "op://V/good/f", "B": "op://V/bad/f"},
cache_ttl_seconds=300, binary=fake_op, home_path=tmp_path,
)
# A pull with any read error must not be persisted.
assert not op._disk_cache_path(tmp_path).exists()
def test_reset_cache_clears_disk(tmp_path):
cache_path = op._disk_cache_path(tmp_path)
cache_path.parent.mkdir(parents=True, exist_ok=True)
cache_path.write_text("{}")
assert cache_path.exists()
op._reset_cache_for_tests(tmp_path)
assert not cache_path.exists()
op._reset_cache_for_tests(tmp_path) # idempotent
# ---------------------------------------------------------------------------
# find_op
# ---------------------------------------------------------------------------
def test_find_op_pinned_path_not_on_path(tmp_path, monkeypatch):
pinned = tmp_path / "op"
pinned.write_text("")
pinned.chmod(0o755)
# PATH lookup must NOT be consulted when a binary_path is pinned.
monkeypatch.setattr(op.shutil, "which", lambda name: "/usr/bin/op")
assert op.find_op(str(pinned)) == pinned
def test_find_op_pinned_missing_returns_none(tmp_path, monkeypatch):
monkeypatch.setattr(op.shutil, "which", lambda name: "/usr/bin/op")
assert op.find_op(str(tmp_path / "nope")) is None
# ---------------------------------------------------------------------------
# apply_onepassword_secrets
# ---------------------------------------------------------------------------
def test_apply_disabled_returns_empty():
result = op.apply_onepassword_secrets(enabled=False, env={"K": "op://V/I/F"})
assert result.ok
assert not result.applied
def test_apply_missing_binary_sets_error(monkeypatch):
monkeypatch.setattr(op, "find_op", lambda binary_path="": None)
result = op.apply_onepassword_secrets(
enabled=True, env={"K": "op://V/I/F"}
)
assert not result.ok
assert "op CLI" in result.error
def test_apply_sets_env(monkeypatch, tmp_path):
fake_op = tmp_path / "op"
fake_op.write_text("")
monkeypatch.setattr(op, "find_op", lambda binary_path="": fake_op)
monkeypatch.setattr(op.subprocess, "run", lambda *a, **k: _ok("resolved-val"))
monkeypatch.delenv("MY_OP_KEY", raising=False)
result = op.apply_onepassword_secrets(
enabled=True, env={"MY_OP_KEY": "op://V/I/F"}, cache_ttl_seconds=0,
)
assert result.ok
assert result.applied == ["MY_OP_KEY"]
assert os.environ["MY_OP_KEY"] == "resolved-val"
def test_apply_skips_before_fetch_when_not_overriding(monkeypatch, tmp_path):
fake_op = tmp_path / "op"
fake_op.write_text("")
monkeypatch.setattr(op, "find_op", lambda binary_path="": fake_op)
monkeypatch.setenv("MY_OP_KEY", "from-env")
calls = {"n": 0}
def fake_run(*a, **k):
calls["n"] += 1
return _ok("from-1password")
monkeypatch.setattr(op.subprocess, "run", fake_run)
result = op.apply_onepassword_secrets(
enabled=True, env={"MY_OP_KEY": "op://V/I/F"},
override_existing=False, cache_ttl_seconds=0,
)
assert "MY_OP_KEY" in result.skipped
assert os.environ["MY_OP_KEY"] == "from-env"
assert calls["n"] == 0 # never even called op for a value we'd discard
def test_apply_never_overrides_token_var(monkeypatch, tmp_path):
fake_op = tmp_path / "op"
fake_op.write_text("")
monkeypatch.setattr(op, "find_op", lambda binary_path="": fake_op)
monkeypatch.setenv("OP_SERVICE_ACCOUNT_TOKEN", "original")
calls = {"n": 0}
def fake_run(*a, **k):
calls["n"] += 1
return _ok("malicious")
monkeypatch.setattr(op.subprocess, "run", fake_run)
result = op.apply_onepassword_secrets(
enabled=True,
env={"OP_SERVICE_ACCOUNT_TOKEN": "op://V/I/F"},
override_existing=True, cache_ttl_seconds=0,
)
assert "OP_SERVICE_ACCOUNT_TOKEN" in result.skipped
assert os.environ["OP_SERVICE_ACCOUNT_TOKEN"] == "original"
assert calls["n"] == 0
def test_apply_never_raises_on_read_failure(monkeypatch, tmp_path):
fake_op = tmp_path / "op"
fake_op.write_text("")
monkeypatch.setattr(op, "find_op", lambda binary_path="": fake_op)
monkeypatch.setattr(op.subprocess, "run", lambda *a, **k: _err(1, "locked"))
monkeypatch.delenv("MY_OP_KEY", raising=False)
result = op.apply_onepassword_secrets(
enabled=True, env={"MY_OP_KEY": "op://V/I/F"}, cache_ttl_seconds=0,
)
# Fail-open: warnings, nothing applied, no fatal error, no exception.
assert result.ok
assert result.applied == []
assert result.warnings
def test_apply_no_valid_refs_is_noop(monkeypatch):
# find_op must never be reached when there's nothing to fetch.
monkeypatch.setattr(
op, "find_op",
lambda binary_path="": (_ for _ in ()).throw(AssertionError("should not resolve op")),
)
result = op.apply_onepassword_secrets(enabled=True, env={"BAD NAME": "op://V/I/F"})
assert result.ok
assert result.applied == []
assert result.warnings # the bad mapping warned

View file

@ -5,6 +5,7 @@ Hermes can pull API keys from external secret managers at process startup instea
Supported:
- [Bitwarden Secrets Manager](./bitwarden) — `bws` CLI, lazy-installed, free tier works.
- [1Password](./onepassword) — `op://` references via the official `op` CLI; service-account or desktop session auth.
## Multiple sources at once
@ -30,4 +31,4 @@ Every credential injected by a source is labelled with its origin — setup flow
Third-party secret managers ship as standalone plugins, not core PRs. A backend subclasses `agent.secret_sources.base.SecretSource` (one required method: `fetch(cfg, home_path) -> FetchResult`) and registers via `ctx.register_secret_source(MySource())` in the plugin's `register(ctx)`. The orchestrator owns precedence, conflict handling, timeouts, and provenance — your source only fetches. Contract rules: `fetch()` never raises, never prompts, and returns within its timeout budget; validate your implementation against the conformance kit in `tests/secret_sources/conformance.py`.
The bundled set is deliberately closed (same policy as memory providers). Planned in-tree additions: 1Password. Everything else — Infisical, Proton Pass, HashiCorp Vault, AWS Secrets Manager, OS keystores — belongs in plugin repos; share them in the Nous Research Discord (`#plugins-skills-and-skins`).
The bundled set is deliberately closed (same policy as memory providers): Bitwarden and 1Password ship in-tree. Everything else — Infisical, Proton Pass, HashiCorp Vault, AWS Secrets Manager, OS keystores — belongs in plugin repos; share them in the Nous Research Discord (`#plugins-skills-and-skins`).

View file

@ -0,0 +1,140 @@
# 1Password
Resolve provider API keys from [1Password](https://1password.com/) at process startup instead of storing them in plaintext inside `~/.hermes/.env`. You keep your keys as 1Password items and reference them by `op://vault/item/field`; rotating a credential becomes a single change in 1Password.
## How it works
1. You install the official [1Password CLI](https://developer.1password.com/docs/cli/get-started/) (`op`) and authenticate it — either with a **service-account token** (headless servers) or an **interactive/desktop session** (your laptop).
2. You map environment-variable names to `op://` references in `~/.hermes/config.yaml`.
3. Every time `hermes` (or the gateway, or a cron job) starts, after `~/.hermes/.env` has loaded, Hermes runs `op read` for each reference and sets the resolved values into `os.environ`.
4. By default Hermes **overrides** values already in your environment, so 1Password is the source of truth — rotate a credential once and every Hermes process picks it up on next start. Flip `override_existing: false` if you want `.env` to win instead.
Hermes never authenticates on your behalf and never downloads `op`: it shells out to your already-installed, already-trusted CLI. If `op` is missing, your session is locked, or a reference is wrong, Hermes prints a one-line warning and continues with whatever credentials `.env` already had — it never blocks startup.
## Authentication
`op` supports two non-interactive-friendly modes; Hermes works with either:
- **Service accounts** (recommended for servers/CI): create a service account in 1Password, grant it read access to the relevant vault, and export its token as `OP_SERVICE_ACCOUNT_TOKEN` in `~/.hermes/.env`. The token is the credential — treat it like any other bearer token.
- **Desktop / interactive sessions** (laptops): run `op signin` (or enable CLI integration in the 1Password app). Hermes passes your `OP_SESSION_*` variables through to the `op` child process. The 1Password cache key includes those session variables, so signing into a different account never serves a value cached under the previous identity.
## Setup
### 1. Install and sign in to `op`
Follow the [1Password CLI getting-started guide](https://developer.1password.com/docs/cli/get-started/). Verify it works:
```bash
op whoami
```
### 2. Enable the integration
```bash
hermes secrets onepassword setup
```
This verifies `op` is on `PATH` (or use `--binary-path`), records your account/token settings, checks for an active session, and flips `secrets.onepassword.enabled: true`. Non-interactive flags:
```bash
hermes secrets onepassword setup \
--account my.1password.com \
--token-env OP_SERVICE_ACCOUNT_TOKEN \
--token "$OP_SERVICE_ACCOUNT_TOKEN"
```
### 3. Map your credentials
The reference format is `op://<vault>/<item>/<field>`:
```bash
hermes secrets onepassword set OPENAI_API_KEY "op://Private/OpenAI/api key"
hermes secrets onepassword set ANTHROPIC_API_KEY "op://Private/Anthropic/credential"
```
### 4. Preview and confirm
```bash
hermes secrets onepassword sync # dry-run: resolve now, show what would apply
hermes secrets onepassword status # config + binary + references + auth
```
From now on, every `hermes` invocation resolves the references at startup. You'll see a one-line summary in stderr the first time secrets are applied in a process.
## CLI
| Command | What it does |
|---|---|
| `hermes secrets onepassword setup` | Verify `op`, set account / token env var, enable |
| `hermes secrets onepassword status` | Show config, binary, auth, and configured references |
| `hermes secrets onepassword set ENV_VAR "op://…"` | Map an env var to a reference (stored stripped + validated) |
| `hermes secrets onepassword remove ENV_VAR` | Drop a mapping |
| `hermes secrets onepassword sync` | Dry-run: resolve references now and show what would apply |
| `hermes secrets onepassword sync --apply` | Resolve and export into the current shell's environment |
| `hermes secrets onepassword disable` | Flip `enabled: false`; leaves mappings in place |
`op` and `1password` are accepted as aliases for `onepassword`.
## Configuration
Defaults in `~/.hermes/config.yaml`:
```yaml
secrets:
onepassword:
enabled: false
env:
OPENAI_API_KEY: "op://Private/OpenAI/api key"
ANTHROPIC_API_KEY: "op://Private/Anthropic/credential"
account: ""
service_account_token_env: OP_SERVICE_ACCOUNT_TOKEN
binary_path: ""
cache_ttl_seconds: 300
override_existing: true
```
| Key | Default | What it does |
|---|---|---|
| `enabled` | `false` | Master switch. When false, `op` is never invoked. |
| `env` | `{}` | Mapping of env-var name → `op://vault/item/field` reference. Entries whose name isn't a valid env-var name, or whose value isn't an `op://` reference, are skipped with a warning. |
| `account` | `""` | Account shorthand / sign-in address passed as `op read --account`. Empty uses `op`'s default account. |
| `service_account_token_env` | `OP_SERVICE_ACCOUNT_TOKEN` | Env var Hermes reads the service-account token from. Its value is exported to the `op` child as `OP_SERVICE_ACCOUNT_TOKEN` (the name `op` expects). Leave the var unset to use a desktop/interactive session. |
| `binary_path` | `""` | Absolute path to `op`. When set, it is used verbatim and `PATH` is **not** consulted — pin this to avoid trusting whatever `op` appears first on `PATH`. |
| `cache_ttl_seconds` | `300` | How long resolved values are reused (in-process and on disk). Set to `0` to disable **both** cache layers — no values are written to disk at all. |
| `override_existing` | `true` | When true, resolved values overwrite anything already in env (so rotation takes effect). Flip to `false` to let `.env` / shell exports win; those references are then skipped *before* `op` is invoked. |
## Failure modes
1Password never blocks Hermes startup. If anything goes wrong you'll see a one-line warning in stderr and Hermes continues:
| Symptom | Cause | Fix |
|---|---|---|
| `the op CLI was not found on PATH` | `op` not installed / not on PATH | Install the CLI, or set `secrets.onepassword.binary_path` |
| `op read failed for 'op://…': …` | Locked session, expired token, or no vault access | `op signin`, refresh the token, or grant the service account access |
| `op read returned an empty value for 'op://…'` | The referenced field exists but is empty | Fix the item/field in 1Password (an empty value is never applied — your existing env var is left intact) |
| `… is not an op:// secret reference` | A mapping value isn't an `op://` reference | Re-set it with the correct `op://vault/item/field` form |
| `op read timed out` | Network blocked or 1Password slow | Check connectivity / the desktop app integration |
## Caching
Successful, complete pulls are cached in-process and on disk under `<hermes_home>/cache/op_cache.json` (written atomically, mode `0600`), so back-to-back short-lived `hermes` invocations don't re-shell `op` for every reference. The cache:
- stores only resolved secret **values** — never the service-account token or any raw auth material (auth is fingerprinted into the cache key);
- is invalidated when the token, account, `OP_SESSION_*` variables, or the set of references change;
- is **not** written when a pull had any per-reference error, so a transient auth failure isn't frozen in for the TTL;
- is fully disabled — reads *and* writes — when `cache_ttl_seconds: 0`.
## Security notes
- A 1Password service-account token can read every secret the account has access to. Store it in `~/.hermes/.env` (not `config.yaml`), and revoke + regenerate from 1Password if it leaks.
- Hermes refuses to let a resolved value overwrite the token env var itself, even with `override_existing: true`.
- The `op` child process gets a minimal allowlisted environment (auth/session vars + `PATH`/`HOME`), not a copy of the full `os.environ`, so post-dotenv provider credentials aren't all inherited by the child.
- References are validated to start with `op://`, and the reference is passed after a `--` option terminator so a crafted value can't be parsed as an `op` flag.
## When NOT to use this
- **Single-machine personal setups** where `~/.hermes/.env` is fine.
- **Air-gapped environments** that can't reach 1Password.
- **CI/CD** where an existing secrets-injection mechanism is already wired up — pick one path, not two.
The good case for this is multi-machine fleets, shared dev boxes, gateway VPSes, or anywhere you want centralized rotation and revocation across multiple Hermes installations.