diff --git a/agent/vault_backends/__init__.py b/agent/vault_backends/__init__.py new file mode 100644 index 0000000000..f2980cee4b --- /dev/null +++ b/agent/vault_backends/__init__.py @@ -0,0 +1,18 @@ +"""Login backends for the browser credential vault. + +The local Fernet store (``agent/vault_store.py``) is one backend; 1Password +(``op``) and Bitwarden Password Manager (``bw``) are the others. Every backend +hands the agent the same shape — opaque handle + login metadata — and resolves +the password server-side at fill time only. Handles are namespaced by backend +(``vault_…`` local, ``op:…``, ``bw:…``) so the browser tools need no schema +change to route to the right one. + +External managers are locked until the user unlocks them for the current +session (``agent/vault_backends/unlock.py``); the master password is typed +into a masked prompt owned by the surface (CLI panel, Desktop dialog) and is +never a tool argument, never argv, never persisted. +""" + +from agent.vault_backends.base import LoginBackend, UnlockRequired, backend_for_handle, enabled_backends + +__all__ = ["LoginBackend", "UnlockRequired", "backend_for_handle", "enabled_backends"] diff --git a/agent/vault_backends/base.py b/agent/vault_backends/base.py new file mode 100644 index 0000000000..25335c87e7 --- /dev/null +++ b/agent/vault_backends/base.py @@ -0,0 +1,86 @@ +"""Login-backend contract + registry for the browser credential vault. + +A ``LoginBackend`` lists login metadata (never secrets) and resolves ONE +password at fill time. External managers (1Password, Bitwarden) additionally +need a per-session unlock; ``resolve_password`` raises ``UnlockRequired`` +while locked so the tool can ask the surface to prompt. Handles are +namespaced by ``prefix`` so ``backend_for_handle`` needs no lookup table. +""" + +from __future__ import annotations + +import subprocess +from abc import ABC, abstractmethod +from typing import Dict, List, Optional, Sequence + +from agent.vault_store import VaultItemMeta + + +class UnlockRequired(Exception): + """The backend is locked for this session; the surface must prompt for the master password.""" + + def __init__(self, backend: "LoginBackend"): + super().__init__(f"{backend.display_name} is locked") + self.backend = backend + + +class LoginBackend(ABC): + name: str # config key: local | onepassword | bitwarden + display_name: str # user-facing + prefix: str # handle prefix ("vault_", "op:", "bw:") + needs_unlock: bool = False + + def owns(self, handle: str) -> bool: + return handle.startswith(self.prefix) + + def is_unlocked(self) -> bool: + return True + + @abstractmethod + def list_items(self) -> List[VaultItemMeta]: + """Metadata only. Locked external backends return [] (the agent sees a lock hint instead).""" + + @abstractmethod + def get_meta(self, handle: str) -> Optional[VaultItemMeta]: ... + + @abstractmethod + def resolve_password(self, handle: str) -> str: + """Server-side only; raises ``UnlockRequired`` when locked.""" + + +def run_with_stdin_secret(argv: Sequence[str], *, env: Dict[str, str], secret: str, timeout: float, + label: str) -> subprocess.CompletedProcess: + """Run a manager CLI feeding *secret* on stdin (never argv, never env). Spawn/timeout → RuntimeError.""" + try: + return subprocess.run( # noqa: S603 — argv list, no shell + list(argv), env=env, input=secret + "\n", capture_output=True, text=True, + encoding="utf-8", errors="replace", timeout=timeout) + except subprocess.TimeoutExpired as exc: + raise RuntimeError(f"{label} unlock timed out after {timeout:.0f}s") from exc + except OSError as exc: + raise RuntimeError(f"failed to invoke {label}: {exc}") from exc + + +def _cfg() -> Dict: + from hermes_cli.config import load_config_readonly + cfg = load_config_readonly().get("vault") or {} + return cfg if isinstance(cfg, dict) else {} + + +def enabled_backends() -> List[LoginBackend]: + """Local first (always on), then each enabled external manager, in config order.""" + from agent.vault_backends.bitwarden import BitwardenLoginBackend + from agent.vault_backends.local import LocalLoginBackend + from agent.vault_backends.onepassword import OnePasswordLoginBackend + + cfg = _cfg() + out: List[LoginBackend] = [LocalLoginBackend()] + for cls in (OnePasswordLoginBackend, BitwardenLoginBackend): + section = cfg.get(cls.name) or {} + if isinstance(section, dict) and section.get("enabled"): + out.append(cls(section)) + return out + + +def backend_for_handle(handle: str) -> Optional[LoginBackend]: + return next((b for b in enabled_backends() if b.owns(handle)), None) diff --git a/agent/vault_backends/bitwarden.py b/agent/vault_backends/bitwarden.py new file mode 100644 index 0000000000..66f1095caa --- /dev/null +++ b/agent/vault_backends/bitwarden.py @@ -0,0 +1,112 @@ +"""Bitwarden Password Manager logins as a vault backend (``bw`` CLI). + +This is the personal/org *password* vault (``bw``), distinct from the +Bitwarden Secrets Manager (``bws``) source that hydrates API keys at startup. +Unlock: ``bw unlock --raw`` with the master password on stdin mints a +``BW_SESSION`` token. List: ``bw list items`` filtered to type=1 (login) with +a URI. Resolve: ``bw get password ``. +""" + +from __future__ import annotations + +import json +import logging +import os +import shutil +import subprocess +from pathlib import Path +from typing import Dict, List, Optional + +from agent.secret_sources.base import run_cli, scrub_ansi +from agent.vault_backends import unlock as _unlock +from agent.vault_backends.base import LoginBackend, UnlockRequired, run_with_stdin_secret +from agent.vault_store import VaultItemMeta, normalize_origin + +logger = logging.getLogger(__name__) + +_TIMEOUT = 30.0 +_ENV_KEEP = ("PATH", "HOME", "USERPROFILE", "APPDATA", "LOCALAPPDATA", "SystemRoot", + "TMPDIR", "TMP", "TEMP", "XDG_CONFIG_HOME", "BITWARDENCLI_APPDATA_DIR") + + +class BitwardenLoginBackend(LoginBackend): + name = "bitwarden" + display_name = "Bitwarden" + prefix = "bw:" + needs_unlock = True + + def __init__(self, cfg: Optional[Dict] = None): + self.cfg = cfg or {} + + def _bw(self) -> Path: + explicit = str(self.cfg.get("binary_path") or "") + found = explicit or shutil.which("bw") + if not found: + raise RuntimeError("Bitwarden CLI (bw) not found — install it or set vault.bitwarden.binary_path") + return Path(found) + + def _env(self, session_token: Optional[str]) -> Dict[str, str]: + env = {k: os.environ[k] for k in _ENV_KEEP if k in os.environ} + env["NO_COLOR"] = "1" + if session_token: + env["BW_SESSION"] = session_token + return env + + def is_unlocked(self) -> bool: + return _unlock.is_unlocked(self.name) + + def unlock(self, master_password: str) -> None: + proc = run_with_stdin_secret([str(self._bw()), "unlock", "--raw", "--nointeraction"], + env=self._env(None), secret=master_password, timeout=_TIMEOUT, label="bw") + token = (proc.stdout or "").strip() + if proc.returncode != 0 or not token: + err = scrub_ansi(proc.stderr or "").strip()[:200] + if "not logged in" in err.lower(): + err = "not logged in — run `bw login` once in a terminal first" + raise RuntimeError(f"Bitwarden unlock failed: {err or 'no session key'}") + _unlock.store_session_token(self.name, token) + + def _run(self, *args: str) -> str: + token = _unlock.get_session_token(self.name) + if not token: + raise UnlockRequired(self) + proc = run_cli([str(self._bw()), *args, "--nointeraction"], env=self._env(token), timeout=_TIMEOUT, + label="bw", timeout_message="bw timed out", stdin=subprocess.DEVNULL) + if proc.returncode != 0: + err = scrub_ansi(proc.stderr or "") + if "locked" in err.lower() or "session" in err.lower(): + _unlock.lock(self.name) + raise UnlockRequired(self) + raise RuntimeError(f"bw failed: {err[:200]}") + return proc.stdout or "" + + def list_items(self) -> List[VaultItemMeta]: + if not self.is_unlocked(): + return [] + raw = json.loads(self._run("list", "items") or "[]") + out: List[VaultItemMeta] = [] + for item in raw if isinstance(raw, list) else []: + if item.get("type") != 1 or not isinstance(item.get("login"), dict): + continue + login = item["login"] + origin = None + for uri in login.get("uris") or []: + try: + origin = normalize_origin(str(uri.get("uri") or "")) + break + except Exception: + continue + if not origin: + continue + username = str(login.get("username") or "").strip() or None + out.append(VaultItemMeta( + id=f"{self.prefix}{item.get('id')}", kind="login", label=str(item.get("name") or origin), + origin=origin, created_at=str(item.get("creationDate") or ""), + identifier_type="username" if username else None, identifier=username)) + return out + + def get_meta(self, handle: str) -> Optional[VaultItemMeta]: + return next((m for m in self.list_items() if m.id == handle), None) + + def resolve_password(self, handle: str) -> str: + return self._run("get", "password", handle[len(self.prefix):]).rstrip("\r\n") diff --git a/agent/vault_backends/local.py b/agent/vault_backends/local.py new file mode 100644 index 0000000000..ccb87e5680 --- /dev/null +++ b/agent/vault_backends/local.py @@ -0,0 +1,30 @@ +"""Local Fernet vault as a login backend (the always-on default).""" + +from __future__ import annotations + +from typing import List, Optional + +from agent.vault_backends.base import LoginBackend +from agent.vault_store import VaultItemMeta + + +def _store(): + # Late import: tests and callers patch ``agent.vault_store.get_vault_store``; binding it here + # at call time keeps that facade name the single seam. + from agent.vault_store import get_vault_store + return get_vault_store() + + +class LocalLoginBackend(LoginBackend): + name = "local" + display_name = "Hermes vault" + prefix = "vault_" + + def list_items(self) -> List[VaultItemMeta]: + return _store().list_items() + + def get_meta(self, handle: str) -> Optional[VaultItemMeta]: + return _store().get_meta(handle) + + def resolve_password(self, handle: str) -> str: + return str(_store().resolve_secret(handle).get("password") or "") diff --git a/agent/vault_backends/onepassword.py b/agent/vault_backends/onepassword.py new file mode 100644 index 0000000000..a700781904 --- /dev/null +++ b/agent/vault_backends/onepassword.py @@ -0,0 +1,123 @@ +"""1Password Login items as a vault backend (``op`` CLI). + +Unlock: ``op signin --raw`` with the master password on stdin (desktop-app +integration or account-level auth) mints an ``OP_SESSION_`` token. +A configured service-account token skips the prompt entirely (headless). +List: ``op item list --categories Login --format json`` → title, urls, +username. Resolve: ``op item get --fields label=password --reveal``. +""" + +from __future__ import annotations + +import json +import logging +import os +import subprocess +from pathlib import Path +from typing import Dict, List, Optional + +from agent.secret_sources.base import run_cli +from agent.secret_sources.onepassword import _OP_ENV_ALLOWLIST, _scrub, find_op +from agent.vault_backends.base import LoginBackend, UnlockRequired, run_with_stdin_secret +from agent.vault_backends import unlock as _unlock +from agent.vault_store import VaultItemMeta, normalize_origin + +logger = logging.getLogger(__name__) + +_TIMEOUT = 30.0 + + +class OnePasswordLoginBackend(LoginBackend): + name = "onepassword" + display_name = "1Password" + prefix = "op:" + needs_unlock = True + + def __init__(self, cfg: Optional[Dict] = None): + self.cfg = cfg or {} + env_name = str(self.cfg.get("service_account_token_env") or "OP_SERVICE_ACCOUNT_TOKEN") + self._service_token = os.environ.get(env_name, "") or "" + + # ── auth ──────────────────────────────────────────────────────────────── + + def _op(self) -> Path: + op = find_op(str(self.cfg.get("binary_path") or "")) + if op is None: + raise RuntimeError("1Password CLI (op) not found — install it or set vault.onepassword.binary_path") + return op + + def _env(self, session_token: Optional[str]) -> Dict[str, str]: + env = {k: os.environ[k] for k in _OP_ENV_ALLOWLIST if k in os.environ} + env["NO_COLOR"] = "1" + account = str(self.cfg.get("account") or "") + if account: + env["OP_ACCOUNT"] = account + if self._service_token: + env["OP_SERVICE_ACCOUNT_TOKEN"] = self._service_token + elif session_token: + # op signin --raw prints the bare token; the env var name carries the account shorthand, + # which op also accepts as plain OP_SESSION for the default account. + env[f"OP_SESSION_{account}" if account else "OP_SESSION"] = session_token + return env + + def is_unlocked(self) -> bool: + return bool(self._service_token) or _unlock.is_unlocked(self.name) + + def unlock(self, master_password: str) -> None: + """Mint a session token from the master password (consumed on stdin, never argv).""" + cmd = [str(self._op()), "signin", "--raw"] + if account := str(self.cfg.get("account") or ""): + cmd += ["--account", account] + proc = run_with_stdin_secret(cmd, env=self._env(None), secret=master_password, timeout=_TIMEOUT, label="op") + token = (proc.stdout or "").strip() + if proc.returncode != 0 or not token: + raise RuntimeError(f"1Password unlock failed: {_scrub(proc.stderr or '')[:200] or 'no session token'}") + _unlock.store_session_token(self.name, token) + + def _run(self, *args: str) -> str: + token = None if self._service_token else _unlock.get_session_token(self.name) + if not self._service_token and not token: + raise UnlockRequired(self) + proc = run_cli([str(self._op()), *args], env=self._env(token), timeout=_TIMEOUT, label="op", + timeout_message="op timed out", stdin=subprocess.DEVNULL) + if proc.returncode != 0: + err = _scrub(proc.stderr or "") + if "session" in err.lower() or "sign in" in err.lower() or "not signed in" in err.lower(): + _unlock.lock(self.name) + raise UnlockRequired(self) + raise RuntimeError(f"op failed: {err[:200]}") + return proc.stdout or "" + + # ── backend contract ─────────────────────────────────────────────────── + def list_items(self) -> List[VaultItemMeta]: + if not self.is_unlocked(): + return [] + raw = json.loads(self._run("item", "list", "--categories", "Login", "--format", "json") or "[]") + out: List[VaultItemMeta] = [] + for item in raw if isinstance(raw, list) else []: + urls = [str(u["href"]) for u in item.get("urls") or [] if isinstance(u, dict) and u.get("href")] + origin = _first_origin(urls) + if not origin: + continue + username = str(item.get("additional_information") or "").strip() or None + out.append(VaultItemMeta( + id=f"{self.prefix}{item.get('id')}", kind="login", label=str(item.get("title") or origin), + origin=origin, created_at=str(item.get("created_at") or ""), + identifier_type="username" if username else None, identifier=username)) + return out + + def get_meta(self, handle: str) -> Optional[VaultItemMeta]: + return next((m for m in self.list_items() if m.id == handle), None) + + def resolve_password(self, handle: str) -> str: + item_id = handle[len(self.prefix):] + return self._run("item", "get", item_id, "--fields", "label=password", "--reveal").rstrip("\r\n") + + +def _first_origin(urls: List[str]) -> Optional[str]: + for u in urls: + try: + return normalize_origin(u) + except Exception: + continue + return None diff --git a/agent/vault_backends/unlock.py b/agent/vault_backends/unlock.py new file mode 100644 index 0000000000..5f297ef6f0 --- /dev/null +++ b/agent/vault_backends/unlock.py @@ -0,0 +1,80 @@ +"""Per-process unlock state for external password managers. + +An unlock is a session token minted by the manager's CLI from the master +password (``op signin --raw`` / ``bw unlock --raw``). The token lives in +process memory only, keyed by backend, and expires after an idle TTL or an +explicit lock. The master password itself is consumed by the CLI call and +dropped; nothing is written to disk or env. + +The surface owns the prompt: ``set_unlock_prompt_callback`` is installed by +the CLI panel / TUI gateway bridge for the current thread, exactly like the +sudo-password callback. Headless contexts (cron, webhook, api_server, +single-query) install none and the vault stays locked — the same posture +approvals take where nobody can answer. +""" + +from __future__ import annotations + +import threading +import time +from typing import Callable, Dict, Optional + +_IDLE_TTL_S = 30 * 60 + +_lock = threading.Lock() +_sessions: Dict[str, tuple[str, float]] = {} # backend name → (token, last_used) +_callback_tls = threading.local() + +UnlockPrompt = Callable[[str, str], str] # (backend_name, display_name) -> master password ("" = cancelled) + + +def set_unlock_prompt_callback(cb: Optional[UnlockPrompt]) -> None: + """Register the current surface's masked master-password prompt (per-thread slot).""" + _callback_tls.prompt = cb + + +def get_unlock_prompt_callback() -> Optional[UnlockPrompt]: + return getattr(_callback_tls, "prompt", None) + + +def get_session_token(backend: str) -> Optional[str]: + with _lock: + entry = _sessions.get(backend) + if entry is None: + return None + token, last = entry + if time.monotonic() - last > _IDLE_TTL_S: + del _sessions[backend] + return None + _sessions[backend] = (token, time.monotonic()) + return token + + +def store_session_token(backend: str, token: str) -> None: + with _lock: + _sessions[backend] = (token, time.monotonic()) + + +def lock(backend: Optional[str] = None) -> None: + """Forget one backend's session (or every one when *backend* is None).""" + with _lock: + if backend is None: + _sessions.clear() + else: + _sessions.pop(backend, None) + + +def is_unlocked(backend: str) -> bool: + return get_session_token(backend) is not None + + +def can_prompt_here() -> bool: + """False in contexts where no human can answer (cron, webhook, api_server, -q).""" + from tools.approval_context import ( + _is_cron_approval_context, + _is_single_query_approval_context, + _is_unattended_platform_approval_context, + ) + if _is_cron_approval_context() or _is_unattended_platform_approval_context() or _is_single_query_approval_context(): + return False + return get_unlock_prompt_callback() is not None diff --git a/cli.py b/cli.py index 34f95cfc08..97ce5b9fbb 100644 --- a/cli.py +++ b/cli.py @@ -3000,6 +3000,8 @@ class HermesCLI(CLIProcessNotificationsMixin, CLIAgentSetupMixin, CLICommandsMix set_sudo_password_callback(self._sudo_password_callback) set_approval_callback(self._approval_callback) set_secret_capture_callback(self._secret_capture_callback) + from agent.vault_backends.unlock import set_unlock_prompt_callback + set_unlock_prompt_callback(self._vault_unlock_callback) try: from tools.computer_use_tool import set_approval_callback as _set_cu_cb @@ -3940,8 +3942,11 @@ class HermesCLI(CLIProcessNotificationsMixin, CLIAgentSetupMixin, CLICommandsMix with suppress(Exception): from tools.voice_mode import cleanup_temp_recordings cleanup_temp_recordings() - for _unset in (set_sudo_password_callback, set_approval_callback, set_secret_capture_callback): + from agent.vault_backends.unlock import lock as _vault_lock, set_unlock_prompt_callback + for _unset in (set_sudo_password_callback, set_approval_callback, set_secret_capture_callback, + set_unlock_prompt_callback): _unset(None) + _vault_lock() # session tokens for external password managers die with the session # On SIGHUP/SIGTERM the agent thread may be reaped before its own persistence runs. self._persist_active_session_before_close() diff --git a/hermes_cli/cli_chat_turn_mixin.py b/hermes_cli/cli_chat_turn_mixin.py index 86b04712fd..43577ef955 100644 --- a/hermes_cli/cli_chat_turn_mixin.py +++ b/hermes_cli/cli_chat_turn_mixin.py @@ -278,10 +278,12 @@ class CLIChatTurnMixin: _prepend_note_to_message, set_approval_callback, set_secret_capture_callback, set_sudo_password_callback, ) + from agent.vault_backends.unlock import set_unlock_prompt_callback # terminal_tool callbacks are thread-local: run()'s registration is invisible here. set_sudo_password_callback(self._sudo_password_callback) set_approval_callback(self._approval_callback) set_secret_capture_callback(self._secret_capture_callback) + set_unlock_prompt_callback(self._vault_unlock_callback) # Bind the approval session key so ``is_current_session_yolo_enabled()`` resolves # against the same key ``/yolo`` toggles under (``enable_session_yolo(self.session_id)``). try: @@ -341,6 +343,7 @@ class CLIChatTurnMixin: set_sudo_password_callback(None) set_approval_callback(None) set_secret_capture_callback(None) + set_unlock_prompt_callback(None) except Exception: pass # Unbind the per-turn key; ``_session_yolo`` state itself persists across turns. diff --git a/hermes_cli/cli_commands_mixin.py b/hermes_cli/cli_commands_mixin.py index 4355234651..8549b489e4 100644 --- a/hermes_cli/cli_commands_mixin.py +++ b/hermes_cli/cli_commands_mixin.py @@ -1921,8 +1921,10 @@ class CLICommandsMixin: runtime = turn_route["runtime"] def produce(): + from agent.vault_backends.unlock import set_unlock_prompt_callback set_sudo_password_callback(self._sudo_password_callback) set_approval_callback(self._approval_callback) + set_unlock_prompt_callback(self._vault_unlock_callback) with suppress(Exception): set_secret_capture_callback(self._secret_capture_callback) try: @@ -1960,6 +1962,7 @@ class CLICommandsMixin: set_sudo_password_callback(None) set_approval_callback(None) set_secret_capture_callback(None) + set_unlock_prompt_callback(None) def done(): self._background_tasks.pop(task_id, None) diff --git a/hermes_cli/cli_modal_mixin.py b/hermes_cli/cli_modal_mixin.py index 3d6ea5a148..53a5fbd3cd 100644 --- a/hermes_cli/cli_modal_mixin.py +++ b/hermes_cli/cli_modal_mixin.py @@ -852,6 +852,29 @@ class CLIModalMixin: self._approval_state = None self._invalidate() + def _vault_unlock_callback(self, backend_name: str, display_name: str) -> str: + """Masked master-password prompt for an external password manager (agent thread). + Reuses the sudo panel state so rendering, Enter/ESC handling and interrupt cleanup are shared.""" + from cli import _DIM, _RST, _cprint + + response_queue = queue.Queue() + self._capture_modal_input_snapshot() + self._sudo_state = {"response_queue": response_queue, "vault_backend": display_name} + self._sudo_deadline = _time.monotonic() + 120 + self._ring_bell(prompt=True, context=f"unlock {display_name}") + self._paint_now() + + result = self._poll_modal_queue(response_queue, "_sudo_deadline", refresh=0) + self._sudo_state = None + self._sudo_deadline = 0 + self._restore_modal_input_snapshot() + self._paint_now() + if result is _TIMED_OUT or not result: + _cprint(f"\n{_DIM} ⏭ {display_name} stays locked{_RST}") + return "" + _cprint(f"\n{_DIM} ✓ Unlocking {display_name} for this session{_RST}") + return result + def _secret_capture_callback(self, var_name: str, prompt: str, metadata=None) -> dict: self._capture_modal_input_snapshot() try: diff --git a/hermes_cli/cli_tui_mixin.py b/hermes_cli/cli_tui_mixin.py index 44e88b1c09..935ee2ad85 100644 --- a/hermes_cli/cli_tui_mixin.py +++ b/hermes_cli/cli_tui_mixin.py @@ -688,6 +688,12 @@ class CLITuiMixin: def _get_sudo_display_fragments(self): if not self._sudo_state: return [] + if backend := self._sudo_state.get("vault_backend"): + return self._render_sudo_style_panel( + f'🔐 Unlock {backend}', + [f'The agent wants to sign into a site with a login saved in {backend}.', + 'Enter your master password below (hidden) to unlock it for this session, or press Enter to keep it locked.', + 'The password never reaches the model; only a session token is kept in memory.']) return self._render_sudo_style_panel( '🔐 Sudo Password Required', ['Enter password below (hidden), or press Enter to skip']) diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py index 07ead3055f..10e58918cc 100644 --- a/hermes_cli/config_defaults.py +++ b/hermes_cli/config_defaults.py @@ -2158,6 +2158,23 @@ DEFAULT_CONFIG = { }, # External secret sources — pull credentials from secret managers at startup instead of storing # them in ~/.hermes/.env. + # Browser credential vault: which login sources browser_vault_list/fill may draw from. The local + # encrypted vault (`hermes vault add`, Desktop → Settings → Credential Vault) is always on. + # External password managers are unlocked per session with a masked master-password prompt; + # headless sessions (cron, webhook, API) never prompt and see them as locked. + "vault": { + "onepassword": { + "enabled": False, # `op` CLI: Login items with a website URL become fillable handles. + "account": "", # account shorthand for `op --account`; empty = default account. + "binary_path": "", # absolute path to op; empty = PATH. + # Env var holding a service-account token (headless auth, no unlock prompt). Unset = prompt. + "service_account_token_env": "OP_SERVICE_ACCOUNT_TOKEN", + }, + "bitwarden": { + "enabled": False, # `bw` CLI (Password Manager, not Secrets Manager); run `bw login` once first. + "binary_path": "", # absolute path to bw; empty = PATH. + }, + }, "secrets": { # Optional ordering of enabled sources (e.g. [onepassword, bitwarden]); default registration # order. Mapped sources (explicit VAR→ref) always beat bulk sources (BSM project dumps); diff --git a/tests/agent/test_vault_backends.py b/tests/agent/test_vault_backends.py new file mode 100644 index 0000000000..305615c2ba --- /dev/null +++ b/tests/agent/test_vault_backends.py @@ -0,0 +1,133 @@ +"""Invariants for external password-manager vault backends (1Password / Bitwarden). + +Two contracts that must never regress: +1. A locked manager never prompts where nobody can answer (cron/headless) and never leaks a + value: browser_vault_list reports it under ``locked``, browser_vault_fill refuses. +2. The unlock path hands the master password to the manager CLI on stdin only (never argv), + keeps just the session token in memory, and a fill then routes by handle prefix through + the real subprocess path. Locking forgets the token. +""" + +from __future__ import annotations + +import json +import os +import stat +from unittest.mock import patch + +import pytest + +from agent.vault_backends import unlock as unlock_mod +from agent.vault_backends.bitwarden import BitwardenLoginBackend + +# A stand-in `bw` that mimics the three commands the backend uses. It records argv + stdin so the +# test can prove the master password travelled on stdin only, and a fake session key gates `list`. +# (No env passthrough: the backend's allowlisted child env is part of what is under test.) +_FAKE_BW = r'''#!/usr/bin/env python3 +import json, os, sys +log = open(os.path.join(os.path.dirname(os.path.abspath(__file__)), "bw.log"), "a") +argv = sys.argv[1:] +stdin = sys.stdin.read() if not sys.stdin.isatty() else "" +log.write(json.dumps({"argv": argv, "stdin": stdin, "BW_SESSION": os.environ.get("BW_SESSION")}) + "\n") +if argv[:2] == ["unlock", "--raw"]: + if stdin.strip() != "correct horse": + sys.stderr.write("Invalid master password.\n"); sys.exit(1) + print("SESSION-TOKEN-123"); sys.exit(0) +if os.environ.get("BW_SESSION") != "SESSION-TOKEN-123": + sys.stderr.write("Vault is locked.\n"); sys.exit(1) +if argv[:2] == ["list", "items"]: + print(json.dumps([{"id": "abc", "type": 1, "name": "Example", "creationDate": "2026-01-01T00:00:00Z", + "login": {"username": "jane@example.com", "uris": [{"uri": "https://example.com/login"}]}}, + {"id": "note", "type": 2, "name": "Secure note"}])); sys.exit(0) +if argv[:2] == ["get", "password"]: + print("plain sentence nobody would flag 7"); sys.exit(0) +sys.exit(2) +''' + + +pytestmark = pytest.mark.skipif(os.name == "nt", reason="fake bw is a shebang script; the backend under test is host-agnostic") + + +@pytest.fixture +def fake_bw(tmp_path, monkeypatch): + exe = tmp_path / "bw" + exe.write_text(_FAKE_BW, encoding="utf-8") + exe.chmod(exe.stat().st_mode | stat.S_IXUSR) + log = tmp_path / "bw.log" # the backend runs bw with an allowlisted env, so the fake logs beside itself + monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) + unlock_mod.lock() + yield exe, log + unlock_mod.lock() + + +def _enabled(exe): + """The tool late-imports ``enabled_backends`` from the package facade; ``backend_for_handle`` reads + the sibling's own binding — patch both so the fake is the only backend anywhere.""" + backend = BitwardenLoginBackend({"enabled": True, "binary_path": str(exe)}) + return patch("agent.vault_backends.base.enabled_backends", return_value=[backend]), backend + + +def test_locked_manager_is_reported_not_prompted_when_headless(fake_bw, monkeypatch): + exe, log = fake_bw + from tools.browser_vault_tool import browser_vault_fill, browser_vault_list + + patcher, backend = _enabled(exe) + monkeypatch.setenv("HERMES_CRON_SESSION", "1") # headless: nobody can answer a prompt + unlock_mod.set_unlock_prompt_callback(lambda *_: "correct horse") # even a wired prompt must not fire + try: + with patcher, patch("agent.vault_backends.enabled_backends", return_value=[backend]): + listed = json.loads(browser_vault_list()) + assert listed["items"] == [] + assert listed["locked"] == [{"backend": "bitwarden", "display_name": "Bitwarden", + "unlock": "unavailable_in_this_session"}] + filled = json.loads(browser_vault_fill("bw:abc", task_id="t")) + assert filled["success"] is False and filled["error_type"] == "unlock_unavailable" + finally: + unlock_mod.set_unlock_prompt_callback(None) + assert not log.exists(), "bw must not be invoked at all while locked in a headless session" + assert not unlock_mod.is_unlocked("bitwarden") + + +def test_unlock_feeds_stdin_only_then_fill_routes_by_prefix(fake_bw): + exe, log = fake_bw + from tools.browser_vault_tool import browser_vault_fill, browser_vault_list + + patcher, backend = _enabled(exe) + prompts = [] + + def prompt(name, display): + prompts.append((name, display)) + return "correct horse" + + unlock_mod.set_unlock_prompt_callback(prompt) + try: + with patcher, patch("agent.vault_backends.enabled_backends", return_value=[backend]), \ + patch("tools.browser_vault_tool._current_page_origin", return_value="https://example.com"), \ + patch("tools.browser_vault_tool._eval_js", return_value={"success": True, "result": json.dumps([ + {"tag": "input", "type": "password", "name": "password", "id": "pw", "autocomplete": "current-password", + "visible": True}])}), \ + patch("tools.browser_vault_tool._eval_js_secret", return_value={"success": True, "result": json.dumps( + {"filled": 1})}) as secret_eval: + out = json.loads(browser_vault_fill("bw:abc", task_id="t")) + assert out == {"success": True, "filled_fields": 1, "backend": "bitwarden", "kind": "login", + "origin": "https://example.com"} + assert prompts == [("bitwarden", "Bitwarden")] + # Now unlocked: listing exposes metadata only, never the password. + listed = json.loads(browser_vault_list()) + assert listed["items"][0]["handle"] == "bw:abc" + assert listed["items"][0]["identifier"] == "jane@example.com" + assert "plain sentence nobody would flag 7" not in json.dumps(listed) + # The password reached the fill script, and only there. + assert "plain sentence nobody would flag 7" in secret_eval.call_args.args[1] + finally: + unlock_mod.set_unlock_prompt_callback(None) + + calls = [json.loads(line) for line in log.read_text(encoding="utf-8").splitlines()] + unlock_calls = [c for c in calls if c["argv"][:2] == ["unlock", "--raw"]] + assert len(unlock_calls) == 1 and unlock_calls[0]["stdin"].strip() == "correct horse" + assert all("correct horse" not in " ".join(c["argv"]) for c in calls), "master password must never be argv" + assert all(c["BW_SESSION"] == "SESSION-TOKEN-123" for c in calls if c["argv"][0] != "unlock") + + unlock_mod.lock("bitwarden") + assert not backend.is_unlocked() + assert os.environ.get("BW_SESSION") is None, "session token must never touch the process env" diff --git a/tests/test_browser_vault.py b/tests/test_browser_vault.py index 488e1e520b..d2ebdbb4d6 100644 --- a/tests/test_browser_vault.py +++ b/tests/test_browser_vault.py @@ -314,6 +314,7 @@ class TestBrowserVaultTools: assert out == { "success": True, "filled_fields": 1, + "backend": "local", "kind": "login", "origin": "https://example.com", } diff --git a/tools/browser_vault_tool.py b/tools/browser_vault_tool.py index 599fe85960..b0966db07a 100644 --- a/tools/browser_vault_tool.py +++ b/tools/browser_vault_tool.py @@ -37,11 +37,11 @@ logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- def _check_vault_available() -> bool: - """Tools are only in the schema when the local vault has ≥1 item.""" + """Schema-gate: the tools appear only when the local vault has items or an external manager is enabled.""" try: + from agent.vault_backends import enabled_backends from agent.vault_store import get_vault_store - - return get_vault_store().has_items() + return get_vault_store().has_items() or any(b.needs_unlock for b in enabled_backends()) except Exception: return False @@ -157,28 +157,67 @@ def _current_page_origin(task_id: str) -> Optional[str]: # --------------------------------------------------------------------------- def browser_vault_list() -> str: - """List vault items as handles + metadata. Secret values never included. + """List login handles + metadata across every enabled backend. Passwords are never included. - Login identifiers (email/username/phone) ARE included — they are - metadata, not secrets, so the agent can type the identifier itself. + A locked external manager contributes no items; instead it is reported under ``locked`` so the + agent knows to call browser_vault_fill (which prompts the user to unlock) or tell the user. """ - from agent.vault_store import get_vault_store + from agent.vault_backends import enabled_backends + from agent.vault_backends.unlock import can_prompt_here - items = [] - for meta in get_vault_store().list_items(): - entry = { - "handle": meta.id, - "label": meta.label, - "kind": meta.kind, - "origin": meta.origin, - # Phase 1: only login items are fillable. - "available": meta.kind == "login", - } - if meta.identifier: - entry["identifier"] = meta.identifier - entry["identifier_type"] = meta.identifier_type - items.append(entry) - return json.dumps({"success": True, "items": items}, ensure_ascii=False) + items, locked, errors = [], [], [] + for backend in enabled_backends(): + if backend.needs_unlock and not backend.is_unlocked(): + locked.append({"backend": backend.name, "display_name": backend.display_name, + "unlock": "browser_vault_unlock" if can_prompt_here() else "unavailable_in_this_session"}) + continue + try: + metas = backend.list_items() + except Exception as exc: + errors.append({"backend": backend.name, "error": str(exc)[:200]}) + continue + for meta in metas: + entry = {"handle": meta.id, "backend": backend.name, "label": meta.label, "kind": meta.kind, + "origin": meta.origin, "available": meta.kind == "login"} + if meta.identifier: + entry["identifier"] = meta.identifier + entry["identifier_type"] = meta.identifier_type + items.append(entry) + out: Dict[str, Any] = {"success": True, "items": items} + if locked: + out["locked"] = locked + if errors: + out["errors"] = errors + return json.dumps(out, ensure_ascii=False) + + +def browser_vault_unlock(backend_name: str) -> str: + """Ask the user (via the surface's masked prompt) to unlock an external manager for this session.""" + from agent.vault_backends import enabled_backends + from agent.vault_backends.unlock import can_prompt_here, get_unlock_prompt_callback + + backend = next((b for b in enabled_backends() if b.name == backend_name and b.needs_unlock), None) + if backend is None: + return json.dumps({"success": False, "error": f"No unlockable vault backend named {backend_name!r}."}) + if backend.is_unlocked(): + return json.dumps({"success": True, "backend": backend.name, "already_unlocked": True}) + if not can_prompt_here(): + return json.dumps({"success": False, "error_type": "unlock_unavailable", + "error": (f"{backend.display_name} is locked and this session cannot prompt for the " + "master password (headless/cron/API). Unlock it from an interactive Hermes " + "session or the Desktop app first.")}) + prompt = get_unlock_prompt_callback() + master = prompt(backend.name, backend.display_name) if prompt else "" + if not master: + return json.dumps({"success": False, "error_type": "unlock_cancelled", + "error": f"The user declined to unlock {backend.display_name}."}) + try: + backend.unlock(master) # type: ignore[attr-defined] + except Exception as exc: + return json.dumps({"success": False, "error_type": "unlock_failed", "error": str(exc)[:300]}) + finally: + del master + return json.dumps({"success": True, "backend": backend.name}) def browser_vault_fill(handle: str, task_id: Optional[str] = None) -> str: @@ -198,12 +237,21 @@ def browser_vault_fill(handle: str, task_id: Optional[str] = None) -> str: classify_login_control, select_password_fill, ) - from agent.vault_store import VaultError, get_vault_store, scrub_secret_from_text + from agent.vault_backends import UnlockRequired, backend_for_handle + from agent.vault_store import scrub_secret_from_text effective_task_id = task_id or "default" - store = get_vault_store() + backend = backend_for_handle(handle) + if backend is not None and backend.needs_unlock and not backend.is_unlocked(): + unlocked = json.loads(browser_vault_unlock(backend.name)) + if not unlocked.get("success"): + return json.dumps(unlocked) - meta = store.get_meta(handle) + try: + meta = backend.get_meta(handle) if backend is not None else None + except UnlockRequired: + return json.dumps({"success": False, "error_type": "unlock_required", + "error": f"{backend.display_name} locked again; call browser_vault_unlock."}) if meta is None: return json.dumps( { @@ -263,8 +311,12 @@ def browser_vault_fill(handle: str, task_id: Optional[str] = None) -> str: return json.dumps({"success": False, "error": "No login form fields were found on the current page."}) # ── Resolve secret and fill (secret never enters any logged string) ───── - secret = store.resolve_secret(handle) - password = str(secret.get("password") or "") + try: + password = backend.resolve_password(handle) + except UnlockRequired: + return json.dumps({"success": False, "error_type": "unlock_required", + "error": f"{backend.display_name} locked again; call browser_vault_unlock."}) + secret = {"password": password} fills = select_password_fill(classified, password) if not fills: return json.dumps( @@ -314,6 +366,7 @@ def browser_vault_fill(handle: str, task_id: Optional[str] = None) -> str: { "success": bool(filled), "filled_fields": int(filled), + "backend": backend.name, "kind": meta.kind, "origin": meta.origin, } @@ -327,33 +380,48 @@ def browser_vault_fill(handle: str, task_id: Optional[str] = None) -> str: BROWSER_VAULT_LIST_SCHEMA = { "name": "browser_vault_list", "description": ( - "List credentials stored in the local encrypted vault as handles " - "with metadata (label, kind, bound origin, and for logins the " - "identifier + identifier_type — identifiers are visible so you can " - "type them yourself with fill_input). Passwords are NEVER returned. " - "Workflow: type the identifier with fill_input, then call " - "browser_vault_fill with the handle to fill the password." + "List saved website logins as handles with metadata (label, backend, bound origin, and the " + "identifier + identifier_type so you can type the username yourself with fill_input). " + "Passwords are NEVER returned. Sources: the local Hermes vault plus any enabled password " + "manager (1Password, Bitwarden). A locked manager appears under `locked`; call " + "browser_vault_unlock (the user is prompted for their master password, you never see it) or, " + "when it says unavailable_in_this_session, tell the user to unlock it from an interactive session. " + "Workflow: fill_input the identifier, then browser_vault_fill with the handle." ), "input_schema": {"type": "object", "properties": {}, "required": []}, } +BROWSER_VAULT_UNLOCK_SCHEMA = { + "name": "browser_vault_unlock", + "description": ( + "Ask the user to unlock a password manager (1Password or Bitwarden) for this session. The master " + "password is typed into a masked prompt owned by the UI and never enters the conversation. " + "Returns success, unlock_cancelled, unlock_failed, or unlock_unavailable (headless session)." + ), + "input_schema": { + "type": "object", + "properties": {"backend": {"type": "string", "enum": ["onepassword", "bitwarden"], + "description": "Backend name from browser_vault_list `locked`."}}, + "required": ["backend"], + }, +} + BROWSER_VAULT_FILL_SCHEMA = { "name": "browser_vault_fill", "description": ( - "Fill ONLY the password field of the CURRENT browser page's login " - "form from a vault handle (see browser_vault_list). Type the " - "identifier/username yourself first with fill_input (it is visible " - "in the vault metadata), then call this to fill the password. The " - "password is resolved and injected server-side; it never appears in " - "the conversation. Refused unless the page origin exactly matches " - "the credential's bound origin (re-checked atomically at fill time)." + "Fill ONLY the password field of the CURRENT browser page's login form from a vault handle " + "(see browser_vault_list). Type the identifier/username yourself first with fill_input, then " + "call this. The password is resolved from the local vault or the password manager and injected " + "server-side; it never appears in the conversation. Refused unless the page origin exactly " + "matches the credential's bound origin (re-checked atomically at fill time). If the manager is " + "locked the user is prompted to unlock first." ), "input_schema": { "type": "object", "properties": { "handle": { "type": "string", - "description": "Vault item handle from browser_vault_list (e.g. vault_ab12cd34ef56)", + "description": "Handle from browser_vault_list (vault_… local, op:… 1Password, bw:… Bitwarden)", } }, "required": ["handle"], @@ -365,6 +433,10 @@ def _handle_vault_list(args: Dict[str, Any], **kwargs) -> str: return browser_vault_list() +def _handle_vault_unlock(args: Dict[str, Any], **kwargs) -> str: + return browser_vault_unlock(str(args.get("backend") or "")) + + def _handle_vault_fill(args: Dict[str, Any], **kwargs) -> str: return browser_vault_fill( handle=str(args.get("handle") or ""), task_id=kwargs.get("task_id") @@ -382,6 +454,15 @@ registry.register( emoji="🔐", ) +registry.register( + name="browser_vault_unlock", + toolset="browser", + schema=BROWSER_VAULT_UNLOCK_SCHEMA, + handler=_handle_vault_unlock, + check_fn=_check_vault_available, + emoji="🔐", +) + registry.register( name="browser_vault_fill", toolset="browser", diff --git a/toolsets.py b/toolsets.py index f8e98a53bf..cd2afd7bd9 100644 --- a/toolsets.py +++ b/toolsets.py @@ -18,7 +18,7 @@ _HERMES_CORE_TOOLS = [ "browser_type", "browser_scroll", "browser_back", "browser_press", "browser_get_images", "browser_vision", "browser_console", "browser_cdp", "browser_dialog", - "browser_vault_list", "browser_vault_fill", # gated on a non-empty vault via check_fn + "browser_vault_list", "browser_vault_unlock", "browser_vault_fill", # check_fn: vault items or a manager enabled "browser_exec", # replaces the other browser tools when browser.backend is "browser-use" "text_to_speech", "todo_list", "memory", diff --git a/tui_gateway/agent_callbacks.py b/tui_gateway/agent_callbacks.py index 36f421b90b..db15890d16 100644 --- a/tui_gateway/agent_callbacks.py +++ b/tui_gateway/agent_callbacks.py @@ -169,6 +169,11 @@ def _wire_callbacks(sid: str): set_sudo_password_callback(lambda: _block("sudo.request", sid, {}, timeout=120)) set_project_workspace_callback(_apply_project_workspace) set_secret_capture_callback(secret_cb) + # External password-manager unlock: the renderer shows a masked master-password card; the + # answer is consumed by the manager CLI on stdin and only a session token stays in memory. + from agent.vault_backends.unlock import set_unlock_prompt_callback + set_unlock_prompt_callback(lambda backend, display_name: _block( + "vault.unlock.request", sid, {"backend": backend, "display_name": display_name}, timeout=120)) def _available_personalities(cfg: dict | None = None) -> dict: diff --git a/tui_gateway/methods_prompt.py b/tui_gateway/methods_prompt.py index d1fdc3e530..d3e8455566 100644 --- a/tui_gateway/methods_prompt.py +++ b/tui_gateway/methods_prompt.py @@ -1097,7 +1097,7 @@ def _(rid, params: dict) -> dict: _LATE_RESPOND_KEYS = { "terminal.read.respond": "text", "preview.read.respond": "text", "preview.act.respond": "text", "window.read.respond": "text", "tour.respond": "text", "mcp.setup.respond": "result", - "sudo.respond": "password", "secret.respond": "value"} + "sudo.respond": "password", "secret.respond": "value", "vault.unlock.respond": "password"} for _name, _key in _LATE_RESPOND_KEYS.items(): method(_name)(lambda rid, params, _k=_key: _respond(rid, params, _k, allow_expired=True)) del _name, _key diff --git a/tui_gateway/methods_vault.py b/tui_gateway/methods_vault.py index 96ef465c72..5d7a9fec3b 100644 --- a/tui_gateway/methods_vault.py +++ b/tui_gateway/methods_vault.py @@ -12,6 +12,9 @@ JSON-RPC channel every other Settings surface uses. Contracts: encrypted store, and is never logged. Error strings are defensively scrubbed with ``scrub_secret_from_text`` before they leave the handler. - ``vault.remove`` → {removed: bool}. +- ``vault.sources`` / ``vault.source.set`` → external password-manager status and enable toggle. +- ``vault.unlock`` / ``vault.lock`` → per-session unlock of a manager from Settings; the master + password is consumed by the manager CLI on stdin and never stored or logged. Handlers are rebound onto server.py's globals at install time (see method_ctx.py) and may reference server module globals (``_ok``, ``_err``). @@ -29,16 +32,95 @@ method = _registry.method @method("vault.list") def _(rid, params: dict) -> dict: - """Metadata-only listing of vault items. Secret values are never included.""" + """Metadata-only listing across every enabled backend (local + unlocked password managers). + Each item carries ``backend``; locked managers contribute nothing (see vault.sources).""" try: - from agent.vault_store import get_vault_store + from agent.vault_backends import enabled_backends - items = [meta.to_dict() for meta in get_vault_store().list_items()] + items = [] + for backend in enabled_backends(): + if backend.needs_unlock and not backend.is_unlocked(): + continue + items.extend({**meta.to_dict(), "backend": backend.name} for meta in backend.list_items()) return _ok(rid, {"items": items}) except Exception as e: return _err(rid, 5095, str(e)) +_EXTERNAL_SOURCES = ("onepassword", "bitwarden") + + +@method("vault.sources") +def _(rid, params: dict) -> dict: + """Status of every login source: {name, display_name, enabled, needs_unlock, unlocked, installed}.""" + import shutil + + from agent.vault_backends import enabled_backends + from agent.vault_backends.bitwarden import BitwardenLoginBackend + from agent.vault_backends.onepassword import OnePasswordLoginBackend + from agent.secret_sources.onepassword import find_op + + enabled = {b.name: b for b in enabled_backends()} + rows = [{"name": "local", "display_name": "Hermes vault", "enabled": True, "needs_unlock": False, + "unlocked": True, "installed": True}] + for cls, installed in ((OnePasswordLoginBackend, find_op() is not None), + (BitwardenLoginBackend, shutil.which("bw") is not None)): + live = enabled.get(cls.name) + rows.append({"name": cls.name, "display_name": cls.display_name, "enabled": live is not None, + "needs_unlock": True, "unlocked": bool(live and live.is_unlocked()), "installed": installed}) + return _ok(rid, {"sources": rows}) + + +@method("vault.source.set") +def _(rid, params: dict) -> dict: + """Enable/disable an external manager: writes ``vault..enabled`` and locks it when disabling.""" + from agent.vault_backends.unlock import lock + from hermes_cli.config import load_config, save_config + + name = str(params.get("name") or "") + if name not in _EXTERNAL_SOURCES: + return _err(rid, 5095, f"unknown vault source: {name}") + enabled = bool(params.get("enabled")) + cfg = load_config() + section = cfg.setdefault("vault", {}).setdefault(name, {}) + section["enabled"] = enabled + if not enabled: + lock(name) + save_config(cfg) + return _ok(rid, {"name": name, "enabled": enabled}) + + +@method("vault.unlock") +def _(rid, params: dict) -> dict: + """Unlock a manager with the master password typed in the Settings dialog (consumed by the CLI on stdin).""" + from agent.vault_backends import enabled_backends + + name = str(params.get("name") or "") + password = str(params.get("password") or "") + backend = next((b for b in enabled_backends() if b.name == name and b.needs_unlock), None) + if backend is None: + return _err(rid, 5095, f"{name} is not an enabled password manager") + if not password: + return _err(rid, 5095, "master password is required") + try: + backend.unlock(password) # type: ignore[attr-defined] + except Exception as e: + return _err(rid, 5095, str(e).replace(password, "[REDACTED]")) + finally: + del password + return _ok(rid, {"name": name, "unlocked": True}) + + +@method("vault.lock") +def _(rid, params: dict) -> dict: + """Forget a manager's session token (or every one when ``name`` is omitted).""" + from agent.vault_backends.unlock import lock + + name = params.get("name") + lock(str(name) if name else None) + return _ok(rid, {"locked": True}) + + @method("vault.add") def _(rid, params: dict) -> dict: """Add a vault item. ``secret`` values go straight into the encrypted store. diff --git a/tui_gateway/server.py b/tui_gateway/server.py index 638734d0fb..c34dbf05f2 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -1247,7 +1247,7 @@ def _enable_gateway_prompts() -> None: # Blocking bridges whose `*.respond` tolerates a late reply (allow_expired=True): on timeout the tool # returns empty, but a slow renderer could still answer and hit a raw 4009 — `.expire` tears the card down. _EXPIRING_REQUESTS = frozenset({ - "secret.request", "sudo.request", "clarify.request", "terminal.read.request", + "secret.request", "sudo.request", "vault.unlock.request", "clarify.request", "terminal.read.request", "preview.read.request", "preview.act.request", "window.read.request", "mcp.setup.request", "tour.request", })