feat(secrets): one-command token rotation + actionable startup errors for all secret sources (#68605)

* feat(secrets): one-command token rotation + actionable startup errors for all secret sources

When a Bitwarden machine-account token expired, users saw a raw Rust
error dump (invalid_client + Location: + backtrace hints) and the only
fix was manually editing .env or re-running the whole setup wizard.

- New `hermes secrets bitwarden token` / `hermes secrets onepassword
  token`: paste a new token (masked prompt or flag), the command probes
  the backend BEFORE persisting — a rejected token changes nothing; a
  good one is written to .env and the fetch caches are cleared.
- New optional SecretSource.remediation(kind, cfg) hook: startup
  warnings now print a '→ Run `hermes secrets <name> token`…' fix-it
  line after any fetch error, for bundled AND plugin sources (generic
  per-ErrorKind defaults in the ABC).
- bws stderr is summarized to its cause line (Location:/backtrace noise
  dropped) and invalid_client/invalid_grant/400 identity rejects are
  now classified AUTH_FAILED (was INTERNAL) with a plain-English
  explanation naming the token env var.
- op whoami probe accepts a candidate token so rotation validates the
  NEW credential, not the ambient one.

Additive hook with defaults — no SECRET_SOURCE_API_VERSION bump.

* docs: fix MDX parse error in secret-source-plugin hook table

Escaped backticks around a <name> placeholder made MDX parse it as an
unclosed JSX tag, breaking the docs-site build.  Use a plain code span
instead.
This commit is contained in:
Teknium
2026-07-21 06:41:04 -07:00
committed by GitHub
parent ed3c39108b
commit d3b0e61429
13 changed files with 816 additions and 16 deletions
+39
View File
@@ -190,6 +190,45 @@ class SecretSource(ABC):
"""
return {}
def remediation(self, kind: Optional["ErrorKind"], cfg: dict) -> str:
"""One-line, actionable next step for a failed fetch.
Called by the startup status printer (and ``hermes secrets ...
status``) right after a fetch error is surfaced, so the user sees
*what to run* next to fix it — not just what broke. Sources
should override this to point at their own CLI verbs (e.g.
``hermes secrets bitwarden token`` for AUTH_FAILED). Return an
empty string to suppress the hint.
Must never raise and must not perform I/O — it's a pure
kind→string mapping on the startup path.
"""
generic = {
ErrorKind.NOT_CONFIGURED: (
f"Run `hermes secrets {self.name} setup` to finish configuration."
),
ErrorKind.BINARY_MISSING: (
f"Run `hermes secrets {self.name} setup` to install the helper CLI."
),
ErrorKind.AUTH_FAILED: (
f"Credentials rejected — run `hermes secrets {self.name} setup` "
"to re-authenticate."
),
ErrorKind.AUTH_EXPIRED: (
f"Credentials expired — run `hermes secrets {self.name} setup` "
"to re-authenticate."
),
ErrorKind.NETWORK: (
"Network problem reaching the secrets backend — check "
"connectivity and retry."
),
ErrorKind.TIMEOUT: (
f"Backend was slow — raise secrets.{self.name}.timeout_seconds "
"if this recurs."
),
}
return generic.get(kind, "") if kind is not None else ""
# ---------------------------------------------------------------------------
# Shared helpers — use these instead of hand-rolling per backend
+76 -6
View File
@@ -34,6 +34,7 @@ import json
import logging
import os
import platform
import re
import shutil
import stat
import subprocess
@@ -415,6 +416,39 @@ def fetch_bitwarden_secrets(
return secrets, warnings
def _summarize_bws_stderr(raw: str) -> str:
"""Reduce a bws (Rust color-eyre) error dump to its cause line(s).
bws failures look like::
Error:
0: Received error message from server: [400 Bad Request] {"error":"invalid_client"}
Location:
crates/bws/src/main.rs:108
...
Everything from ``Location:`` on is diagnostic noise for a Hermes
user. Keep the numbered cause lines (joined), drop the rest, and
fall back to the stripped raw text when the shape is unrecognized.
"""
text = raw.replace("\x1b", "").strip()
if not text:
return text
causes: List[str] = []
for line in text.splitlines():
stripped = line.strip()
if stripped.startswith(("Location:", "Backtrace omitted", "Run with ")):
break
if stripped in ("", "Error:"):
continue
# Cause lines are numbered "0: ...", "1: ..." — strip the index.
stripped = re.sub(r"^\d+:\s*", "", stripped)
if stripped:
causes.append(stripped)
return "; ".join(causes) if causes else text
def _run_bws_list(
bws: Path, access_token: str, project_id: str, server_url: str = ""
) -> Tuple[Dict[str, str], List[str]]:
@@ -448,9 +482,11 @@ def _run_bws_list(
raise RuntimeError(f"failed to invoke bws: {exc}") from exc
if proc.returncode != 0:
# bws writes auth/network errors to stderr in plain English.
# Strip ANSI just in case and surface the first 200 chars.
err = (proc.stderr or proc.stdout or "").strip().replace("\x1b", "")
# bws writes auth/network errors to stderr as a Rust error-report
# dump (color-eyre): an "Error:" header, indented cause lines, then
# "Location:" / "Backtrace omitted" noise. Strip ANSI and boil it
# down to the meaningful cause line(s) before surfacing.
err = _summarize_bws_stderr(proc.stderr or proc.stdout or "")
raise RuntimeError(
f"bws exited {proc.returncode}: {err[:200]}"
)
@@ -690,12 +726,30 @@ class BitwardenSource(SecretSource):
except RuntimeError as exc:
result.error = str(exc)
result.error_kind = _classify_bws_error(str(exc))
if result.error_kind == ErrorKind.AUTH_FAILED:
# Translate the raw OAuth reject into what it actually means
# for the user before the mechanics.
result.error = (
"Bitwarden rejected the machine-account access token "
f"({access_token_env}) — it was likely revoked, expired, "
f"or belongs to another region. ({result.error})"
)
return result
result.secrets = secrets
result.warnings.extend(warnings)
return result
def remediation(self, kind, cfg: dict) -> str:
if kind in (ErrorKind.AUTH_FAILED, ErrorKind.AUTH_EXPIRED):
return (
"Run `hermes secrets bitwarden token` to paste a fresh access "
"token (create one in the Bitwarden web app: Secrets Manager → "
"Machine accounts → Access tokens). Wrong region? Re-run "
"`hermes secrets bitwarden setup` and pick EU/self-hosted."
)
return super().remediation(kind, cfg)
def _classify_bws_error(message: str) -> ErrorKind:
"""Best-effort mapping of bws failure text onto the shared taxonomy."""
@@ -705,7 +759,13 @@ def _classify_bws_error(message: str) -> ErrorKind:
if "binary not available" in lowered or "failed to invoke" in lowered:
return ErrorKind.BINARY_MISSING
if any(tok in lowered for tok in ("unauthorized", "invalid token",
"access token", "401", "403")):
"access token", "401", "403",
# The BSM identity endpoint rejects a
# revoked/expired/deleted machine-account
# token with an OAuth-style
# `[400 Bad Request] {"error":"invalid_client"}`.
"invalid_client", "invalid_grant",
"400 bad request")):
return ErrorKind.AUTH_FAILED
if any(tok in lowered for tok in ("network", "connection", "resolve",
"download", "dns")):
@@ -718,6 +778,17 @@ def _classify_bws_error(message: str) -> ErrorKind:
# ---------------------------------------------------------------------------
def clear_caches(home_path: Optional[Path] = None) -> None:
"""Drop in-process AND disk caches.
Used after a token rotation (`hermes secrets bitwarden token`) so the
next startup fetches fresh with the new credential instead of serving
a pull cached under the old token's fingerprint.
"""
_CACHE.clear()
_DISK_CACHE.clear(home_path)
def _reset_cache_for_tests(home_path: Optional[Path] = None) -> None:
"""Clear in-process AND disk caches.
@@ -725,5 +796,4 @@ def _reset_cache_for_tests(home_path: Optional[Path] = None) -> None:
Without it we fall back to the same default resolution as the cache
writer itself.
"""
_CACHE.clear()
_DISK_CACHE.clear(home_path)
clear_caches(home_path)
+30 -2
View File
@@ -607,6 +607,24 @@ class OnePasswordSource(SecretSource):
result.warnings.extend(fetch_warnings)
return result
def remediation(self, kind, cfg: dict) -> str:
if kind in (ErrorKind.AUTH_FAILED, ErrorKind.AUTH_EXPIRED):
token_env = _DEFAULT_TOKEN_ENV
if isinstance(cfg, dict):
token_env = str(cfg.get("service_account_token_env") or token_env)
return (
"Run `hermes secrets onepassword token` to paste a fresh "
f"service-account token ({token_env}), or `op signin` for an "
"interactive session."
)
if kind == ErrorKind.BINARY_MISSING:
return (
"Install the 1Password CLI "
"(https://developer.1password.com/docs/cli/get-started/) or "
"set secrets.onepassword.binary_path."
)
return super().remediation(kind, cfg)
def _classify_op_error(message: str) -> ErrorKind:
"""Best-effort mapping of op failure text onto the shared taxonomy."""
@@ -633,11 +651,21 @@ def _classify_op_error(message: str) -> ErrorKind:
# ---------------------------------------------------------------------------
def clear_caches(home_path: Optional[Path] = None) -> None:
"""Drop in-process AND disk caches.
Used after a token rotation (`hermes secrets onepassword token`) so
the next startup resolves fresh with the new credential instead of
serving values cached under the old token's fingerprint.
"""
_CACHE.clear()
_DISK_CACHE.clear(home_path)
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)
clear_caches(home_path)
+23
View File
@@ -437,12 +437,35 @@ def _apply_external_secret_sources(home_path: Path) -> None:
)
if src.result.error:
print(f" {src.label}: {src.result.error}", file=sys.stderr)
hint = _remediation_hint(src.name, src.result.error_kind, cfg)
if hint:
print(f" {src.label}: → {hint}", file=sys.stderr)
for warn in src.result.warnings:
print(f" {src.label}: {warn}", file=sys.stderr)
for conflict in report.conflicts:
print(f" Secret sources: {conflict}", file=sys.stderr)
def _remediation_hint(source_name: str, error_kind, secrets_cfg: dict) -> str:
"""Ask the failed source for its one-line fix-it hint.
Defensive wrapper: remediation() is a pure mapping and shouldn't
raise, but a plugin source could — and startup must never break on
a status line.
"""
try:
from agent.secret_sources.registry import get_source
source = get_source(source_name)
if source is None:
return ""
src_cfg = secrets_cfg.get(source_name)
src_cfg = src_cfg if isinstance(src_cfg, dict) else {}
return str(source.remediation(error_kind, src_cfg) or "").strip()
except Exception: # noqa: BLE001 — hints must never block startup
return ""
def _load_secrets_config(home_path: Path) -> dict:
"""Read just the ``secrets:`` section out of config.yaml.
+95 -3
View File
@@ -18,6 +18,7 @@ from __future__ import annotations
import argparse
import os
import subprocess
import sys
from pathlib import Path
from typing import Optional
@@ -32,6 +33,7 @@ from hermes_cli.config import (
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/"
@@ -71,6 +73,21 @@ def register_cli(parent_parser: argparse.ArgumentParser) -> None:
status = sub.add_parser("status", help="Show config + op binary + references")
status.set_defaults(func=cmd_status)
token = sub.add_parser(
"token",
help="Rotate the service-account token: validate and store it in .env",
)
token.add_argument(
"--token",
help="Provide the new token non-interactively (default: masked prompt)",
)
token.add_argument(
"--no-verify",
action="store_true",
help="Store without probing 1Password first (not recommended)",
)
token.set_defaults(func=cmd_token)
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")
@@ -282,6 +299,68 @@ def cmd_remove(args: argparse.Namespace) -> int:
return 0
def cmd_token(args: argparse.Namespace) -> int:
"""Rotate the 1Password service-account token without the full setup flow.
Prompts for (or accepts via ``--token``) a new service-account token,
verifies it with ``op whoami`` (unless ``--no-verify``), and only then
persists it to .env — so a bad paste never bricks the working token.
"""
console = Console()
cfg = load_config()
op_cfg = (cfg.get("secrets") or {}).get("onepassword") or {}
token_env = op_cfg.get("service_account_token_env", _DEFAULT_TOKEN_ENV)
account = str(op_cfg.get("account", "") or "").strip()
binary_path = str(op_cfg.get("binary_path", "") or "").strip()
token = (args.token or "").strip()
if not token:
if not sys.stdin.isatty():
console.print("[red]No TTY — pass the token with --token.[/red]")
return 1
console.print(
"Create a new service-account token at "
"https://my.1password.com → Developer → Service Accounts.\n"
)
token = masked_secret_prompt(f"Paste new token ({token_env}): ").strip()
if not token:
console.print("[red]Empty token, aborting.[/red]")
return 1
if not args.no_verify:
binary = op_src.find_op(binary_path)
if binary is None:
console.print(
f"[red]op CLI not found — install it ({_DOCS_URL}) or "
"re-run with --no-verify to store anyway.[/red]"
)
return 1
console.print("Verifying with `op whoami`…")
who = _op_whoami(binary, account, token_value=token)
if who is None:
console.print(
"[red]✗ New token was rejected by op — nothing was changed.[/red]"
)
return 1
console.print(f"[green]✓ Token accepted[/green] ({who}).")
save_env_value(token_env, token)
os.environ[token_env] = token
# Cached resolutions are keyed on the previous token's fingerprint;
# drop them so the next startup resolves fresh with the new credential.
op_src.clear_caches()
console.print(
f"[green]✓[/green] stored in {get_env_path()} as {token_env}. "
"Takes effect on the next Hermes invocation."
)
if not op_cfg.get("enabled"):
console.print(
"[yellow]Note: the 1Password integration is currently disabled — "
"run `hermes secrets onepassword setup` to turn it on.[/yellow]"
)
return 0
def cmd_sync(args: argparse.Namespace) -> int:
console = Console()
cfg = load_config()
@@ -417,13 +496,26 @@ def _op_version(binary: Path) -> str:
return "version unknown"
def _op_whoami(binary: Path, account: str) -> Optional[str]:
"""Return a short identity string if op is authenticated, else None."""
def _op_whoami(
binary: Path, account: str, *, token_value: str = ""
) -> Optional[str]:
"""Return a short identity string if op is authenticated, else None.
``token_value``, when given, is passed to the child as
``OP_SERVICE_ACCOUNT_TOKEN`` so a candidate token can be probed
without touching the caller's environment.
"""
cmd = [str(binary), "whoami"]
if account:
cmd += ["--account", account]
env = os.environ.copy()
env.setdefault("NO_COLOR", "1")
if token_value:
env["OP_SERVICE_ACCOUNT_TOKEN"] = token_value
try:
res = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
res = subprocess.run(
cmd, env=env, capture_output=True, text=True, timeout=10
)
except (OSError, subprocess.TimeoutExpired):
return None
if res.returncode != 0:
+96
View File
@@ -71,6 +71,21 @@ def register_cli(parent_parser: argparse.ArgumentParser) -> None:
status = sub.add_parser("status", help="Show config + binary + last fetch")
status.set_defaults(func=cmd_status)
token = sub.add_parser(
"token",
help="Rotate the access token: validate a new one and store it in .env",
)
token.add_argument(
"--access-token",
help="Provide the new token non-interactively (default: masked prompt)",
)
token.add_argument(
"--no-verify",
action="store_true",
help="Store without probing Bitwarden first (not recommended)",
)
token.set_defaults(func=cmd_token)
sync = sub.add_parser("sync", help="Fetch secrets now and report what changed")
sync.add_argument(
"--apply",
@@ -337,6 +352,87 @@ def cmd_status(args: argparse.Namespace) -> int:
return 0
def cmd_token(args: argparse.Namespace) -> int:
"""Rotate the BSM access token without re-running the whole setup wizard.
Prompts for (or accepts via ``--access-token``) a new machine-account
token, probes Bitwarden with it (unless ``--no-verify``), and only then
persists it to .env — so a bad paste never bricks the working token.
"""
console = Console()
cfg = load_config()
bw_cfg = (cfg.get("secrets") or {}).get("bitwarden") or {}
token_env = bw_cfg.get("access_token_env", "BWS_ACCESS_TOKEN")
server_url = str(bw_cfg.get("server_url", "") or "").strip()
token = (args.access_token or "").strip()
if not token:
if not sys.stdin.isatty():
console.print(
"[red]No TTY — pass the token with --access-token.[/red]"
)
return 1
console.print(
"Create a new token in the Bitwarden web app:\n"
" Secrets Manager → Machine accounts → [your account] → "
"Access tokens → Create access token\n"
)
token = masked_secret_prompt(f"Paste new access token ({token_env}): ").strip()
if not token:
console.print("[red]Empty token, aborting.[/red]")
return 1
if not token.startswith("0."):
console.print(
"[yellow]Warning: token doesn't start with '0.' — usually that means "
"you pasted something other than a BSM access token.[/yellow]"
)
if not args.no_verify:
binary = bw.find_bws(install_if_missing=True)
if binary is None:
console.print(
"[red]bws binary not available — cannot verify. "
"Re-run with --no-verify to store anyway.[/red]"
)
return 1
console.print("Verifying against Bitwarden…")
projects = _list_projects(binary, token, console, server_url=server_url)
if projects is None:
console.print(
"[red]✗ New token was rejected — nothing was changed.[/red]"
)
return 1
console.print(
f"[green]✓ Token accepted[/green] "
f"({len(projects)} project{'s' if len(projects) != 1 else ''} visible)."
)
project_id = str(bw_cfg.get("project_id", "") or "")
if project_id and projects and project_id not in {p["id"] for p in projects}:
console.print(
f"[yellow]Warning: configured project {project_id} is not visible "
"to this machine account. Grant it access in the Bitwarden web "
"app or re-run `hermes secrets bitwarden setup` to pick a "
"different project.[/yellow]"
)
save_env_value(token_env, token)
os.environ[token_env] = token
# Old cached pulls are keyed on the previous token's fingerprint; drop
# them so the next startup fetches fresh with the new credential.
bw.clear_caches()
console.print(
f"[green]✓[/green] stored in {get_env_path()} as {token_env}. "
"Takes effect on the next Hermes invocation."
)
if not bw_cfg.get("enabled"):
console.print(
"[yellow]Note: the Bitwarden integration is currently disabled — "
"run `hermes secrets bitwarden setup` (or set "
"secrets.bitwarden.enabled: true) to turn it on.[/yellow]"
)
return 0
def cmd_sync(args: argparse.Namespace) -> int:
console = Console()
cfg = load_config()
@@ -0,0 +1,180 @@
"""Tests for `hermes secrets bitwarden token` / `hermes secrets onepassword token`.
The rotation command must: verify the candidate token BEFORE persisting,
never touch .env on a rejected token, store + clear caches on success,
and fail cleanly without a TTY.
"""
from __future__ import annotations
import argparse
from pathlib import Path
from unittest import mock
import pytest
from hermes_cli import onepassword_secrets_cli as op_cli
from hermes_cli import secrets_cli as bw_cli
# ---------------------------------------------------------------------------
# Bitwarden
# ---------------------------------------------------------------------------
def _bw_args(**overrides):
return argparse.Namespace(
access_token=overrides.get("access_token", ""),
no_verify=overrides.get("no_verify", False),
)
@pytest.fixture
def bw_env(monkeypatch, tmp_path):
saved = {}
monkeypatch.setattr(bw_cli, "load_config", lambda: {
"secrets": {"bitwarden": {
"enabled": True,
"access_token_env": "BWS_ACCESS_TOKEN",
"project_id": "proj-1",
"server_url": "",
}},
})
monkeypatch.setattr(
bw_cli, "save_env_value",
lambda name, value: saved.__setitem__(name, value),
)
monkeypatch.setattr(bw_cli, "get_env_path", lambda: tmp_path / ".env")
monkeypatch.setattr(
bw_cli.bw, "find_bws",
lambda install_if_missing=True: Path("/fake/bws"),
)
return saved
def test_bw_token_rejected_token_never_persisted(bw_env, monkeypatch):
monkeypatch.setattr(
bw_cli, "_list_projects",
lambda binary, token, console, server_url="": None, # probe fails
)
rc = bw_cli.cmd_token(_bw_args(access_token="0.bad"))
assert rc == 1
assert bw_env == {} # nothing written to .env
def test_bw_token_accepted_token_persisted_and_caches_cleared(bw_env, monkeypatch):
cleared = []
monkeypatch.setattr(
bw_cli, "_list_projects",
lambda binary, token, console, server_url="": [{"id": "proj-1"}],
)
monkeypatch.setattr(bw_cli.bw, "clear_caches", lambda *a, **kw: cleared.append(True))
rc = bw_cli.cmd_token(_bw_args(access_token="0.fresh"))
assert rc == 0
assert bw_env == {"BWS_ACCESS_TOKEN": "0.fresh"}
assert cleared
def test_bw_token_warns_when_project_not_visible(bw_env, monkeypatch, capsys):
monkeypatch.setattr(
bw_cli, "_list_projects",
lambda binary, token, console, server_url="": [{"id": "other-proj"}],
)
monkeypatch.setattr(bw_cli.bw, "clear_caches", lambda *a, **kw: None)
rc = bw_cli.cmd_token(_bw_args(access_token="0.fresh"))
assert rc == 0 # stored anyway — the token itself is valid
out = capsys.readouterr().out
assert "proj-1" in out and "not visible" in out
def test_bw_token_no_verify_skips_probe(bw_env, monkeypatch):
probe = mock.Mock()
monkeypatch.setattr(bw_cli, "_list_projects", probe)
monkeypatch.setattr(bw_cli.bw, "clear_caches", lambda *a, **kw: None)
rc = bw_cli.cmd_token(_bw_args(access_token="0.x", no_verify=True))
assert rc == 0
probe.assert_not_called()
assert bw_env == {"BWS_ACCESS_TOKEN": "0.x"}
def test_bw_token_non_tty_requires_flag(bw_env, monkeypatch):
monkeypatch.setattr("sys.stdin.isatty", lambda: False)
rc = bw_cli.cmd_token(_bw_args())
assert rc == 1
assert bw_env == {}
# ---------------------------------------------------------------------------
# 1Password
# ---------------------------------------------------------------------------
def _op_args(**overrides):
return argparse.Namespace(
token=overrides.get("token", ""),
no_verify=overrides.get("no_verify", False),
)
@pytest.fixture
def op_env(monkeypatch, tmp_path):
saved = {}
monkeypatch.setattr(op_cli, "load_config", lambda: {
"secrets": {"onepassword": {
"enabled": True,
"service_account_token_env": "OP_SERVICE_ACCOUNT_TOKEN",
}},
})
monkeypatch.setattr(
op_cli, "save_env_value",
lambda name, value: saved.__setitem__(name, value),
)
monkeypatch.setattr(op_cli, "get_env_path", lambda: tmp_path / ".env")
monkeypatch.setattr(
op_cli.op_src, "find_op", lambda binary_path="": Path("/fake/op")
)
return saved
def test_op_token_rejected_never_persisted(op_env, monkeypatch):
monkeypatch.setattr(
op_cli, "_op_whoami",
lambda binary, account, token_value="": None,
)
rc = op_cli.cmd_token(_op_args(token="ops_bad"))
assert rc == 1
assert op_env == {}
def test_op_token_accepted_persisted_and_caches_cleared(op_env, monkeypatch):
cleared = []
monkeypatch.setattr(
op_cli, "_op_whoami",
lambda binary, account, token_value="": "service-account test",
)
monkeypatch.setattr(
op_cli.op_src, "clear_caches", lambda *a, **kw: cleared.append(True)
)
rc = op_cli.cmd_token(_op_args(token="ops_fresh"))
assert rc == 0
assert op_env == {"OP_SERVICE_ACCOUNT_TOKEN": "ops_fresh"}
assert cleared
def test_op_token_probe_uses_candidate_token(op_env, monkeypatch):
seen = {}
def fake_whoami(binary, account, token_value=""):
seen["token"] = token_value
return "ok"
monkeypatch.setattr(op_cli, "_op_whoami", fake_whoami)
monkeypatch.setattr(op_cli.op_src, "clear_caches", lambda *a, **kw: None)
op_cli.cmd_token(_op_args(token="ops_candidate"))
assert seen["token"] == "ops_candidate"
def test_op_token_non_tty_requires_flag(op_env, monkeypatch):
monkeypatch.setattr("sys.stdin.isatty", lambda: False)
rc = op_cli.cmd_token(_op_args())
assert rc == 1
assert op_env == {}
@@ -0,0 +1,232 @@
"""Error remediation for secret sources.
Covers the ErrorKind classification of Bitwarden's `invalid_client`
identity reject, the bws stderr summarizer, the per-source
``remediation()`` hook, and the env_loader startup hint printer.
"""
from __future__ import annotations
from pathlib import Path
from unittest import mock
import pytest
from agent.secret_sources import bitwarden as bw
from agent.secret_sources import onepassword as op
from agent.secret_sources.base import ErrorKind, SecretSource
from agent.secret_sources.bitwarden import (
BitwardenSource,
_classify_bws_error,
_summarize_bws_stderr,
)
from agent.secret_sources.onepassword import OnePasswordSource
_BWS_INVALID_CLIENT_DUMP = """\
Error:
0: Received error message from server: [400 Bad Request] {"error":"invalid_client"}
Location:
crates/bws/src/main.rs:108
Backtrace omitted. Run with RUST_BACKTRACE=1 environment variable to display it.
Run with RUST_BACKTRACE=full to include source snippets.
"""
# ---------------------------------------------------------------------------
# _summarize_bws_stderr
# ---------------------------------------------------------------------------
def test_summarize_strips_rust_report_noise():
summary = _summarize_bws_stderr(_BWS_INVALID_CLIENT_DUMP)
assert "invalid_client" in summary
assert "Location:" not in summary
assert "main.rs" not in summary
assert "Backtrace" not in summary
assert "Error:" not in summary
def test_summarize_joins_multiple_cause_lines():
raw = "Error:\n 0: outer cause\n 1: inner cause\n\nLocation:\n x.rs:1"
assert _summarize_bws_stderr(raw) == "outer cause; inner cause"
def test_summarize_falls_back_to_raw_on_unknown_shape():
assert _summarize_bws_stderr("plain failure text") == "plain failure text"
assert _summarize_bws_stderr("") == ""
# ---------------------------------------------------------------------------
# _classify_bws_error — the invalid_client identity reject is an auth failure
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("message", [
'bws exited 1: Received error message from server: [400 Bad Request] {"error":"invalid_client"}',
"invalid_grant returned by identity",
"server said 401 unauthorized",
])
def test_classify_auth_failures(message):
assert _classify_bws_error(message) == ErrorKind.AUTH_FAILED
def test_classify_unknown_stays_internal():
assert _classify_bws_error("some novel explosion") == ErrorKind.INTERNAL
# ---------------------------------------------------------------------------
# BitwardenSource.fetch — auth failures get a human explanation
# ---------------------------------------------------------------------------
def test_fetch_auth_failure_gets_friendly_error(monkeypatch, tmp_path):
src = BitwardenSource()
monkeypatch.setenv("BWS_ACCESS_TOKEN", "0.dead")
monkeypatch.setattr(bw, "find_bws", lambda install_if_missing=True: tmp_path / "bws")
def boom(**kwargs):
raise RuntimeError(
'bws exited 1: Received error message from server: '
'[400 Bad Request] {"error":"invalid_client"}'
)
monkeypatch.setattr(bw, "fetch_bitwarden_secrets", boom)
result = src.fetch({"enabled": True, "project_id": "p"}, tmp_path)
assert result.error_kind == ErrorKind.AUTH_FAILED
assert "revoked, expired" in result.error
assert "BWS_ACCESS_TOKEN" in result.error
assert "invalid_client" in result.error # mechanics preserved
# ---------------------------------------------------------------------------
# remediation() hook
# ---------------------------------------------------------------------------
def test_bitwarden_auth_remediation_points_at_token_command():
hint = BitwardenSource().remediation(ErrorKind.AUTH_FAILED, {})
assert "hermes secrets bitwarden token" in hint
def test_onepassword_auth_remediation_points_at_token_command():
hint = OnePasswordSource().remediation(ErrorKind.AUTH_FAILED, {})
assert "hermes secrets onepassword token" in hint
assert "OP_SERVICE_ACCOUNT_TOKEN" in hint
def test_onepassword_remediation_uses_configured_token_env():
hint = OnePasswordSource().remediation(
ErrorKind.AUTH_FAILED, {"service_account_token_env": "MY_OP_TOKEN"}
)
assert "MY_OP_TOKEN" in hint
def test_base_remediation_covers_common_kinds():
class _Src(SecretSource):
name = "dummy"
label = "Dummy"
def fetch(self, cfg, home_path): # pragma: no cover
raise NotImplementedError
src = _Src()
for kind in (ErrorKind.NOT_CONFIGURED, ErrorKind.BINARY_MISSING,
ErrorKind.AUTH_FAILED, ErrorKind.AUTH_EXPIRED,
ErrorKind.NETWORK, ErrorKind.TIMEOUT):
hint = src.remediation(kind, {})
assert hint, f"no default hint for {kind}"
if kind in (ErrorKind.NOT_CONFIGURED, ErrorKind.BINARY_MISSING,
ErrorKind.AUTH_FAILED, ErrorKind.AUTH_EXPIRED):
assert "hermes secrets dummy" in hint
# Kinds without a sensible generic action stay silent.
assert _Src().remediation(ErrorKind.INTERNAL, {}) == ""
assert _Src().remediation(None, {}) == ""
def test_remediation_never_raises_on_junk_cfg():
for cfg in (None, [], "nope", 42):
assert isinstance(BitwardenSource().remediation(ErrorKind.AUTH_FAILED, cfg), str)
assert isinstance(OnePasswordSource().remediation(ErrorKind.AUTH_FAILED, cfg), str)
# ---------------------------------------------------------------------------
# env_loader startup hint
# ---------------------------------------------------------------------------
def test_env_loader_prints_remediation_hint(tmp_path, monkeypatch, capsys):
from hermes_cli import env_loader
from agent.secret_sources import registry
registry._reset_registry_for_tests()
env_loader.reset_secret_source_cache()
home = tmp_path / ".hermes"
home.mkdir()
(home / "config.yaml").write_text(
"secrets:\n"
" bitwarden:\n"
" enabled: true\n"
" project_id: proj\n"
)
monkeypatch.setenv("BWS_ACCESS_TOKEN", "0.dead")
monkeypatch.setattr(bw, "find_bws", lambda install_if_missing=True: tmp_path / "bws")
def boom(**kwargs):
raise RuntimeError(
'bws exited 1: Received error message from server: '
'[400 Bad Request] {"error":"invalid_client"}'
)
monkeypatch.setattr(bw, "fetch_bitwarden_secrets", boom)
try:
env_loader._apply_external_secret_sources(home)
finally:
registry._reset_registry_for_tests()
env_loader.reset_secret_source_cache()
err = capsys.readouterr().err
assert "rejected the machine-account access token" in err
assert "hermes secrets bitwarden token" in err
def test_env_loader_hint_survives_broken_remediation(tmp_path, monkeypatch, capsys):
"""A plugin source whose remediation() raises must not break startup."""
from hermes_cli import env_loader
from agent.secret_sources import registry
class _Broken(SecretSource):
name = "brokensrc"
label = "Broken"
shape = "bulk"
def fetch(self, cfg, home_path):
from agent.secret_sources.base import FetchResult
res = FetchResult()
res.error = "kaput"
res.error_kind = ErrorKind.AUTH_FAILED
return res
def remediation(self, kind, cfg):
raise RuntimeError("hint machine broke")
registry._reset_registry_for_tests()
registry._BUILTINS_LOADED = True # keep real builtins out of this test
registry.register_source(_Broken())
env_loader.reset_secret_source_cache()
home = tmp_path / ".hermes"
home.mkdir()
(home / "config.yaml").write_text(
"secrets:\n brokensrc:\n enabled: true\n"
)
try:
env_loader._apply_external_secret_sources(home)
finally:
registry._reset_registry_for_tests()
env_loader.reset_secret_source_cache()
err = capsys.readouterr().err
assert "kaput" in err # error still surfaced, no crash
@@ -110,6 +110,7 @@ class MyVaultSource(SecretSource):
| `protected_env_vars(cfg)` | empty | You have a bootstrap token (you almost certainly do) |
| `fetch_timeout_seconds(cfg)` | 120s | Your backend needs a different budget |
| `config_schema()` | `{}` | Declare config keys for setup surfaces |
| `remediation(kind, cfg)` | generic per-`ErrorKind` hints | You want failure warnings to point at your own fix-it command (e.g. the bundled sources return `Run hermes secrets <name> token…` for `AUTH_FAILED`). Must be a pure kind→string mapping: no I/O, never raises. Return `""` to suppress the hint. |
## Subprocess safety: use `run_secret_cli()`
+1
View File
@@ -432,6 +432,7 @@ Pull API keys from an external secret manager at process startup instead of stor
|------------|-------------|
| `setup` | Interactive wizard: install the pinned `bws` binary, store an access token, and pick a project. Accepts `--project-id`, `--access-token`, and `--server-url` for non-interactive use. |
| `status` | Show current config, binary path/version, and last fetch info. |
| `token` | Rotate the access token: validates the new token against Bitwarden before storing it in `.env` (a rejected token changes nothing). Accepts `--access-token` for non-interactive use and `--no-verify` to skip the probe. |
| `sync` | Fetch secrets now and report what changed. Add `--apply` to actually export the secrets into the current shell's environment (default is dry-run). |
| `install` | Download and verify the pinned `bws` binary. `--force` re-downloads even if a managed copy already exists. |
| `disable` | Turn off the Bitwarden integration. |
+23 -2
View File
@@ -69,11 +69,30 @@ From now on, every `hermes` invocation pulls fresh secrets at startup. You'll se
|---|---|
| `hermes secrets bitwarden setup` | Interactive wizard (install binary, prompt for token, pick project, test fetch) |
| `hermes secrets bitwarden status` | Show config + binary version + token presence |
| `hermes secrets bitwarden token` | Rotate the access token: validate the new token against Bitwarden, then store it in `.env` |
| `hermes secrets bitwarden sync` | Dry-run: pull secrets now and show what would be applied |
| `hermes secrets bitwarden sync --apply` | Pull and export into the current shell's environment |
| `hermes secrets bitwarden install` | Just download the pinned `bws` binary (no auth required) |
| `hermes secrets bitwarden disable` | Flip `enabled: false`; leaves token + project id in place |
## Rotating an expired or revoked token
When the machine-account token expires, gets revoked, or the account is deleted, startup shows:
```
Bitwarden Secrets Manager: Bitwarden rejected the machine-account access token (BWS_ACCESS_TOKEN) — it was likely revoked, expired, or belongs to another region. (...)
Bitwarden Secrets Manager: → Run `hermes secrets bitwarden token` to paste a fresh access token ...
```
Fix it without re-running the whole wizard:
```bash
hermes secrets bitwarden token # masked prompt
hermes secrets bitwarden token --access-token 0.… # non-interactive
```
The command probes Bitwarden with the new token **before** writing anything — a rejected token leaves your current `.env` untouched. On success it stores the token, clears the fetch caches, and warns if the configured project is not visible to the new machine account.
## Configuration
Defaults in `~/.hermes/config.yaml`:
@@ -107,12 +126,14 @@ Bitwarden never blocks Hermes startup. If anything goes wrong, you'll see a one-
| Symptom | Cause | Fix |
|---|---|---|
| `BWS_ACCESS_TOKEN is not set` | Enabled in config but token cleared from `.env` | Re-run `hermes secrets bitwarden setup` |
| `bws exited 1: invalid access token` | Token revoked or wrong | Generate a new token, re-run setup |
| `[400 Bad Request] {"error":"invalid_client"}` | Token is for a Bitwarden region other than the one `bws` is calling (e.g. EU token hitting the US identity endpoint) | Re-run setup and pick the right region, or set `secrets.bitwarden.server_url` to `https://vault.bitwarden.eu` (or your self-hosted URL) |
| `Bitwarden rejected the machine-account access token … invalid_client` | Token revoked, expired, machine account deleted — or the token belongs to another region (e.g. EU token hitting the US identity endpoint) | Run `hermes secrets bitwarden token` to paste a fresh token; for region mismatches re-run setup and pick EU/self-hosted (or set `secrets.bitwarden.server_url`) |
| `bws exited 1: invalid access token` | Token revoked or wrong | Run `hermes secrets bitwarden token` with a new token |
| `bws timed out` | Network blocked or Bitwarden API slow | Check connectivity to `api.bitwarden.com` (or your `server_url`) |
| `bws binary not available` | `auto_install: false` and `bws` not on PATH | Install manually from [github.com/bitwarden/sdk-sm/releases](https://github.com/bitwarden/sdk-sm/releases) or flip `auto_install` back on |
| `Checksum mismatch` | Download corrupted or tampered | Re-run, will retry; if it persists, file an issue |
Startup warnings now include a `→` remediation line telling you exactly which command fixes the failure.
## Security notes
- The bootstrap token (`BWS_ACCESS_TOKEN`) is itself sensitive — anyone with it can read every secret the machine account has access to. Treat it the same as any other API key.
@@ -93,6 +93,7 @@ From now on, every `hermes` invocation resolves the references at startup. You'l
|---|---|
| `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 token` | Rotate the service-account token: validate with `op whoami`, then store it in `.env` |
| `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 |
@@ -136,11 +137,13 @@ secrets:
| 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 failed for 'op://…': …` | Locked session, expired token, or no vault access | `op signin`, run `hermes secrets onepassword token` to rotate the service-account 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 |
Startup warnings now include a `→` remediation line telling you exactly which command fixes the failure.
## 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:
@@ -69,11 +69,23 @@ hermes secrets bitwarden status
|---|---|
| `hermes secrets bitwarden setup` | 交互式向导(安装二进制文件、提示输入令牌、选择项目、测试拉取) |
| `hermes secrets bitwarden status` | 显示配置、二进制版本及令牌是否存在 |
| `hermes secrets bitwarden token` | 轮换访问令牌:先向 Bitwarden 验证新令牌,验证通过后再写入 `.env` |
| `hermes secrets bitwarden sync` | 演习模式:立即拉取 secret 并显示将应用的内容 |
| `hermes secrets bitwarden sync --apply` | 拉取并导出到当前 shell 的环境中 |
| `hermes secrets bitwarden install` | 仅下载固定版本的 `bws` 二进制文件(无需认证) |
| `hermes secrets bitwarden disable` | 将 `enabled` 设为 `false`;保留令牌和项目 ID |
## 轮换已过期或已吊销的令牌
当机器账户令牌过期、被吊销或账户被删除时,启动信息会显示令牌被拒绝的说明,并附带 `→` 修复提示。无需重新运行整个向导即可修复:
```bash
hermes secrets bitwarden token # 隐藏输入提示
hermes secrets bitwarden token --access-token 0.… # 非交互式
```
该命令会在写入任何内容**之前**用新令牌探测 Bitwarden——令牌被拒绝时不会改动现有 `.env`。成功后会存储令牌、清除拉取缓存,并在配置的项目对新机器账户不可见时发出警告。
## 配置
`~/.hermes/config.yaml` 中的默认值:
@@ -107,12 +119,14 @@ Bitwarden 永远不会阻塞 Hermes 启动。如果出现任何问题,stderr
| 现象 | 原因 | 修复方法 |
|---|---|---|
| `BWS_ACCESS_TOKEN is not set` | 配置中已启用,但令牌已从 `.env` 中清除 | 重新运行 `hermes secrets bitwarden setup` |
| `bws exited 1: invalid access token` | 令牌已吊销或有误 | 生成新令牌,重新运行 setup |
| `[400 Bad Request] {"error":"invalid_client"}` | 令牌所属的 Bitwarden 区域与 `bws` 调用的区域不匹配(例如欧盟令牌访问了美国 identity 端点) | 重新运行 setup 并选择正确区域,或将 `secrets.bitwarden.server_url` 设为 `https://vault.bitwarden.eu`(或自托管 URL) |
| `Bitwarden rejected the machine-account access token … invalid_client` | 令牌已吊销、过期、机器账户被删除——或令牌属于其他区域(例如欧盟令牌访问了美国 identity 端点) | 运行 `hermes secrets bitwarden token` 粘贴新令牌;区域不匹配时重新运行 setup 选择欧盟/自托管(或设置 `secrets.bitwarden.server_url`) |
| `bws exited 1: invalid access token` | 令牌已吊销或有误 | 运行 `hermes secrets bitwarden token` 提供新令牌 |
| `bws timed out` | 网络受阻或 Bitwarden API 响应缓慢 | 检查到 `api.bitwarden.com`(或你的 `server_url`)的连通性 |
| `bws binary not available` | `auto_install: false` 且 `bws` 不在 PATH 中 | 从 [github.com/bitwarden/sdk-sm/releases](https://github.com/bitwarden/sdk-sm/releases) 手动安装,或重新开启 `auto_install` |
| `Checksum mismatch` | 下载内容损坏或被篡改 | 重新运行,将自动重试;如持续出现,请提交 issue |
启动警告现在会附带一行 `→` 修复提示,直接告诉你运行哪条命令即可修复。
## 安全说明
- 引导令牌(`BWS_ACCESS_TOKEN`)本身是敏感信息——任何持有它的人都可以读取机器账户有权访问的所有 secret。请与其他 API 密钥同等对待。