hermes-agent/hermes_cli/_secrets_common.py

"""Helpers shared by the Bitwarden and 1Password ``hermes secrets`` CLIs.

Import-light on purpose: ``hermes_cli.secrets_cli`` must stay free of the Bitwarden backend
(``cryptography``) at import time, so nothing here touches a secret-source backend.
"""

from __future__ import annotations

import argparse
import os
import re
import subprocess
import sys
from pathlib import Path
from typing import Callable, Iterable, Optional, Sequence

from rich.console import Console
from rich.markup import escape as _escape_markup
from rich.panel import Panel
from rich.table import Table

from hermes_cli.config import load_config, save_config
from hermes_cli.secret_prompt import masked_secret_prompt


_CONTROL_CHARS_RE = re.compile(r"[\x00-\x08\x0b-\x1f\x7f-\x9f]")


def console_safe(text: str) -> str:
    """Make externally-derived text safe for a markup console.

    Strips terminal control characters (C0 other than tab/newline, DEL, C1) so
    embedded ANSI/OSC sequences cannot reach the terminal, then escapes Rich
    markup so ``[...]`` sequences print literally.
    """
    return _escape_markup(_CONTROL_CHARS_RE.sub("", text))


def yn(b: bool) -> str:
    return "[green]yes[/green]" if b else "[dim]no[/dim]"


def section_cfg(cfg: dict, section: str) -> dict:
    """``cfg["secrets"][section]`` or ``{}``."""
    return (cfg.get("secrets") or {}).get(section) or {}


def cfg_str(cfg: dict, key: str) -> str:
    """A stripped string config value; ``""`` for missing/None."""
    return str(cfg.get(key, "") or "").strip()


def arg(name: str, help_text: str, **kwargs) -> tuple:
    """One ``add_argument`` spec for :func:`register_subcommands`."""
    return name, dict(help=help_text, **kwargs)


def flag(name: str, help_text: str) -> tuple:
    """A boolean ``store_true`` spec for :func:`register_subcommands`."""
    return arg(name, help_text, action="store_true")


def register_subcommands(parent: argparse.ArgumentParser, dest: str, commands: Iterable) -> None:
    """Attach ``(name, help, handler, [arg(...), ...])`` subcommands to ``parent``."""
    sub = parent.add_subparsers(dest=dest)
    for name, help_text, func, arguments in commands:
        parser = sub.add_parser(name, help=help_text)
        for arg_name, kwargs in arguments:
            parser.add_argument(arg_name, **kwargs)
        parser.set_defaults(func=func)


def require_enabled(console: Console, cfg: dict, product: str, command: str) -> bool:
    """Print the "integration is disabled" hint and return False unless ``cfg["enabled"]``."""
    if cfg.get("enabled"):
        return True
    console.print(f"[yellow]{product} integration is disabled.  Run "
                  f"`hermes secrets {command} setup` first.[/yellow]")
    return False


def print_status_panel(console: Console, title: str, rows: Iterable) -> None:
    """Two-column key/value table inside a cyan panel (the ``status`` layout)."""
    table = Table(show_header=False, box=None, padding=(0, 2))
    table.add_column("", style="bold")
    table.add_column("")
    for label, value in rows:
        table.add_row(label, value)
    console.print(Panel(table, title=title, border_style="cyan"))


def print_table(console: Console, columns: Sequence, rows: Iterable,
                warnings: Iterable[str] = (), indent: str = "") -> None:
    """Headed table; ``columns`` entries are ``header`` or ``(header, add_column kwargs)``.

    ``warnings`` are echoed after the table, one ``warning:`` line each.
    """
    table = Table(show_header=True, header_style="bold")
    for col in columns:
        header, kwargs = (col, {}) if isinstance(col, str) else col
        table.add_column(header, **kwargs)
    for row in rows:
        table.add_row(*row)
    console.print(table)
    for w in warnings:
        console.print(f"{indent}[yellow]warning:[/yellow] {console_safe(w)}")


def cli_version(binary: Path) -> str:
    """Return the first line of ``<binary> --version`` or ``"version unknown"``."""
    try:
        res = subprocess.run([str(binary), "--version"], env=secret_cli_env(), capture_output=True, text=True, encoding='utf-8',
                             errors='replace', timeout=5, check=False)
        if res.returncode == 0:
            return (res.stdout or res.stderr).strip().splitlines()[0]
    except (OSError, subprocess.TimeoutExpired):
        pass
    return "version unknown"


def secret_cli_env(*allow_env: str) -> dict:
    """Allowlisted env for a secret-manager CLI probe (``bws`` / ``op``): the real HOME
    (both CLIs keep state there) plus ``allow_env``; callers add the token explicitly."""
    from agent.secret_sources.base import build_minimal_provider_env

    return build_minimal_provider_env(os.environ, allow_env=allow_env)


def rotate_token(
    console: Console, given: Optional[str], token_env: str, *, flag: str, intro: str, prompt: str,
    verify: Optional[Callable[[str], bool]], save: Callable[[str, str], object],
    env_path: Callable[[], object], clear_caches: Callable[[], object],
    disabled_note: Optional[str],
) -> int:
    """Shared ``token`` subcommand: prompt, optionally verify, then persist. Returns the exit code.

    ``verify(token)`` prints its own diagnostics and returns False to abort without touching
    .env — so a bad paste never bricks the working token; None skips verification
    (``--no-verify``). ``save``/``env_path`` are passed in (not imported here) so each CLI
    module's own ``save_env_value``/``get_env_path`` bindings — which tests monkeypatch — stay
    in effect. Old cached pulls are keyed on the previous token's fingerprint; clearing them
    makes the next startup fetch fresh with the new credential.
    """
    token = (given or "").strip()
    if not token:
        if not sys.stdin.isatty():
            console.print(f"[red]No TTY — pass the token with {flag}.[/red]")
            return 1
        console.print(intro)
        token = masked_secret_prompt(prompt).strip()
    if not token:
        console.print("[red]Empty token, aborting.[/red]")
        return 1
    if verify is not None and not verify(token):
        return 1
    save(token_env, token)
    os.environ[token_env] = token
    clear_caches()
    console.print(f"[green]✓[/green] stored in {env_path()} as {token_env}.  "
                  "Takes effect on the next Hermes invocation.")
    if disabled_note:
        console.print(disabled_note)
    return 0


def prompt_index(console: Console, prompt: str, count: int, *,
                 allow_empty: bool = False, empty_message: Optional[str] = None) -> int:
    """Loop until the user enters an integer in ``1..count``; return it.

    Blank input returns 0 when ``allow_empty``; otherwise ``empty_message`` (if any)
    is printed and the prompt repeats.
    """
    while True:
        choice = console.input(prompt).strip()
        if not choice:
            if allow_empty:
                return 0
            if empty_message:
                console.print(empty_message)
            continue
        try:
            idx = int(choice)
        except ValueError:
            console.print("  [red]Enter a number.[/red]")
            continue
        if 1 <= idx <= count:
            return idx
        console.print(f"  [red]Out of range — pick 1-{count}.[/red]")


def disable_secret_source(section: str, message: str) -> int:
    """Set ``secrets.<section>.enabled = False`` in config.yaml and print ``message``."""
    cfg = load_config()
    cfg.setdefault("secrets", {}).setdefault(section, {})["enabled"] = False
    save_config(cfg)
    Console().print(message)
    return 0