"""Security advisory checker for Hermes Agent. Cheap (one ``importlib.metadata.version()`` call per advisory package, safe on every CLI startup) and silent unless a compromised package is actually installed. """ from __future__ import annotations import logging import os import sys from dataclasses import dataclass from pathlib import Path from typing import Iterable, Optional from hermes_constants import display_hermes_home logger = logging.getLogger(__name__) # Advisory catalog. To add one: append an ``Advisory`` to ``ADVISORIES`` with ``compromised`` as # ``(pkg_name, frozenset_of_versions)`` pairs (versions exactly as ``importlib.metadata.version()`` # returns them; an empty frozenset flags ANY installed version — rare, namespace compromise only) # and 2-4 copy/pasteable ``remediation`` lines. Never remove old advisories: users on an older # release with the compromised package must still get warned. @dataclass(frozen=True) class Advisory: """``id`` is lowercase-hyphen, stable and never reused (it is what acks key on). ``remediation`` is ordered: uninstall command first, then credential audit/rotation guidance. Steps may use ``{hermes_home}``; ``full_remediation_text`` fills it with ``display_hermes_home()`` at render time so a profile / HERMES_HOME user is sent to their own .env. """ id: str title: str summary: str url: str compromised: tuple[tuple[str, frozenset[str]], ...] remediation: tuple[str, ...] published: str = "" severity: str = "high" # low / medium / high / critical ADVISORIES: tuple[Advisory, ...] = ( Advisory( id="shai-hulud-2026-05", title="Mini Shai-Hulud worm — mistralai 2.4.6 compromised on PyPI", summary=( "PyPI quarantined the mistralai package on 2026-05-12 after a " "malicious 2.4.6 release. The worm steals credentials from " "environment variables and credential files (~/.npmrc, ~/.pypirc, " "~/.aws/credentials, GitHub PATs, cloud SDK tokens) and exfils " "them to a hardcoded webhook. If you ran any Python process that " "imported mistralai 2.4.6 — including hermes when configured " "with provider=mistral for TTS or STT — assume those credentials " "are exposed. PyPI has since removed 2.4.6 and the project ships " "clean releases again (2.4.7, 2.4.8); this advisory only fires if " "the compromised 2.4.6 is still installed." ), url="https://socket.dev/blog/mini-shai-hulud-worm-pypi", compromised=(("mistralai", frozenset({"2.4.6"})),), remediation=( "Run: pip uninstall -y mistralai (or: uv pip uninstall mistralai)", "Rotate API keys in {hermes_home}/.env (OpenRouter, Anthropic, OpenAI, " "Nous, GitHub, AWS, Google, Mistral, etc.).", "Audit ~/.npmrc, ~/.pypirc, ~/.aws/credentials, ~/.config/gh/hosts.yml, " "and any other credential files for tokens that may have been read.", "Check GitHub for unexpected new SSH keys, deploy keys, or webhook " "additions on repos you have admin on.", "After cleanup: hermes doctor --ack shai-hulud-2026-05 to dismiss " "this warning.", ), published="2026-05-12", severity="critical", ), ) @dataclass(frozen=True) class AdvisoryHit: """One package-version match against an advisory.""" advisory: Advisory package: str installed_version: str def _installed_version(pkg_name: str) -> Optional[str]: """Installed version of ``pkg_name`` via ``importlib.metadata`` (uv venvs may lack pip), or None if not installed or metadata is corrupt — never crash the CLI startup path. """ from importlib.metadata import PackageNotFoundError, version try: return version(pkg_name) except PackageNotFoundError: return None except Exception: logger.debug("importlib.metadata.version(%s) raised", pkg_name, exc_info=True) return None def detect_compromised(advisories: Iterable[Advisory] = ADVISORIES) -> list[AdvisoryHit]: """All hits: package installed AND version in the compromised set (or the set is empty).""" return [ AdvisoryHit(advisory, pkg_name, installed) for advisory in advisories for pkg_name, bad_versions in advisory.compromised if (installed := _installed_version(pkg_name)) is not None and (not bad_versions or installed in bad_versions) ] # ─── Acknowledgement persistence ────────────────────────────────────────────── # Acks live under ``security.acked_advisories`` in config.yaml as a list of advisory IDs — the only # state (no per-host data or timestamps), so a shared config.yaml dismisses everywhere. def get_acked_ids() -> set[str]: """Advisory IDs the user has dismissed; empty when config can't be loaded (never block startup — the advisory keeps firing until config is repaired). """ try: from hermes_cli.config import load_config raw = (load_config().get("security") or {}).get("acked_advisories") or [] except Exception: logger.debug("Could not load config for advisory acks", exc_info=True) return set() return {str(x).strip() for x in raw if str(x).strip()} if isinstance(raw, list) else set() def ack_advisory(advisory_id: str) -> bool: """Persist an ack for ``advisory_id``. Returns True on success.""" advisory_id = advisory_id.strip() if not advisory_id: return False try: from hermes_cli.config import load_config, save_config except Exception: logger.warning("Could not import config module to persist ack") return False try: cfg = load_config() sec = cfg.setdefault("security", {}) existing = sec.get("acked_advisories") or [] if not isinstance(existing, list): existing = [] if advisory_id not in existing: sec["acked_advisories"] = existing + [advisory_id] save_config(cfg) return True except Exception: logger.exception("Failed to persist advisory ack for %s", advisory_id) return False def filter_unacked(hits: list[AdvisoryHit]) -> list[AdvisoryHit]: """Only hits whose advisories the user has not dismissed.""" acked = get_acked_ids() if hits else set() return [h for h in hits if h.advisory.id not in acked] def _term_supports_color() -> bool: return not os.environ.get("NO_COLOR") and sys.stdout.isatty() def short_banner_lines(hits: list[AdvisoryHit]) -> list[str]: """1-3 short unstyled lines for a startup banner; always names the worst hit explicitly.""" if not hits: return [] primary = hits[0] lines = [ f"SECURITY ADVISORY [{primary.advisory.id}]: {primary.advisory.title}", f" Detected: {primary.package}=={primary.installed_version}", " Run 'hermes doctor' for remediation steps.", ] if len(hits) > 1: lines.insert(1, f" ({len(hits) - 1} additional advisor{'ies' if len(hits) > 2 else 'y'} also active.)") return lines def full_remediation_text(hit: AdvisoryHit) -> list[str]: """Multi-line block describing the advisory + remediation.""" a = hit.advisory return [ f"=== {a.title} ===", f"ID: {a.id} Severity: {a.severity} Published: {a.published}", f"Detected: {hit.package}=={hit.installed_version}", f"Reference: {a.url}", "", a.summary, "", "Remediation:", *(f" {i}. {step.format(hermes_home=display_hermes_home())}" for i, step in enumerate(a.remediation, 1)), ] # ─── Startup-banner gating ──────────────────────────────────────────────────── # Once the banner is seen we cache that in ``~/.hermes/cache/advisory_banner_seen`` (one # `` `` line per advisory). Acked advisories never re-banner; cached-but-not-acked # ones re-banner after 24h so the user doesn't fully forget. _BANNER_CACHE_FILE = "advisory_banner_seen" _BANNER_REPEAT_HOURS = 24 def _banner_cache_path() -> Optional[Path]: try: from hermes_constants import get_hermes_home cache_dir = Path(get_hermes_home()) / "cache" cache_dir.mkdir(parents=True, exist_ok=True) return cache_dir / _BANNER_CACHE_FILE except Exception: return None def _read_banner_cache() -> dict[str, float]: p = _banner_cache_path() try: lines = p.read_text(encoding="utf-8-sig").splitlines() if p is not None and p.exists() else [] except Exception: return {} out: dict[str, float] = {} for parts in (line.split(None, 1) for line in lines): try: if len(parts) == 2: out[parts[0]] = float(parts[1]) except ValueError: continue return out def _write_banner_cache(seen: dict[str, float]) -> None: try: if (p := _banner_cache_path()) is not None: p.write_text("\n".join(f"{aid} {ts}" for aid, ts in seen.items()) + "\n", encoding="utf-8") except Exception: logger.debug("Could not write advisory banner cache", exc_info=True) def hits_due_for_banner(hits: list[AdvisoryHit], *, repeat_hours: int = _BANNER_REPEAT_HOURS) -> list[AdvisoryHit]: """Hits whose banner is due (not acked, not recently shown). Side effect: stamps the banner cache for every returned hit, so callers must render the result. """ import time fresh = filter_unacked(hits) if not fresh: return [] now = time.time() cache = _read_banner_cache() cutoff = now - (repeat_hours * 3600) due = [hit for hit in fresh if cache.get(hit.advisory.id, 0.0) < cutoff] for hit in due: cache[hit.advisory.id] = now if due: _write_banner_cache(cache) return due def startup_banner(hits: list[AdvisoryHit]) -> Optional[str]: """Printable startup banner, or None if nothing is due (updates the banner cache).""" due = hits_due_for_banner(hits) if not due: return None text = "\n".join(short_banner_lines(due)) return f"\x1b[1;31m{text}\x1b[0m" if _term_supports_color() else text def gateway_log_message(hits: list[AdvisoryHit]) -> Optional[str]: """One-line log message for gateway operators, or None.""" fresh = filter_unacked(hits) if not fresh: return None if len(fresh) == 1: h = fresh[0] return (f"Security advisory [{h.advisory.id}] active: {h.package}=={h.installed_version} " f"matches {h.advisory.title}. See {h.advisory.url}") return (f"{len(fresh)} security advisories active (IDs: {', '.join(h.advisory.id for h in fresh)}). " "Run `hermes doctor` on the gateway host for details.")