mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-05-29 06:31:32 +00:00
feat(secrets): Bitwarden Secrets Manager integration with lazy bws install (#30035)
* feat(secrets): Bitwarden Secrets Manager integration with lazy bws install
Pull API keys from Bitwarden Secrets Manager at process startup
instead of storing them all in plaintext in ~/.hermes/.env. One
bootstrap token (BWS_ACCESS_TOKEN) replaces N per-provider keys, and
rotating a credential becomes a single change in the Bitwarden web
app.
Bitwarden defaults to source of truth: secrets pulled from BSM
overwrite any matching env vars on startup so rotations actually
take effect. Set secrets.bitwarden.override_existing: false in
config.yaml to invert.
The bws binary is auto-downloaded into ~/.hermes/bin/bws on first
use (pinned to v2.0.0, SHA-256 verified against the GitHub release
checksum file). No apt, brew, or sudo required.
New surfaces:
hermes secrets bitwarden setup — interactive wizard
hermes secrets bitwarden status — config + binary + token state
hermes secrets bitwarden sync — dry-run fetch / --apply exports
hermes secrets bitwarden disable — flip enabled: false
hermes secrets bitwarden install — just download the binary
Failures (missing binary, bad token, no network) never block Hermes
startup — they emit a one-line warning to stderr and continue with
whatever credentials .env already had.
Docs: website/docs/user-guide/secrets/{index,bitwarden}.md
Tests: tests/test_bitwarden_secrets.py (26 tests, hermetic — bws
subprocess and HTTP downloads fully mocked)
* chore(infographic): add bitwarden-secrets-manager bento-grid retro-pop-grid
Generated for PR #30035 — Bitwarden Secrets Manager integration.
Style picked via pick_pr_infographic_style.py rotation:
layout: bento-grid
style: retro-pop-grid
aspect: 1:1 square
Saved at infographic/bitwarden-secrets-manager/infographic.png
This commit is contained in:
parent
18cd1e5c72
commit
552e9c7881
13 changed files with 1921 additions and 1 deletions
445
hermes_cli/secrets_cli.py
Normal file
445
hermes_cli/secrets_cli.py
Normal file
|
|
@ -0,0 +1,445 @@
|
|||
"""CLI handlers for ``hermes secrets bitwarden ...``.
|
||||
|
||||
Subcommands:
|
||||
setup — interactive wizard: install bws, prompt for token + project, test fetch
|
||||
status — show current config + binary version + last fetch outcome
|
||||
sync — run a fetch right now and show what would be applied (dry-run friendly)
|
||||
disable — flip ``secrets.bitwarden.enabled`` to False
|
||||
install — just download the bws binary (no token / project required)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import getpass
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import List, Optional, Tuple
|
||||
|
||||
from rich.console import Console
|
||||
from rich.panel import Panel
|
||||
from rich.table import Table
|
||||
|
||||
from agent.secret_sources import bitwarden as bw
|
||||
from hermes_cli.config import (
|
||||
get_env_path,
|
||||
load_config,
|
||||
save_config,
|
||||
save_env_value,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Argparse wiring — called from hermes_cli.main
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def register_cli(parent_parser: argparse.ArgumentParser) -> None:
|
||||
"""Attach the ``bitwarden`` subcommand tree to a parent parser.
|
||||
|
||||
Called from ``hermes_cli.main`` as part of building the top-level
|
||||
``hermes secrets`` parser.
|
||||
"""
|
||||
sub = parent_parser.add_subparsers(dest="secrets_bw_command")
|
||||
|
||||
setup = sub.add_parser(
|
||||
"setup",
|
||||
help="Interactive wizard: install bws, store access token, pick project",
|
||||
)
|
||||
setup.add_argument(
|
||||
"--project-id",
|
||||
help="Pre-select a project UUID instead of prompting",
|
||||
)
|
||||
setup.add_argument(
|
||||
"--access-token",
|
||||
help="Provide the access token non-interactively (will be stored in .env)",
|
||||
)
|
||||
setup.set_defaults(func=cmd_setup)
|
||||
|
||||
status = sub.add_parser("status", help="Show config + binary + last fetch")
|
||||
status.set_defaults(func=cmd_status)
|
||||
|
||||
sync = sub.add_parser("sync", help="Fetch secrets now and report what changed")
|
||||
sync.add_argument(
|
||||
"--apply",
|
||||
action="store_true",
|
||||
help="Actually export the secrets into the current shell's env (default: dry-run)",
|
||||
)
|
||||
sync.set_defaults(func=cmd_sync)
|
||||
|
||||
disable = sub.add_parser("disable", help="Turn off the Bitwarden integration")
|
||||
disable.set_defaults(func=cmd_disable)
|
||||
|
||||
install = sub.add_parser(
|
||||
"install",
|
||||
help=f"Download and verify the pinned bws binary (v{bw._BWS_VERSION})",
|
||||
)
|
||||
install.add_argument(
|
||||
"--force",
|
||||
action="store_true",
|
||||
help="Re-download even if a managed copy already exists",
|
||||
)
|
||||
install.set_defaults(func=cmd_install)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Handlers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def cmd_setup(args: argparse.Namespace) -> int:
|
||||
console = Console()
|
||||
console.print(
|
||||
Panel.fit(
|
||||
"[bold]Bitwarden Secrets Manager setup[/bold]\n\n"
|
||||
"Need an access token? In the Bitwarden web app:\n"
|
||||
" Secrets Manager → Machine accounts → [your account] →\n"
|
||||
" Access tokens → Create access token\n\n"
|
||||
"Copy the token (starts with [cyan]0.[/cyan]…) — it cannot be retrieved later.",
|
||||
border_style="cyan",
|
||||
)
|
||||
)
|
||||
|
||||
# ------------------------------------------------------------------ binary
|
||||
console.print()
|
||||
console.print("[bold]Step 1[/bold] Install the bws CLI")
|
||||
try:
|
||||
binary = bw.find_bws(install_if_missing=False)
|
||||
if binary is None:
|
||||
console.print(" No bws on PATH — downloading…")
|
||||
binary = bw.install_bws()
|
||||
version = _bws_version(binary)
|
||||
console.print(f" [green]✓[/green] {binary} ({version})")
|
||||
except Exception as exc: # noqa: BLE001
|
||||
console.print(f" [red]✗ Could not install bws: {exc}[/red]")
|
||||
console.print(
|
||||
" Manual install: "
|
||||
"https://github.com/bitwarden/sdk-sm/releases"
|
||||
)
|
||||
return 1
|
||||
|
||||
# ------------------------------------------------------------------- token
|
||||
console.print()
|
||||
console.print("[bold]Step 2[/bold] Provide your access token")
|
||||
cfg = load_config()
|
||||
secrets_cfg = (cfg.setdefault("secrets", {})
|
||||
.setdefault("bitwarden", {}))
|
||||
token_env = secrets_cfg.get("access_token_env", "BWS_ACCESS_TOKEN")
|
||||
|
||||
token = (args.access_token or "").strip()
|
||||
if not token:
|
||||
token = getpass.getpass(f" Paste 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. Continuing anyway.[/yellow]"
|
||||
)
|
||||
|
||||
save_env_value(token_env, token)
|
||||
os.environ[token_env] = token # so the test fetch below sees it
|
||||
console.print(f" [green]✓[/green] stored in {get_env_path()} as {token_env}")
|
||||
|
||||
# ------------------------------------------------------------------- project
|
||||
if args.project_id and args.project_id.strip():
|
||||
project_id = args.project_id.strip()
|
||||
else:
|
||||
console.print()
|
||||
console.print("[bold]Step 3[/bold] Pick a project")
|
||||
project_id = ""
|
||||
projects = _list_projects(binary, token, console)
|
||||
if projects is None:
|
||||
return 1
|
||||
if not projects:
|
||||
console.print(" [yellow]No projects visible to this machine account.[/yellow]")
|
||||
console.print(
|
||||
" In the Bitwarden web app, open the machine account → Projects tab "
|
||||
"and grant it access to at least one project."
|
||||
)
|
||||
return 1
|
||||
|
||||
table = Table(show_header=True, header_style="bold")
|
||||
table.add_column("#", style="cyan", width=4)
|
||||
table.add_column("Name")
|
||||
table.add_column("ID", style="dim")
|
||||
for i, p in enumerate(projects, 1):
|
||||
table.add_row(str(i), p.get("name", "?"), p.get("id", "?"))
|
||||
console.print(table)
|
||||
|
||||
while True:
|
||||
choice = console.input(f" Select project [1-{len(projects)}]: ").strip()
|
||||
if not choice:
|
||||
continue
|
||||
try:
|
||||
idx = int(choice)
|
||||
except ValueError:
|
||||
console.print(" [red]Enter a number.[/red]")
|
||||
continue
|
||||
if 1 <= idx <= len(projects):
|
||||
project_id = projects[idx - 1]["id"]
|
||||
break
|
||||
console.print(f" [red]Out of range — pick 1-{len(projects)}.[/red]")
|
||||
|
||||
# ------------------------------------------------------------------- test
|
||||
console.print()
|
||||
step_num = 4 if not (args.project_id and args.project_id.strip()) else 3
|
||||
console.print(f"[bold]Step {step_num}[/bold] Test fetch")
|
||||
try:
|
||||
secrets, warnings = bw.fetch_bitwarden_secrets(
|
||||
access_token=token,
|
||||
project_id=project_id,
|
||||
binary=binary,
|
||||
use_cache=False,
|
||||
)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
console.print(f" [red]✗ Fetch failed: {exc}[/red]")
|
||||
return 1
|
||||
|
||||
if not secrets:
|
||||
console.print(" [yellow]Fetch succeeded but the project has no secrets.[/yellow]")
|
||||
else:
|
||||
table = Table(show_header=True, header_style="bold")
|
||||
table.add_column("Name", style="cyan")
|
||||
table.add_column("Status")
|
||||
for key in sorted(secrets):
|
||||
if key == token_env:
|
||||
status = "[dim]bootstrap token — never overrides itself[/dim]"
|
||||
elif os.environ.get(key):
|
||||
status = "[yellow]already set in env (will be overwritten)[/yellow]"
|
||||
else:
|
||||
status = "[green]new[/green]"
|
||||
table.add_row(key, status)
|
||||
console.print(table)
|
||||
for w in warnings:
|
||||
console.print(f" [yellow]warning:[/yellow] {w}")
|
||||
|
||||
# ------------------------------------------------------------------- save
|
||||
secrets_cfg["enabled"] = True
|
||||
secrets_cfg["project_id"] = project_id
|
||||
secrets_cfg.setdefault("access_token_env", token_env)
|
||||
secrets_cfg.setdefault("cache_ttl_seconds", 300)
|
||||
secrets_cfg.setdefault("override_existing", True)
|
||||
secrets_cfg.setdefault("auto_install", True)
|
||||
save_config(cfg)
|
||||
|
||||
console.print()
|
||||
console.print(
|
||||
"[green]✓ Bitwarden Secrets Manager is enabled.[/green] "
|
||||
"Secrets will be pulled at the start of every Hermes process."
|
||||
)
|
||||
console.print(
|
||||
" Status: [cyan]hermes secrets bitwarden status[/cyan]\n"
|
||||
" Refresh: [cyan]hermes secrets bitwarden sync[/cyan]\n"
|
||||
" Disable: [cyan]hermes secrets bitwarden disable[/cyan]"
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_status(args: argparse.Namespace) -> int:
|
||||
console = Console()
|
||||
cfg = load_config()
|
||||
bw_cfg = (cfg.get("secrets") or {}).get("bitwarden") or {}
|
||||
|
||||
enabled = bool(bw_cfg.get("enabled"))
|
||||
token_env = bw_cfg.get("access_token_env", "BWS_ACCESS_TOKEN")
|
||||
project_id = bw_cfg.get("project_id", "")
|
||||
token_set = bool(os.environ.get(token_env))
|
||||
|
||||
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("Token env var", token_env)
|
||||
table.add_row("Token in env", _yn(token_set))
|
||||
table.add_row("Project ID", project_id or "[dim](unset)[/dim]")
|
||||
table.add_row("Override existing", _yn(bool(bw_cfg.get("override_existing", False))))
|
||||
table.add_row("Cache TTL (s)", str(bw_cfg.get("cache_ttl_seconds", 300)))
|
||||
table.add_row("Auto-install", _yn(bool(bw_cfg.get("auto_install", True))))
|
||||
|
||||
binary = bw.find_bws(install_if_missing=False)
|
||||
if binary:
|
||||
table.add_row("bws binary", f"{binary} ({_bws_version(binary)})")
|
||||
else:
|
||||
table.add_row("bws binary", "[yellow]not installed[/yellow]")
|
||||
|
||||
console.print(Panel(table, title="Bitwarden Secrets Manager", border_style="cyan"))
|
||||
|
||||
if not enabled:
|
||||
console.print("\n Run [cyan]hermes secrets bitwarden setup[/cyan] to enable.")
|
||||
return 0
|
||||
if not token_set:
|
||||
console.print(
|
||||
f"\n [yellow]Enabled but {token_env} is not set — Hermes will skip BSM "
|
||||
"and warn on next startup.[/yellow]"
|
||||
)
|
||||
if not project_id:
|
||||
console.print(
|
||||
"\n [yellow]Enabled but no project_id — nothing to fetch.[/yellow]"
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_sync(args: argparse.Namespace) -> int:
|
||||
console = Console()
|
||||
cfg = load_config()
|
||||
bw_cfg = (cfg.get("secrets") or {}).get("bitwarden") or {}
|
||||
if not bw_cfg.get("enabled"):
|
||||
console.print(
|
||||
"[yellow]Bitwarden integration is disabled. Run "
|
||||
"`hermes secrets bitwarden setup` first.[/yellow]"
|
||||
)
|
||||
return 1
|
||||
|
||||
token_env = bw_cfg.get("access_token_env", "BWS_ACCESS_TOKEN")
|
||||
token = os.environ.get(token_env, "").strip()
|
||||
if not token:
|
||||
console.print(f"[red]{token_env} is not set.[/red]")
|
||||
return 1
|
||||
|
||||
project_id = bw_cfg.get("project_id", "")
|
||||
if not project_id:
|
||||
console.print("[red]No project_id configured.[/red]")
|
||||
return 1
|
||||
|
||||
try:
|
||||
secrets, warnings = bw.fetch_bitwarden_secrets(
|
||||
access_token=token,
|
||||
project_id=project_id,
|
||||
use_cache=False,
|
||||
)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
console.print(f"[red]Fetch failed: {exc}[/red]")
|
||||
return 1
|
||||
|
||||
if not secrets:
|
||||
console.print("[yellow]No secrets in project.[/yellow]")
|
||||
return 0
|
||||
|
||||
override = bool(bw_cfg.get("override_existing", False)) or args.apply
|
||||
table = Table(show_header=True, header_style="bold")
|
||||
table.add_column("Name", style="cyan")
|
||||
table.add_column("Action")
|
||||
applied = 0
|
||||
for key in sorted(secrets):
|
||||
if key == token_env:
|
||||
table.add_row(key, "[dim]skip (bootstrap token)[/dim]")
|
||||
continue
|
||||
already = bool(os.environ.get(key))
|
||||
if already and not override:
|
||||
table.add_row(key, "[dim]skip (already set)[/dim]")
|
||||
continue
|
||||
if args.apply:
|
||||
os.environ[key] = secrets[key]
|
||||
applied += 1
|
||||
table.add_row(key, "[green]exported[/green]" + (" (overrode)" if already else ""))
|
||||
else:
|
||||
table.add_row(key, "[green]would export[/green]" + (" (overrides)" if already else ""))
|
||||
|
||||
console.print(table)
|
||||
for w in warnings:
|
||||
console.print(f"[yellow]warning:[/yellow] {w}")
|
||||
|
||||
if not args.apply:
|
||||
console.print(
|
||||
"\n This was a dry-run — secrets are picked up automatically on the "
|
||||
"next [cyan]hermes[/cyan] invocation. Re-run with [cyan]--apply[/cyan] "
|
||||
"to export into the current shell instead."
|
||||
)
|
||||
else:
|
||||
console.print(f"\n [green]Exported {applied} secret(s) into current process.[/green]")
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_disable(args: argparse.Namespace) -> int:
|
||||
console = Console()
|
||||
cfg = load_config()
|
||||
bw_cfg = (cfg.setdefault("secrets", {})
|
||||
.setdefault("bitwarden", {}))
|
||||
bw_cfg["enabled"] = False
|
||||
save_config(cfg)
|
||||
console.print(
|
||||
"[green]Disabled.[/green] Bitwarden secrets will NOT be pulled on the next "
|
||||
"Hermes invocation.\n"
|
||||
" Your access token is left in .env — remove it manually if you also want "
|
||||
"to revoke the credential."
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_install(args: argparse.Namespace) -> int:
|
||||
console = Console()
|
||||
try:
|
||||
path = bw.install_bws(force=bool(args.force))
|
||||
console.print(f"[green]✓[/green] {path} ({_bws_version(path)})")
|
||||
return 0
|
||||
except Exception as exc: # noqa: BLE001
|
||||
console.print(f"[red]Install failed: {exc}[/red]")
|
||||
return 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _yn(b: bool) -> str:
|
||||
return "[green]yes[/green]" if b else "[dim]no[/dim]"
|
||||
|
||||
|
||||
def _bws_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 _list_projects(
|
||||
binary: Path, token: str, console: Console
|
||||
) -> Optional[List[dict]]:
|
||||
"""Call ``bws project list`` and return the parsed list, or None on failure."""
|
||||
env = os.environ.copy()
|
||||
env["BWS_ACCESS_TOKEN"] = token
|
||||
env.setdefault("NO_COLOR", "1")
|
||||
try:
|
||||
res = subprocess.run(
|
||||
[str(binary), "project", "list", "--output", "json"],
|
||||
env=env,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=15,
|
||||
)
|
||||
except (OSError, subprocess.TimeoutExpired) as exc:
|
||||
console.print(f" [red]Couldn't list projects: {exc}[/red]")
|
||||
return None
|
||||
|
||||
if res.returncode != 0:
|
||||
err = (res.stderr or res.stdout).strip()[:300]
|
||||
console.print(f" [red]bws project list failed: {err}[/red]")
|
||||
if "authorization" in err.lower() or "invalid" in err.lower():
|
||||
console.print(
|
||||
" [yellow]This usually means the access token is wrong or revoked. "
|
||||
"Double-check it in the Bitwarden web app.[/yellow]"
|
||||
)
|
||||
return None
|
||||
|
||||
try:
|
||||
data = json.loads(res.stdout or "[]")
|
||||
except json.JSONDecodeError as exc:
|
||||
console.print(f" [red]bws returned non-JSON: {exc}[/red]")
|
||||
return None
|
||||
if not isinstance(data, list):
|
||||
return []
|
||||
return [p for p in data if isinstance(p, dict) and p.get("id")]
|
||||
Loading…
Add table
Add a link
Reference in a new issue