diff --git a/hermes_cli/plugin_catalog.py b/hermes_cli/plugin_catalog.py new file mode 100644 index 00000000000..3de7c1ca9e6 --- /dev/null +++ b/hermes_cli/plugin_catalog.py @@ -0,0 +1,307 @@ +"""Plugin catalog — curated, Nous-approved Hermes plugins shipped with the repo. + +Mirrors the ``optional-mcps/`` MCP-catalog pattern (see +:mod:`hermes_cli.mcp_catalog`): each catalog entry is a single YAML file under +the in-tree ``plugin-catalog/`` directory, pinned to an exact 40-character +commit SHA. Users discover entries via ``hermes plugins catalog`` / +``hermes plugins search`` and install them with +``hermes plugins install ``, which clones the pinned commit. + +Catalog policy (see plugin-catalog/README.md for the full admission policy): +- Entries are added only by merging a PR into hermes-agent — presence in the + ``plugin-catalog/`` directory is the human-merged approval gate. +- Every entry pins an exact 40-hex commit SHA. SHA bumps are new PRs, + re-reviewed as diffs. The pinned release should be at least 2 weeks old at + pin time, mirroring the optional-mcps supply-chain rules. +- ``plugin-catalog/removed.yaml`` is the blocklist: entries pulled from the + catalog for security or policy reasons are recorded there so installs of + the same name/repo are refused with the recorded reason. +""" + +from __future__ import annotations + +import logging +import os +import re +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, List, Optional + +import yaml + +logger = logging.getLogger(__name__) + +CATALOG_TIERS = ("official", "community") + +_SHA_RE = re.compile(r"^[0-9a-f]{40}$") +_NAME_RE = re.compile(r"^[a-z0-9_-]{1,64}$") + + +# ─── Data classes ──────────────────────────────────────────────────────────── + + +@dataclass +class RemovedEntry: + name: str + repo: str = "" + reason: str = "" + date: str = "" # ISO date string + + +@dataclass +class CatalogCapabilities: + provides_tools: List[str] = field(default_factory=list) + provides_hooks: List[str] = field(default_factory=list) + provides_middleware: List[str] = field(default_factory=list) + requires_env: List[str] = field(default_factory=list) + + +@dataclass +class PluginCatalogEntry: + name: str # catalog key, [a-z0-9_-]{1,64} + repo: str # https:// git URL + sha: str # 40-hex pinned commit — MANDATORY, validated + description: str + maintainer: str + tier: str = "community" # one of CATALOG_TIERS + requires_hermes: str = "" # e.g. ">=0.19" (optional) + subdir: str = "" # optional path within the repo + docs_url: str = "" + platforms: List[str] = field(default_factory=list) # empty = all OSes + capabilities: CatalogCapabilities = field(default_factory=CatalogCapabilities) + + +# ─── Directory resolution ──────────────────────────────────────────────────── + + +def get_catalog_dir() -> Path: + """Return the ``plugin-catalog/`` directory shipped with this checkout. + + ``HERMES_PLUGIN_CATALOG_DIR`` overrides the location for tests only — + read via ``os.getenv`` at call time so monkeypatched values take effect. + """ + override = os.getenv("HERMES_PLUGIN_CATALOG_DIR", "").strip() + if override: + return Path(override) + return Path(__file__).resolve().parent.parent / "plugin-catalog" + + +# ─── Loading / validation ──────────────────────────────────────────────────── + + +def _str_list(raw: Any) -> List[str]: + """Coerce a YAML value into a list of strings (drop non-strings).""" + if not isinstance(raw, list): + return [] + return [str(item) for item in raw if isinstance(item, (str, int, float))] + + +def _parse_entry(path: Path) -> Optional[PluginCatalogEntry]: + """Parse and validate one catalog YAML file. + + Returns ``None`` (after logging a warning) on any validation failure — + the loader never raises for a bad entry. + """ + try: + data = yaml.safe_load(path.read_text(encoding="utf-8")) or {} + except Exception as exc: + logger.warning("Plugin catalog: failed to read %s: %s", path, exc) + return None + + if not isinstance(data, dict): + logger.warning("Plugin catalog: %s: entry must be a mapping", path) + return None + + name = str(data.get("name") or "") + if not _NAME_RE.match(name): + logger.warning( + "Plugin catalog: %s: invalid name %r (must match [a-z0-9_-]{1,64})", + path, name, + ) + return None + + repo = str(data.get("repo") or "") + if not repo.startswith("https://"): + logger.warning( + "Plugin catalog: %s: repo must be an https:// URL (got %r)", + path, repo, + ) + return None + + sha = str(data.get("sha") or "").strip().lower() + if not _SHA_RE.match(sha): + logger.warning( + "Plugin catalog: %s: sha must be a full 40-character hex commit " + "SHA (got %r)", path, data.get("sha"), + ) + return None + + tier = str(data.get("tier") or "community") + if tier not in CATALOG_TIERS: + logger.warning( + "Plugin catalog: %s: tier must be one of %s (got %r)", + path, "/".join(CATALOG_TIERS), tier, + ) + return None + + caps_raw = data.get("capabilities") or {} + if not isinstance(caps_raw, dict): + caps_raw = {} + capabilities = CatalogCapabilities( + provides_tools=_str_list(caps_raw.get("provides_tools")), + provides_hooks=_str_list(caps_raw.get("provides_hooks")), + provides_middleware=_str_list(caps_raw.get("provides_middleware")), + requires_env=_str_list(caps_raw.get("requires_env")), + ) + + return PluginCatalogEntry( + name=name, + repo=repo, + sha=sha, + description=str(data.get("description") or "").strip(), + maintainer=str(data.get("maintainer") or "").strip(), + tier=tier, + requires_hermes=str(data.get("requires_hermes") or "").strip(), + subdir=str(data.get("subdir") or "").strip(), + docs_url=str(data.get("docs_url") or "").strip(), + platforms=_str_list(data.get("platforms")), + capabilities=capabilities, + ) + + +def load_catalog() -> List[PluginCatalogEntry]: + """Return all valid catalog entries, sorted by name. + + Parses every ``*.yaml`` in the catalog dir except ``removed.yaml``. + Invalid entries are skipped with a logged warning; this function never + raises for a malformed entry. + """ + root = get_catalog_dir() + if not root.is_dir(): + return [] + entries: List[PluginCatalogEntry] = [] + for path in sorted(root.glob("*.yaml")): + if path.name == "removed.yaml": + continue + entry = _parse_entry(path) + if entry is not None: + entries.append(entry) + return entries + + +def get_catalog_entry(name: str) -> Optional[PluginCatalogEntry]: + """Look up a single catalog entry by name.""" + for entry in load_catalog(): + if entry.name == name: + return entry + return None + + +def search_catalog(query: str) -> List[PluginCatalogEntry]: + """Case-insensitive substring search over name, description, and + declared tools. An empty query returns the whole catalog.""" + entries = load_catalog() + q = (query or "").strip().lower() + if not q: + return entries + results: List[PluginCatalogEntry] = [] + for entry in entries: + haystacks = [entry.name, entry.description] + haystacks.extend(entry.capabilities.provides_tools) + if any(q in h.lower() for h in haystacks): + results.append(entry) + return results + + +# ─── Removed / blocklist ───────────────────────────────────────────────────── + + +def _normalize_repo(url: str) -> str: + """Normalize a repo URL for comparison (.git suffix and trailing slash + stripped, lowercased).""" + return url.strip().rstrip("/").removesuffix(".git").lower() + + +def load_removed_list() -> List[RemovedEntry]: + """Load ``plugin-catalog/removed.yaml`` (the ``removed:`` list). + + Missing or malformed files yield an empty list — never raises. + """ + path = get_catalog_dir() / "removed.yaml" + if not path.is_file(): + return [] + try: + data = yaml.safe_load(path.read_text(encoding="utf-8")) or {} + except Exception as exc: + logger.warning("Plugin catalog: failed to read %s: %s", path, exc) + return [] + raw_list = data.get("removed") if isinstance(data, dict) else None + if not isinstance(raw_list, list): + return [] + removed: List[RemovedEntry] = [] + for raw in raw_list: + if not isinstance(raw, dict): + continue + name = str(raw.get("name") or "") + if not name: + continue + removed.append( + RemovedEntry( + name=name, + repo=str(raw.get("repo") or ""), + reason=str(raw.get("reason") or ""), + date=str(raw.get("date") or ""), + ) + ) + return removed + + +def find_removed(name_or_repo: str) -> Optional[RemovedEntry]: + """Match *name_or_repo* against the removed blocklist. + + Matches by exact catalog name OR by repo URL (normalized — ``.git`` + suffix and trailing slashes are ignored). + """ + if not name_or_repo: + return None + candidate = name_or_repo.strip() + candidate_repo = _normalize_repo(candidate) + for entry in load_removed_list(): + if candidate == entry.name: + return entry + if entry.repo and candidate_repo == _normalize_repo(entry.repo): + return entry + return None + + +# ─── Human summaries ───────────────────────────────────────────────────────── + + +def entry_capability_summary(entry: PluginCatalogEntry) -> str: + """One-paragraph human summary of what an entry declares, shown at + install prompts so the user knows what they're granting.""" + caps = entry.capabilities + parts: List[str] = [] + if caps.provides_tools: + parts.append(f"registers tool(s): {', '.join(caps.provides_tools)}") + if caps.provides_hooks: + parts.append(f"hook(s): {', '.join(caps.provides_hooks)}") + if caps.provides_middleware: + parts.append(f"middleware: {', '.join(caps.provides_middleware)}") + if caps.requires_env: + parts.append(f"requires env var(s): {', '.join(caps.requires_env)}") + if not parts: + capability_text = "declares no tools, hooks, middleware, or env vars" + else: + capability_text = "; ".join(parts) + bits = [ + f"{entry.name} ({entry.tier}, maintained by {entry.maintainer})", + ] + if entry.description: + bits.append(entry.description) + bits.append(f"This plugin {capability_text}.") + if entry.platforms: + bits.append(f"Platforms: {', '.join(entry.platforms)}.") + if entry.requires_hermes: + bits.append(f"Requires Hermes {entry.requires_hermes}.") + return " ".join(bits) diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py index 6ca393fca53..1a5e93949fd 100644 --- a/hermes_cli/plugins.py +++ b/hermes_cli/plugins.py @@ -39,6 +39,7 @@ import importlib.util import inspect import logging import os +import re import sys import threading import types @@ -79,6 +80,93 @@ class PluginToolOverrideError(PermissionError): logger = logging.getLogger(__name__) +# --------------------------------------------------------------------------- +# Hermes version gate (manifest ``requires_hermes``) +# --------------------------------------------------------------------------- + +_VERSION_COMPARATOR_RE = re.compile(r"^\s*(>=|<=|==|!=|>|<)\s*(.+?)\s*$") + + +def _running_hermes_version() -> str: + """Return the running Hermes version string. + + Prefers installed package metadata (matches ``hermes_cli/main.py``'s + version reporting), falling back to ``hermes_cli.__version__`` for + source checkouts, then ``"0.0.0"`` as a last resort. + """ + try: + return importlib.metadata.version("hermes-agent") + except Exception: + pass + try: + from hermes_cli import __version__ + return __version__ + except Exception: + return "0.0.0" + + +def _version_tuple(v: str) -> Optional[tuple]: + """Parse ``major.minor.patch`` into a comparable tuple. + + Leading ``v`` and pre-release/build metadata (``-rc1``, ``+abc``) are + stripped; missing segments default to 0. Returns ``None`` when any + segment is non-numeric. + """ + s = str(v).strip().lstrip("v") + s = re.split(r"[-+]", s, 1)[0] + parts = s.split(".") + while len(parts) < 3: + parts.append("0") + try: + return tuple(int(p) for p in parts[:3]) + except ValueError: + return None + + +def _version_satisfies(spec: str, current: str) -> bool: + """Return True when *current* satisfies *spec*. + + *spec* supports ``>=``, ``>``, ``<=``, ``<``, ``==``, ``!=`` and + comma-separated combinations (all must hold). A bare version is treated + as ``>=``. Non-numeric version segments fall back to permissive True + (with a debug log) — no new dependency, so no full PEP 440 handling. + """ + if not spec or not spec.strip(): + return True + cur = _version_tuple(current) + if cur is None: + logger.debug( + "requires_hermes: unparseable running version %r — allowing", current, + ) + return True + for clause in spec.split(","): + clause = clause.strip() + if not clause: + continue + m = _VERSION_COMPARATOR_RE.match(clause) + if m: + op, target = m.group(1), m.group(2) + else: + op, target = ">=", clause + tgt = _version_tuple(target) + if tgt is None: + logger.debug( + "requires_hermes: unparseable version spec %r — allowing", clause, + ) + continue + ok = { + ">=": cur >= tgt, + "<=": cur <= tgt, + "==": cur == tgt, + "!=": cur != tgt, + ">": cur > tgt, + "<": cur < tgt, + }[op] + if not ok: + return False + return True + + # --------------------------------------------------------------------------- # Plugin developer debug logging # --------------------------------------------------------------------------- @@ -312,6 +400,16 @@ class PluginManifest: # category plugin at ``plugins/image_gen/openai/`` the key is # ``image_gen/openai``. When empty, falls back to ``name``. key: str = "" + # Minimum/exact Hermes version requirement, e.g. ``">=0.19"``. Empty = + # no requirement. Checked at load time; unsatisfied plugins are recorded + # with an error and skipped (no register() call, no traceback). + requires_hermes: str = "" + # Declared config keys from the manifest's ``config:`` section — a list + # of ``{key, prompt, type (str|bool|int), default, secret (bool)}`` + # dicts. secret=true values are prompted into ~/.hermes/.env; secret + # =false values live under ``plugins.entries..`` in + # config.yaml. Exposed to plugins via ``ctx.plugin_config``. + config_spec: List[Dict[str, Any]] = field(default_factory=list) @dataclass @@ -386,6 +484,40 @@ class PluginContext: except Exception: return "default" + # -- declared plugin config --------------------------------------------- + + @property + def plugin_config(self) -> Dict[str, Any]: + """Return this plugin's effective config values. + + Built from the manifest's ``config:`` spec defaults, overlaid with + whatever the operator set under ``plugins.entries.`` in + config.yaml (config.yaml wins on key collision). Secret keys + (``secret: true``) are stored in ``~/.hermes/.env`` instead and are + NOT surfaced here — read them via ``os.environ``. + """ + merged: Dict[str, Any] = {} + for spec in self.manifest.config_spec or []: + key = spec.get("key") + if not key: + continue + if spec.get("secret"): + continue # secrets live in .env, never in config.yaml + if "default" in spec: + merged[key] = spec.get("default") + try: + from hermes_cli.config import load_config + cfg = load_config() or {} + except Exception: + cfg = {} + plugin_id = self.manifest.key or self.manifest.name + entries = (cfg.get("plugins") or {}).get("entries") or {} + entry = entries.get(plugin_id) or {} + if isinstance(entry, dict): + for key, value in entry.items(): + merged[key] = value + return merged + # -- tool registration -------------------------------------------------- def register_tool( @@ -1632,6 +1764,23 @@ class PluginManager: "Parsed manifest: key=%s name=%s kind=%s source=%s path=%s", key, name, kind, source, plugin_dir, ) + raw_config = data.get("config", []) + config_spec: List[Dict[str, Any]] = [] + if isinstance(raw_config, list): + for item in raw_config: + if isinstance(item, dict) and item.get("key"): + config_spec.append(dict(item)) + else: + logger.warning( + "Plugin %s: ignoring invalid config entry %r " + "(must be a mapping with a 'key')", key, item, + ) + elif raw_config: + logger.warning( + "Plugin %s: 'config' must be a list of mappings; ignoring", + key, + ) + return PluginManifest( name=name, version=str(data.get("version", "")), @@ -1644,6 +1793,8 @@ class PluginManager: path=str(plugin_dir), kind=kind, key=key, + requires_hermes=str(data.get("requires_hermes") or "").strip(), + config_spec=config_spec, ) except Exception as exc: logger.warning( @@ -1748,6 +1899,24 @@ class PluginManager: def _load_plugin(self, manifest: PluginManifest) -> None: """Import a plugin module and call its ``register(ctx)`` function.""" loaded = LoadedPlugin(manifest=manifest) + + # requires_hermes gate — skip cleanly (no import, no traceback) when + # the running Hermes version doesn't satisfy the manifest spec. + if manifest.requires_hermes: + current = _running_hermes_version() + if not _version_satisfies(manifest.requires_hermes, current): + loaded.enabled = False + loaded.error = ( + f"requires hermes {manifest.requires_hermes}, " + f"running {current}" + ) + self._plugins[manifest.key or manifest.name] = loaded + logger.warning( + "Plugin '%s' skipped: %s", + manifest.key or manifest.name, loaded.error, + ) + return + logger.debug( "Loading plugin '%s' (source=%s, kind=%s, path=%s)", manifest.key or manifest.name, manifest.source, manifest.kind, manifest.path, diff --git a/hermes_cli/plugins_cmd.py b/hermes_cli/plugins_cmd.py index f5c57bb88f2..731f35c6efa 100644 --- a/hermes_cli/plugins_cmd.py +++ b/hermes_cli/plugins_cmd.py @@ -446,19 +446,61 @@ def _require_installed_plugin(name: str, plugins_dir: Path, console) -> Path: # --------------------------------------------------------------------------- -def _install_plugin_core(identifier: str, *, force: bool) -> tuple[Path, dict, str]: +def _raise_removed(removed) -> None: + """Raise PluginOperationError describing a blocklisted plugin.""" + detail = removed.reason or "no reason recorded" + if removed.date: + detail += f" (removed {removed.date})" + raise PluginOperationError( + f"Plugin '{removed.name}' was removed from the Hermes plugin " + f"catalog and is blocked from installation: {detail}" + ) + + +def _install_plugin_core( + identifier: str, + *, + force: bool, + ref: Optional[str] = None, + skip_removed_check: bool = False, +) -> tuple[Path, dict, str]: """Clone Git plugin into ``~/.hermes/plugins``. + ``ref`` — optional git commit SHA (or tag) checked out after clone. + When given, the clone is full-depth (no ``--depth 1``) so any commit is + reachable. + + Unless ``skip_removed_check`` is set, the identifier and the resolved + repo URL are checked against the plugin catalog's removed blocklist + (``plugin-catalog/removed.yaml``); a hit raises ``PluginOperationError`` + with the recorded reason and date. + Returns ``(target_dir, installed_manifest, canonical_name)``. Raises ``PluginOperationError`` on failure. """ import tempfile + if not skip_removed_check: + from hermes_cli.plugin_catalog import find_removed + + # Check the raw identifier first (catches catalog names before URL + # resolution), then the resolved repo URL below. + removed = find_removed(identifier) + if removed is not None: + _raise_removed(removed) + try: git_url, subdir = _resolve_git_url(identifier) except ValueError as e: raise PluginOperationError(str(e)) from e + if not skip_removed_check: + from hermes_cli.plugin_catalog import find_removed + + removed = find_removed(git_url) + if removed is not None: + _raise_removed(removed) + plugins_dir = _plugins_dir() with tempfile.TemporaryDirectory() as tmp: @@ -468,9 +510,15 @@ def _install_plugin_core(identifier: str, *, force: bool) -> tuple[Path, dict, s if not git_exe: raise PluginOperationError("git is not installed or not in PATH.") + clone_cmd = [git_exe, "clone"] + if ref is None: + # Fast path — only the tip is needed. + clone_cmd += ["--depth", "1"] + clone_cmd += [git_url, str(tmp_clone)] + try: result = subprocess.run( - [git_exe, "clone", "--depth", "1", git_url, str(tmp_clone)], + clone_cmd, capture_output=True, text=True, timeout=60, @@ -488,6 +536,24 @@ def _install_plugin_core(identifier: str, *, force: bool) -> tuple[Path, dict, s err = (result.stderr or result.stdout or "").strip() raise PluginOperationError(f"Git clone failed:\n{err}") + if ref is not None: + try: + checkout = subprocess.run( + [git_exe, "-C", str(tmp_clone), "checkout", ref], + capture_output=True, + text=True, + timeout=60, + ) + except subprocess.TimeoutExpired as e: + raise PluginOperationError( + f"Git checkout of ref '{ref}' timed out after 60 seconds.", + ) from e + if checkout.returncode != 0: + err = (checkout.stderr or checkout.stdout or "").strip() + raise PluginOperationError( + f"Git checkout of ref '{ref}' failed:\n{err}" + ) + # Resolve the directory within the clone that holds the plugin. if subdir: tmp_target = _resolve_subdir_within(tmp_clone, subdir) diff --git a/plugin-catalog/README.md b/plugin-catalog/README.md new file mode 100644 index 00000000000..6a44c3a6e68 --- /dev/null +++ b/plugin-catalog/README.md @@ -0,0 +1,60 @@ +# Hermes Plugin Catalog + +Curated, Nous-approved Hermes plugins. Each YAML file in this directory +(except `removed.yaml`) is one catalog entry, discoverable via +`hermes plugins catalog` / `hermes plugins search` and installable with +`hermes plugins install `. + +## Admission policy + +Presence in this directory **is** the trust signal. The rules that keep it +meaningful: + +1. **Human-merged gate.** Entries are added *only* via a PR to the + `hermes-agent` repository, reviewed and merged by a maintainer. There is + no self-serve registry, no automated ingestion. +2. **Exact SHA pins are mandatory.** Every entry pins a full 40-character + commit SHA. Branches, tags, and short SHAs are rejected by the loader. + Installs clone the repository and check out exactly that commit. +3. **Pin maturity.** The pinned release should be **at least 2 weeks old** + at pin time, mirroring the supply-chain policy used for `optional-mcps/` + and pyproject dependencies. This gives the community time to notice a + compromised release before Hermes ships a pointer to it. +4. **SHA bumps are new PRs.** Updating an entry's pin is a new PR whose diff + (old SHA → new SHA) is re-reviewed like any other change — reviewers are + expected to look at the upstream commit range being adopted. +5. **Owner-or-major-contributor submissions only.** An entry may only be + submitted by the plugin repository's owner or a major contributor to it. + Drive-by submissions of third-party repos are declined. +6. **Declared capabilities must match reality.** The `capabilities:` block + (tools, hooks, middleware, env vars) must match what the plugin actually + registers at the pinned commit. Validation fails the entry otherwise — + undeclared capability creep is treated as a security issue. + +## Entry schema + +```yaml +name: example-plugin # [a-z0-9_-]{1,64}, the catalog key +repo: https://github.com/owner/repo # https:// only +sha: <40-hex commit sha> # mandatory exact pin +subdir: "" # optional path within the repo +description: One-line description. +maintainer: OwnerName +tier: official # official | community (default community) +requires_hermes: ">=0.19" # optional +docs_url: "" # optional +platforms: [] # optional, e.g. [linux, macos]; empty = all +capabilities: + provides_tools: [] + provides_hooks: [] + provides_middleware: [] + requires_env: [] +``` + +## removed.yaml — the blocklist + +When an entry is pulled from the catalog for security or policy reasons, it +is recorded in `removed.yaml` with a reason and date. The installer refuses +to install anything matching a removed entry's name or repo URL, so a +malicious plugin cannot be re-installed from a stale identifier after +removal. Removals, like additions, land via reviewed PRs. diff --git a/plugin-catalog/example-plugin.yaml b/plugin-catalog/example-plugin.yaml new file mode 100644 index 00000000000..236b6fa9a77 --- /dev/null +++ b/plugin-catalog/example-plugin.yaml @@ -0,0 +1,14 @@ +name: example-plugin +repo: https://github.com/NousResearch/hermes-example-plugins +sha: 38fe0fb53eff98d477f807432e965429e665ca33 +subdir: "" +description: Reference example plugins for the Hermes plugin system. +maintainer: NousResearch +tier: official +docs_url: "" +platforms: [] +capabilities: + provides_tools: [] + provides_hooks: [] + provides_middleware: [] + requires_env: [] diff --git a/plugin-catalog/removed.yaml b/plugin-catalog/removed.yaml new file mode 100644 index 00000000000..75c6ab8c6cf --- /dev/null +++ b/plugin-catalog/removed.yaml @@ -0,0 +1,6 @@ +# Blocklist for plugins pulled from the catalog for security or policy +# reasons. The installer refuses to install anything whose name or repo URL +# matches an entry here (unless the caller explicitly bypasses the check). +# Each entry: {name, repo, reason, date}. Removals land via reviewed PRs, +# same as additions. +removed: [] diff --git a/tests/hermes_cli/test_plugin_catalog.py b/tests/hermes_cli/test_plugin_catalog.py new file mode 100644 index 00000000000..18096bfb5e8 --- /dev/null +++ b/tests/hermes_cli/test_plugin_catalog.py @@ -0,0 +1,594 @@ +"""Tests for the Hermes plugin catalog (hermes_cli.plugin_catalog) and the +catalog-driven install/manifest extensions in plugins_cmd.py / plugins.py.""" + +from __future__ import annotations + +import os +import subprocess +from pathlib import Path + +import pytest +import yaml + +from hermes_cli.plugin_catalog import ( + CATALOG_TIERS, + PluginCatalogEntry, + RemovedEntry, + entry_capability_summary, + find_removed, + get_catalog_dir, + get_catalog_entry, + load_catalog, + load_removed_list, + search_catalog, +) + + +VALID_SHA = "38fe0fb53eff98d477f807432e965429e665ca33" + + +# ── Helpers ──────────────────────────────────────────────────────────────── + + +def _write_entry(catalog_dir: Path, name: str, **overrides) -> Path: + """Write a minimal valid catalog entry yaml, applying overrides.""" + data = { + "name": name, + "repo": f"https://github.com/example/{name}", + "sha": VALID_SHA, + "description": f"Test entry {name}.", + "maintainer": "Example", + } + data.update(overrides) + catalog_dir.mkdir(parents=True, exist_ok=True) + path = catalog_dir / f"{name}.yaml" + path.write_text(yaml.safe_dump(data), encoding="utf-8") + return path + + +def _write_removed(catalog_dir: Path, removed: list) -> Path: + catalog_dir.mkdir(parents=True, exist_ok=True) + path = catalog_dir / "removed.yaml" + path.write_text(yaml.safe_dump({"removed": removed}), encoding="utf-8") + return path + + +@pytest.fixture() +def catalog_dir(tmp_path, monkeypatch): + d = tmp_path / "catalog" + d.mkdir() + monkeypatch.setenv("HERMES_PLUGIN_CATALOG_DIR", str(d)) + return d + + +# ── get_catalog_dir ──────────────────────────────────────────────────────── + + +class TestGetCatalogDir: + def test_env_override_wins(self, catalog_dir): + assert get_catalog_dir() == catalog_dir + + def test_default_is_repo_plugin_catalog(self, monkeypatch): + monkeypatch.delenv("HERMES_PLUGIN_CATALOG_DIR", raising=False) + d = get_catalog_dir() + assert d.name == "plugin-catalog" + + +# ── load_catalog ─────────────────────────────────────────────────────────── + + +class TestLoadCatalog: + def test_valid_entry_parses(self, catalog_dir): + _write_entry( + catalog_dir, + "my-plugin", + tier="official", + requires_hermes=">=0.19", + subdir="plugins/my-plugin", + docs_url="https://example.com/docs", + platforms=["linux"], + capabilities={ + "provides_tools": ["my_tool"], + "provides_hooks": ["on_start"], + "provides_middleware": ["llm_request"], + "requires_env": ["MY_API_KEY"], + }, + ) + entries = load_catalog() + assert len(entries) == 1 + e = entries[0] + assert isinstance(e, PluginCatalogEntry) + assert e.name == "my-plugin" + assert e.repo == "https://github.com/example/my-plugin" + assert e.sha == VALID_SHA + assert e.tier == "official" + assert e.requires_hermes == ">=0.19" + assert e.subdir == "plugins/my-plugin" + assert e.docs_url == "https://example.com/docs" + assert e.platforms == ["linux"] + assert e.capabilities.provides_tools == ["my_tool"] + assert e.capabilities.provides_hooks == ["on_start"] + assert e.capabilities.provides_middleware == ["llm_request"] + assert e.capabilities.requires_env == ["MY_API_KEY"] + + def test_tier_defaults_to_community(self, catalog_dir): + _write_entry(catalog_dir, "no-tier") + (entry,) = load_catalog() + assert entry.tier == "community" + assert entry.tier in CATALOG_TIERS + + def test_bad_sha_rejected(self, catalog_dir, caplog): + _write_entry(catalog_dir, "bad-sha", sha="main") + _write_entry(catalog_dir, "short-sha", sha="38fe0fb") + _write_entry(catalog_dir, "good", sha=VALID_SHA) + with caplog.at_level("WARNING"): + entries = load_catalog() + assert [e.name for e in entries] == ["good"] + + def test_bad_name_rejected(self, catalog_dir, caplog): + _write_entry(catalog_dir, "BadName") + _write_entry(catalog_dir, "has spaces") + with caplog.at_level("WARNING"): + entries = load_catalog() + assert entries == [] + + def test_non_https_repo_rejected(self, catalog_dir, caplog): + _write_entry(catalog_dir, "sshrepo", repo="git@github.com:x/y.git") + with caplog.at_level("WARNING"): + entries = load_catalog() + assert entries == [] + + def test_invalid_tier_rejected(self, catalog_dir, caplog): + _write_entry(catalog_dir, "weird-tier", tier="platinum") + with caplog.at_level("WARNING"): + entries = load_catalog() + assert entries == [] + + def test_removed_yaml_is_not_an_entry(self, catalog_dir): + _write_entry(catalog_dir, "real-entry") + _write_removed(catalog_dir, []) + entries = load_catalog() + assert [e.name for e in entries] == ["real-entry"] + + def test_unparseable_yaml_skipped_without_raising(self, catalog_dir, caplog): + (catalog_dir / "broken.yaml").write_text( + "name: [unclosed", encoding="utf-8" + ) + _write_entry(catalog_dir, "ok-entry") + with caplog.at_level("WARNING"): + entries = load_catalog() + assert [e.name for e in entries] == ["ok-entry"] + + def test_missing_dir_returns_empty(self, tmp_path, monkeypatch): + monkeypatch.setenv( + "HERMES_PLUGIN_CATALOG_DIR", str(tmp_path / "does-not-exist") + ) + assert load_catalog() == [] + + +# ── get_catalog_entry / search_catalog ───────────────────────────────────── + + +class TestLookupAndSearch: + def test_get_catalog_entry_by_name(self, catalog_dir): + _write_entry(catalog_dir, "alpha") + _write_entry(catalog_dir, "beta") + entry = get_catalog_entry("beta") + assert entry is not None and entry.name == "beta" + assert get_catalog_entry("nope") is None + + def test_search_matches_name_case_insensitive(self, catalog_dir): + _write_entry(catalog_dir, "weather-tools") + _write_entry(catalog_dir, "other") + results = search_catalog("WEATHER") + assert [e.name for e in results] == ["weather-tools"] + + def test_search_matches_description(self, catalog_dir): + _write_entry(catalog_dir, "abc", description="Fetches Stock Quotes.") + results = search_catalog("stock") + assert [e.name for e in results] == ["abc"] + + def test_search_matches_declared_tools(self, catalog_dir): + _write_entry( + catalog_dir, + "toolful", + capabilities={"provides_tools": ["get_forecast"]}, + ) + _write_entry(catalog_dir, "toolless") + results = search_catalog("Forecast") + assert [e.name for e in results] == ["toolful"] + + def test_empty_query_returns_all(self, catalog_dir): + _write_entry(catalog_dir, "one") + _write_entry(catalog_dir, "two") + assert len(search_catalog("")) == 2 + + +# ── removed list ─────────────────────────────────────────────────────────── + + +class TestRemovedList: + def test_load_removed_list(self, catalog_dir): + _write_removed( + catalog_dir, + [ + { + "name": "evil-plugin", + "repo": "https://github.com/evil/evil-plugin", + "reason": "Exfiltrated env vars", + "date": "2026-07-02", + } + ], + ) + removed = load_removed_list() + assert len(removed) == 1 + r = removed[0] + assert isinstance(r, RemovedEntry) + assert r.name == "evil-plugin" + assert r.reason == "Exfiltrated env vars" + assert r.date == "2026-07-02" + + def test_missing_removed_yaml_returns_empty(self, catalog_dir): + assert load_removed_list() == [] + assert find_removed("anything") is None + + def test_find_removed_by_name(self, catalog_dir): + _write_removed(catalog_dir, [{"name": "evil-plugin", "reason": "bad"}]) + hit = find_removed("evil-plugin") + assert hit is not None and hit.reason == "bad" + + def test_find_removed_by_repo_url_with_and_without_git_suffix( + self, catalog_dir + ): + _write_removed( + catalog_dir, + [ + { + "name": "evil-plugin", + "repo": "https://github.com/evil/evil-plugin", + "reason": "bad", + } + ], + ) + assert find_removed("https://github.com/evil/evil-plugin") is not None + assert find_removed("https://github.com/evil/evil-plugin.git") is not None + assert find_removed("https://github.com/good/fine.git") is None + + +# ── entry_capability_summary ─────────────────────────────────────────────── + + +class TestCapabilitySummary: + def test_summary_contains_declared_capabilities(self): + entry = PluginCatalogEntry( + name="cap-plugin", + repo="https://github.com/example/cap-plugin", + sha=VALID_SHA, + description="Does capable things.", + maintainer="Example", + ) + entry.capabilities.provides_tools = ["tool_a", "tool_b"] + entry.capabilities.provides_hooks = ["session_start"] + entry.capabilities.requires_env = ["CAP_API_KEY"] + summary = entry_capability_summary(entry) + assert "tool_a" in summary + assert "tool_b" in summary + assert "session_start" in summary + assert "CAP_API_KEY" in summary + + def test_summary_for_empty_capabilities_mentions_none(self): + entry = PluginCatalogEntry( + name="plain", + repo="https://github.com/example/plain", + sha=VALID_SHA, + description="Plain.", + maintainer="Example", + ) + summary = entry_capability_summary(entry) + assert summary # non-empty human text + + +# ── shipped catalog seed ─────────────────────────────────────────────────── + + +class TestShippedCatalog: + def test_shipped_catalog_entries_are_valid(self, monkeypatch): + """Every yaml shipped in /plugin-catalog must load cleanly.""" + monkeypatch.delenv("HERMES_PLUGIN_CATALOG_DIR", raising=False) + shipped = get_catalog_dir() + yaml_files = [ + p for p in shipped.glob("*.yaml") if p.name != "removed.yaml" + ] + entries = load_catalog() + assert len(entries) == len(yaml_files) + # removed.yaml must exist and parse + assert (shipped / "removed.yaml").exists() + load_removed_list() + + +# ── _version_satisfies ───────────────────────────────────────────────────── + + +class TestVersionSatisfies: + @pytest.fixture(autouse=True) + def _import(self): + from hermes_cli.plugins import _version_satisfies + + self.satisfies = _version_satisfies + + def test_ge(self): + assert self.satisfies(">=0.19", "0.19.0") is True + assert self.satisfies(">=0.19", "0.20.1") is True + assert self.satisfies(">=0.19", "0.18.2") is False + + def test_gt_lt_le(self): + assert self.satisfies(">0.19", "0.19.1") is True + assert self.satisfies(">0.19", "0.19.0") is False + assert self.satisfies("<1.0", "0.19.0") is True + assert self.satisfies("<=0.19.0", "0.19.0") is True + + def test_eq_ne(self): + assert self.satisfies("==0.19.0", "0.19.0") is True + assert self.satisfies("==0.19.0", "0.19.1") is False + assert self.satisfies("!=0.19.0", "0.19.1") is True + assert self.satisfies("!=0.19.0", "0.19.0") is False + + def test_comma_separated_all_must_hold(self): + assert self.satisfies(">=0.10, <1.0", "0.19.0") is True + assert self.satisfies(">=0.10, <0.15", "0.19.0") is False + + def test_bare_version_treated_as_ge(self): + assert self.satisfies("0.10", "0.19.0") is True + assert self.satisfies("999", "0.19.0") is False + + def test_empty_spec_is_satisfied(self): + assert self.satisfies("", "0.19.0") is True + + def test_non_numeric_segments_fall_back_permissive(self): + assert self.satisfies(">=abc.def", "0.19.0") is True + assert self.satisfies(">=0.19", "unknown") is True + + +# ── requires_hermes manifest gate ────────────────────────────────────────── + + +def _make_plugin(base: Path, name: str, *, manifest_extra: dict | None = None, + register_body: str = "pass", enable: bool = True) -> Path: + """Create a plugin dir under /plugins and opt it in.""" + plugin_dir = base / name + plugin_dir.mkdir(parents=True, exist_ok=True) + manifest = {"name": name, "version": "0.1.0", "description": name} + if manifest_extra: + manifest.update(manifest_extra) + (plugin_dir / "plugin.yaml").write_text(yaml.safe_dump(manifest)) + (plugin_dir / "__init__.py").write_text( + f"def register(ctx):\n {register_body}\n" + ) + if enable: + hermes_home = Path(os.environ["HERMES_HOME"]) + cfg_path = hermes_home / "config.yaml" + cfg: dict = {} + if cfg_path.exists(): + cfg = yaml.safe_load(cfg_path.read_text()) or {} + cfg.setdefault("plugins", {}).setdefault("enabled", []).append(name) + cfg_path.write_text(yaml.safe_dump(cfg)) + return plugin_dir + + +class TestRequiresHermesGate: + def test_unsatisfied_requires_hermes_skips_load(self, monkeypatch): + from hermes_cli.plugins import PluginManager + + hermes_home = Path(os.environ["HERMES_HOME"]) + plugins_dir = hermes_home / "plugins" + _make_plugin( + plugins_dir, "future_plugin", + manifest_extra={"requires_hermes": ">=999.0"}, + ) + mgr = PluginManager() + mgr.discover_and_load() + loaded = mgr._plugins["future_plugin"] + assert loaded.enabled is False + assert loaded.error is not None + assert "requires hermes" in loaded.error + assert ">=999.0" in loaded.error + assert loaded.module is None # register() never ran + + def test_satisfied_requires_hermes_loads_normally(self, monkeypatch): + from hermes_cli.plugins import PluginManager + + hermes_home = Path(os.environ["HERMES_HOME"]) + plugins_dir = hermes_home / "plugins" + _make_plugin( + plugins_dir, "old_ok_plugin", + manifest_extra={"requires_hermes": ">=0.1"}, + ) + mgr = PluginManager() + mgr.discover_and_load() + loaded = mgr._plugins["old_ok_plugin"] + assert loaded.enabled is True + assert loaded.error is None + + def test_requires_hermes_parsed_onto_manifest(self): + from hermes_cli.plugins import PluginManager + + hermes_home = Path(os.environ["HERMES_HOME"]) + plugins_dir = hermes_home / "plugins" + _make_plugin( + plugins_dir, "spec_plugin", + manifest_extra={"requires_hermes": ">=0.19"}, + enable=False, + ) + mgr = PluginManager() + mgr.discover_and_load() + assert mgr._plugins["spec_plugin"].manifest.requires_hermes == ">=0.19" + + +# ── config: spec parsing + ctx.plugin_config ─────────────────────────────── + + +class TestPluginConfig: + def test_config_spec_parsed_onto_manifest(self): + from hermes_cli.plugins import PluginManager + + hermes_home = Path(os.environ["HERMES_HOME"]) + plugins_dir = hermes_home / "plugins" + spec = [ + {"key": "api_url", "prompt": "API URL", "type": "str", + "default": "https://api.example.com", "secret": False}, + {"key": "token", "prompt": "Token", "type": "str", "secret": True}, + ] + _make_plugin( + plugins_dir, "cfg_plugin", + manifest_extra={"config": spec}, + enable=False, + ) + mgr = PluginManager() + mgr.discover_and_load() + manifest = mgr._plugins["cfg_plugin"].manifest + assert isinstance(manifest.config_spec, list) + assert manifest.config_spec[0]["key"] == "api_url" + assert manifest.config_spec[1]["secret"] is True + + def test_plugin_config_merges_defaults_under_config_entries(self): + from hermes_cli.plugins import PluginContext, PluginManifest, PluginManager + + hermes_home = Path(os.environ["HERMES_HOME"]) + cfg_path = hermes_home / "config.yaml" + cfg_path.write_text(yaml.safe_dump({ + "plugins": {"entries": {"merge_plugin": {"api_url": "https://override"}}} + })) + + manifest = PluginManifest( + name="merge_plugin", + key="merge_plugin", + config_spec=[ + {"key": "api_url", "default": "https://default"}, + {"key": "retries", "type": "int", "default": 3}, + ], + ) + ctx = PluginContext(manifest, PluginManager()) + cfg = ctx.plugin_config + assert cfg["api_url"] == "https://override" # config.yaml wins + assert cfg["retries"] == 3 # default fills the gap + + def test_plugin_config_empty_without_spec_or_entries(self): + from hermes_cli.plugins import PluginContext, PluginManifest, PluginManager + + manifest = PluginManifest(name="bare_plugin", key="bare_plugin") + ctx = PluginContext(manifest, PluginManager()) + assert ctx.plugin_config == {} + + +# ── _install_plugin_core: ref checkout + removed blocklist ──────────────── + + +def _make_git_repo(tmp_path: Path) -> tuple[Path, str, str]: + """Create a local git repo with two commits; return (path, sha1, sha2).""" + repo = tmp_path / "src-repo" + repo.mkdir() + + def git(*args): + subprocess.run( + ["git", *args], cwd=repo, check=True, capture_output=True, text=True, + env={**os.environ, + "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@t", + "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@t"}, + ) + + git("init", "-b", "main") + (repo / "plugin.yaml").write_text( + yaml.safe_dump({"name": "refplugin", "version": "1"}) + ) + (repo / "__init__.py").write_text("def register(ctx):\n pass\n") + (repo / "marker.txt").write_text("first\n") + git("add", "-A") + git("commit", "-m", "first") + sha1 = subprocess.run( + ["git", "rev-parse", "HEAD"], cwd=repo, check=True, + capture_output=True, text=True, + ).stdout.strip() + (repo / "marker.txt").write_text("second\n") + git("add", "-A") + git("commit", "-m", "second") + sha2 = subprocess.run( + ["git", "rev-parse", "HEAD"], cwd=repo, check=True, + capture_output=True, text=True, + ).stdout.strip() + return repo, sha1, sha2 + + +class TestInstallPluginCore: + def test_ref_checkout_installs_pinned_commit(self, tmp_path, catalog_dir): + from hermes_cli.plugins_cmd import _install_plugin_core + + repo, sha1, _sha2 = _make_git_repo(tmp_path) + target, manifest, name = _install_plugin_core( + f"file://{repo}", force=False, ref=sha1 + ) + assert name == "refplugin" + assert (target / "marker.txt").read_text() == "first\n" + + def test_default_install_gets_head(self, tmp_path, catalog_dir): + from hermes_cli.plugins_cmd import _install_plugin_core + + repo, _sha1, _sha2 = _make_git_repo(tmp_path) + target, _manifest, _name = _install_plugin_core( + f"file://{repo}", force=False + ) + assert (target / "marker.txt").read_text() == "second\n" + + def test_bad_ref_raises(self, tmp_path, catalog_dir): + from hermes_cli.plugins_cmd import PluginOperationError, _install_plugin_core + + repo, _sha1, _sha2 = _make_git_repo(tmp_path) + with pytest.raises(PluginOperationError): + _install_plugin_core( + f"file://{repo}", force=False, + ref="0000000000000000000000000000000000000000", + ) + + def test_removed_repo_blocked(self, tmp_path, catalog_dir): + from hermes_cli.plugins_cmd import PluginOperationError, _install_plugin_core + + repo, _sha1, _sha2 = _make_git_repo(tmp_path) + _write_removed( + catalog_dir, + [{ + "name": "refplugin", + "repo": f"file://{repo}", + "reason": "exfiltrated env vars", + "date": "2026-07-02", + }], + ) + with pytest.raises(PluginOperationError, match="exfiltrated env vars"): + _install_plugin_core(f"file://{repo}", force=False) + + def test_removed_identifier_blocked_by_name(self, tmp_path, catalog_dir): + from hermes_cli.plugins_cmd import PluginOperationError, _install_plugin_core + + _write_removed( + catalog_dir, + [{"name": "evil-plugin", "reason": "malware", "date": "2026-01-01"}], + ) + with pytest.raises(PluginOperationError, match="malware"): + _install_plugin_core("evil-plugin", force=False) + + def test_skip_removed_check_bypasses_block(self, tmp_path, catalog_dir): + from hermes_cli.plugins_cmd import _install_plugin_core + + repo, _sha1, _sha2 = _make_git_repo(tmp_path) + _write_removed( + catalog_dir, + [{ + "name": "refplugin", + "repo": f"file://{repo}", + "reason": "bad", + "date": "2026-07-02", + }], + ) + target, _manifest, name = _install_plugin_core( + f"file://{repo}", force=False, skip_removed_check=True + ) + assert name == "refplugin" + assert target.exists()