"""``hermes plugins validate`` — admission checks for a plugin directory. This is the command the plugin-catalog admission CI (and the ``.github/actions/plugin-validate`` composite action) runs against a candidate plugin. It performs static manifest checks plus a subprocess-isolated capability probe: the plugin is imported and its ``register(ctx)`` called against a minimal recording stub context in a scratch child process (with a throwaway ``HERMES_HOME``), so a crashing or malicious plugin cannot take down the CLI, and the *actually registered* tools/hooks/middleware are compared against the manifest's declared ``provides_*`` lists. """ from __future__ import annotations import json import os import re import subprocess import sys import tempfile from dataclasses import dataclass, field from pathlib import Path from typing import Any, Dict, List, Optional, Tuple from hermes_cli.plugin_validate_core_override import check_core_override from hermes_cli.plugin_validate_desktop import check_desktop_surface from hermes_cli.plugin_validate_locales import check_language_packs from hermes_cli.plugins_manifest import _CONFIG_SCHEMA_TYPES _UPPER_SNAKE_RE = re.compile(r"^[A-Z][A-Z0-9_]*$") # Admission accepts exactly the ``config_schema`` types the loader type-checks at load time (and the # Desktop settings renderer keys its field table on) — a private copy drifted and rejected ``secret``. _CONFIG_TYPES = frozenset(_CONFIG_SCHEMA_TYPES) _PROBE_TIMEOUT = 30 _PROBE_SENTINEL = "HERMES_VALIDATE_JSON:" @dataclass class ValidationReport: """Result of validating one plugin directory.""" checks: List[Tuple[str, bool, str]] = field(default_factory=list) warnings: List[str] = field(default_factory=list) @property def failures(self) -> List[str]: return [detail or name for name, ok, detail in self.checks if not ok] @property def ok(self) -> bool: return all(ok for _name, ok, _detail in self.checks) @property def exit_code(self) -> int: return 0 if self.ok else 1 def add(self, name: str, ok: bool, detail: str = "") -> None: self.checks.append((name, ok, detail)) def warn(self, message: str) -> None: self.warnings.append(message) def to_dict(self) -> Dict[str, Any]: return { "ok": self.ok, "checks": [ {"name": name, "ok": ok, "detail": detail} for name, ok, detail in self.checks ], "warnings": list(self.warnings), } # ─── Static checks ─────────────────────────────────────────────────────────── def _requires_hermes_spec_valid(spec: str) -> bool: """Strictly validate a ``requires_hermes`` spec. Unlike :func:`hermes_cli.plugins_manifest.version_satisfies` (permissive at load time), validation REJECTS clauses whose version segment doesn't parse — a typo'd spec should fail admission, not silently gate nothing. """ from hermes_cli.plugins_manifest import _VERSION_COMPARATOR_RE, _version_tuple for clause in spec.split(","): clause = clause.strip() if not clause: continue m = _VERSION_COMPARATOR_RE.match(clause) target = m.group(2) if m else clause if _version_tuple(target) is None: return False return True def _check_manifest_fields(report: ValidationReport, manifest: dict) -> None: missing = [ f for f in ("name", "version", "description") if not manifest.get(f) ] if missing: report.add( "manifest fields", False, f"plugin.yaml missing required field(s): {', '.join(missing)}", ) else: report.add("manifest fields", True, "name, version, description present") def _check_requires_hermes(report: ValidationReport, manifest: dict) -> None: spec = str(manifest.get("requires_hermes") or "").strip() if not spec: report.add("requires_hermes", True, "not declared") return if _requires_hermes_spec_valid(spec): report.add("requires_hermes", True, f"spec {spec!r} parses") else: report.add( "requires_hermes", False, f"requires_hermes spec {spec!r} does not parse " "(expected e.g. \">=0.19\" or \">=0.19, <1.0\")", ) def _check_config_spec(report: ValidationReport, manifest: dict) -> None: """Validate the manifest ``config_schema`` mapping (#64165 format).""" raw = manifest.get("config_schema") if raw in (None, [], {}): report.add("config schema", True, "not declared") return problems: List[str] = [] if not isinstance(raw, dict): problems.append("config_schema: must be a mapping of key -> spec") else: for skey, spec in raw.items(): if not isinstance(spec, dict): problems.append( f"config_schema.{skey}: must be a mapping (e.g. {{type: str}})" ) continue typ = spec.get("type") if typ is not None and str(typ).lower() not in _CONFIG_TYPES: problems.append( f"config_schema.{skey}: type must be one of " f"{'/'.join(sorted(_CONFIG_TYPES))}" ) required = spec.get("required") if required is not None and not isinstance(required, bool): problems.append( f"config_schema.{skey}: required must be a boolean" ) if problems: report.add("config schema", False, "; ".join(problems)) else: report.add("config schema", True, "shape valid") def _check_requires_env(report: ValidationReport, manifest: dict) -> None: raw = manifest.get("requires_env") or [] problems: List[str] = [] if not isinstance(raw, list): problems.append("requires_env: must be a list") raw = [] for i, entry in enumerate(raw): if isinstance(entry, str): name = entry elif isinstance(entry, dict): name = str(entry.get("name") or "") else: problems.append(f"requires_env[{i}]: must be a string or mapping") continue if not _UPPER_SNAKE_RE.match(name): problems.append( f"requires_env[{i}]: {name!r} is not UPPER_SNAKE_CASE" ) if problems: report.add("requires_env", False, "; ".join(problems)) else: report.add("requires_env", True, "all entries UPPER_SNAKE") # ─── Capability probe (subprocess-isolated) ────────────────────────────────── # Self-contained harness run in a scratch child process. Imports the plugin # module using the same file-location mechanics PluginManager uses, calls # register() against a recording stub ctx, and prints a sentinel-prefixed # JSON line of what was actually registered. Deliberately imports NOTHING # from hermes so a hostile plugin only sees a bare interpreter — the one # exception is `providers` for `kind: model-provider`, whose contract IS # calling providers.register_provider at import. _PROBE_SCRIPT = r""" import importlib.util import json import sys plugin_dir = sys.argv[1] sentinel = sys.argv[2] options = json.loads(sys.argv[3]) # Public method names of the real PluginContext, computed by the parent so the # stub's attribute surface cannot drift from the class plugins run against. context_methods = set(options["context_methods"]) provider_kind = options["kind"] == "model-provider" recorded = {"tools": [], "hooks": [], "middleware": [], "commands": [], "providers": []} class RecordingContext: plugin_config = {} profile_name = "default" plugin_id = "hermes_validate_probe_plugin" def register_tool(self, name, *args, **kwargs): recorded["tools"].append(str(name)) def register_hook(self, hook_name, callback): recorded["hooks"].append(str(hook_name)) def register_middleware(self, kind, callback): recorded["middleware"].append(str(kind)) def register_command(self, name, *args, **kwargs): recorded["commands"].append(str(name)) def register_cli_command(self, name, *args, **kwargs): recorded["commands"].append(str(name)) def get_config(self, key, default=None): # Mirrors PluginContext.get_config with no config on disk: the DEFAULT, never None — # plugins do `int(ctx.get_config("timeout", 180))` in register(). return default def __getattr__(self, name): # Any other REAL registration surface (platforms, providers, skills, # context engines, ...) is accepted as a no-op — the probe only audits # the declared-capability categories. Names the real PluginContext does # not have raise AttributeError exactly like it would; handing back a # callable made `getattr(ctx, "profile_path", None)` truthy and crashed # register() in the probe alone. if name in context_methods: def _noop(*args, **kwargs): return None return _noop raise AttributeError(name) def emit(payload): print(sentinel + json.dumps(payload)) if provider_kind: # `kind: model-provider` plugins register at import via # providers.register_provider(ProviderProfile) — the PluginManager never # calls a register(ctx) on them (plugins_discovery skips the kind), so the # probe records that call instead of demanding an entry point that would # be dead code. try: import providers as _providers except Exception as exc: emit({"error": "model-provider probe could not import providers: %s" % exc}) sys.exit(0) _real_register_provider = _providers.register_provider def _recording_register_provider(profile): recorded["providers"].append(str(getattr(profile, "name", profile))) return _real_register_provider(profile) _providers.register_provider = _recording_register_provider try: spec = importlib.util.spec_from_file_location( "hermes_validate_probe_plugin", plugin_dir + "/__init__.py", submodule_search_locations=[plugin_dir], ) module = importlib.util.module_from_spec(spec) module.__path__ = [plugin_dir] sys.modules[spec.name] = module spec.loader.exec_module(module) except Exception as exc: emit({"error": "import failed: %s" % exc}) sys.exit(0) if provider_kind: if not recorded["providers"]: emit({"error": "model-provider plugin registered no ProviderProfile at import"}) else: emit(recorded) sys.exit(0) register = getattr(module, "register", None) if register is None: emit({"error": "no register() function"}) sys.exit(0) try: register(RecordingContext()) except Exception as exc: emit({"error": "register() raised: %s" % exc}) sys.exit(0) emit(recorded) """ def _probe_options(manifest: dict) -> dict: from hermes_cli.plugins import PluginContext return { "kind": str(manifest.get("kind") or ""), "context_methods": sorted( n for n in dir(PluginContext) if not n.startswith("_") and callable(getattr(PluginContext, n)) ), } def _run_capability_probe(plugin_dir: Path, manifest: dict) -> Tuple[Optional[dict], str]: """Run the recording probe in a scratch subprocess. Returns ``(recorded, error)`` — exactly one is meaningful: *recorded* is the ``{tools, hooks, middleware, commands, providers}`` dict on success, and *error* is a human-readable failure description otherwise. """ with tempfile.TemporaryDirectory(prefix="hermes-validate-") as scratch: env = dict(os.environ) env["HERMES_HOME"] = scratch try: result = subprocess.run( [ sys.executable, "-c", _PROBE_SCRIPT, str(plugin_dir), _PROBE_SENTINEL, json.dumps(_probe_options(manifest)), ], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=_PROBE_TIMEOUT, env=env, ) except subprocess.TimeoutExpired: return None, f"capability probe timed out after {_PROBE_TIMEOUT}s" payload: Optional[dict] = None for line in (result.stdout or "").splitlines(): if line.startswith(_PROBE_SENTINEL): try: payload = json.loads(line[len(_PROBE_SENTINEL):]) except json.JSONDecodeError: payload = None if payload is None: err = (result.stderr or "").strip() return None, ( "capability probe produced no result " f"(exit {result.returncode})" + (f": {err}" if err else "") ) if "error" in payload: return None, str(payload["error"]) return payload, "" def _declared_list(manifest: dict, key: str) -> List[str]: raw = manifest.get(key) or [] if not isinstance(raw, list): return [] return [str(item) for item in raw if isinstance(item, str)] def _check_capabilities( report: ValidationReport, manifest: dict, plugin_dir: Path ) -> Optional[dict]: """Probe actual registrations and diff against declared capabilities. Returns the recorded dict (for the built-in collision check) or None when the probe failed / was skipped. """ if not (plugin_dir / "__init__.py").is_file(): report.warn( "no __init__.py — capability probe skipped (manifest-only plugin)" ) report.add("capability probe", True, "skipped (no __init__.py)") return None recorded, error = _run_capability_probe(plugin_dir, manifest) if recorded is None: report.add("capability probe", False, error) return None if recorded.get("providers"): report.add( "capability probe", True, "import registered provider(s): " + ", ".join(recorded["providers"]), ) else: report.add("capability probe", True, "register() ran in isolation") for kind, manifest_key in ( ("tools", "provides_tools"), ("hooks", "provides_hooks"), ("middleware", "provides_middleware"), ): declared = set(_declared_list(manifest, manifest_key)) actual = set(recorded.get(kind) or []) undeclared = sorted(actual - declared) unregistered = sorted(declared - actual) if undeclared: report.add( f"declared {kind}", False, f"undeclared {kind} registered (not in {manifest_key}): " f"{', '.join(undeclared)}", ) else: report.add(f"declared {kind}", True, "matches registrations") if unregistered: report.warn( f"{manifest_key} declares {', '.join(unregistered)} " f"but register() did not register them" ) return recorded def _builtin_tool_names() -> List[str]: """Return the built-in tool registry names (discovery-timing safe). ``tools.registry`` starts empty — built-in tool modules self-register on import, so we must run ``discover_builtin_tools()`` first (idempotent; see the AGENTS.md discover_plugins timing pitfall). """ try: from tools.registry import discover_builtin_tools, registry discover_builtin_tools() return list(registry.get_all_tool_names()) except Exception: return [] def _check_builtin_collisions( report: ValidationReport, manifest: dict, recorded: Optional[dict] ) -> None: candidate_tools = set(_declared_list(manifest, "provides_tools")) if recorded: candidate_tools.update(recorded.get("tools") or []) if not candidate_tools: report.add("built-in tool collisions", True, "no tools to check") return builtin = set(_builtin_tool_names()) collisions = sorted(candidate_tools & builtin) if collisions: report.add( "built-in tool collisions", False, "tool name(s) collide with built-in tools: " f"{', '.join(collisions)}", ) else: report.add("built-in tool collisions", True, "no collisions") # ─── Entry point ───────────────────────────────────────────────────────────── def validate_plugin_dir(plugin_dir: Path) -> ValidationReport: """Run every admission check against *plugin_dir* and return the report.""" report = ValidationReport() plugin_dir = Path(plugin_dir) if not plugin_dir.is_dir(): report.add( "plugin directory", False, f"{plugin_dir} is not a directory" ) return report manifest_file = plugin_dir / "plugin.yaml" if not manifest_file.is_file(): manifest_file = plugin_dir / "plugin.yml" if not manifest_file.is_file(): # Portable Agent Plugins v1 (#81196): a plugin.json-only package is # admissible — validated through the portable manifest reader. When a # package carries both manifests the native plugin.yaml always wins # (this branch is only reached when no native manifest exists). portable_file = plugin_dir / "plugin.json" if portable_file.is_file(): return _validate_portable_plugin(report, plugin_dir) report.add( "manifest", False, "no plugin.yaml (or portable plugin.json) in the plugin directory", ) return report import hermes_yaml as yaml try: manifest = yaml.safe_load( manifest_file.read_text(encoding="utf-8-sig") ) except Exception as exc: report.add("manifest", False, f"plugin.yaml failed to parse: {exc}") return report if not isinstance(manifest, dict): report.add("manifest", False, "plugin.yaml must be a mapping") return report report.add("manifest", True, "plugin.yaml parses") _check_manifest_fields(report, manifest) _check_requires_hermes(report, manifest) _check_config_spec(report, manifest) _check_requires_env(report, manifest) _check_loadable(report, plugin_dir, manifest) _check_python_dependencies(report, plugin_dir) recorded = _check_capabilities(report, manifest, plugin_dir) _check_builtin_collisions(report, manifest, recorded) _check_security_scan(report, plugin_dir) check_core_override(report, plugin_dir) check_desktop_surface(report, plugin_dir) check_language_packs(report, manifest, plugin_dir) return report _LOADABLE_ENTRYPOINTS = ("__init__.py", "desktop/plugin.js", "plugin.json") def _check_loadable(report: ValidationReport, plugin_dir: Path, manifest: Optional[dict] = None) -> None: """A plugin.yaml with nothing beside it that Hermes can load (no ``register()`` module, no desktop bundle, no portable manifest, no declared language pack) installs "successfully" and does nothing — a pip-layout repo whose code lives under ``src/`` behind an entry point is the usual shape.""" present = [rel for rel in _LOADABLE_ENTRYPOINTS if (plugin_dir / rel).is_file()] if (manifest or {}).get("provides_locales") and (plugin_dir / "locales").is_dir(): present.append("locales/ (language pack)") report.add( "loadable", bool(present), f"entry: {', '.join(present)}" if present else "nothing to load: no __init__.py, desktop/plugin.js, plugin.json or provides_locales + locales/ beside " "plugin.yaml (pip-layout packages need a directory-plugin wrapper with a pyproject.toml declaring the deps)", ) def _check_python_dependencies(report: ValidationReport, plugin_dir: Path) -> None: """Declared deps (pyproject ``[project].dependencies`` or manifest ``python_dependencies``) must be well-formed PEP 508 specs the installer will accept; a plugin opting out with ``python_runtime: external`` declares none.""" from pm.plugin_declarations import read_python_declaration, unsupported_requirements try: decl = read_python_declaration(plugin_dir) if decl.external: report.add("python dependencies", True, "external runtime (plugin manages its own)") return urls = unsupported_requirements(decl.requirements) installable = decl.install_requirements except (ValueError, OSError) as exc: report.add("python dependencies", False, f"declaration invalid: {exc}") return if urls: report.warn("python dependencies: direct URL requirement(s) are not managed by PM; " f"use a plugin-owned external runtime: {', '.join(urls)}") source = "pyproject" if decl.pyproject is not None else "manifest" detail = f"{len(installable)} requirements from {source}" if decl.requirements else "none declared" report.add("python dependencies", True, detail) def _check_security_scan(report: ValidationReport, plugin_dir: Path) -> None: """Run the install-time scanner at admission, so a pin a reviewer approves is one the installer will accept: ``dangerous`` fails the entry; ``caution`` findings surface as warnings for the reviewer (the installer trusts them once the pin is merged).""" from tools.plugin_guard import scan_plugin result = scan_plugin(plugin_dir) flagged = [f for f in result.findings if f.severity in ("critical", "high")] summary = ", ".join(sorted({f"{f.pattern_id} ({Path(f.file).name}:{f.line})" for f in flagged})) or "no findings" if result.verdict == "dangerous": report.add("security scan", False, f"dangerous: {summary}") return report.add("security scan", True, result.verdict) if result.verdict == "caution": report.warn(f"security scan caution: {summary}") def _validate_portable_plugin(report: ValidationReport, plugin_dir: Path) -> ValidationReport: """Admission checks for a portable Agent Plugins v1 (plugin.json) package. Portable packages have no register() entry point, so the capability probe does not apply; validation is the manifest reader's own diagnostics (schema shape, name, supported subset). """ try: from hermes_cli.agent_plugins import load_agent_plugin from hermes_platform.resolver.availability import availability with tempfile.TemporaryDirectory() as data_root: package = load_agent_plugin(plugin_dir, Path(data_root)) manifest = package.manifest diagnostics = package.diagnostics except Exception as exc: report.add("portable manifest", False, f"plugin.json failed validation: {exc}") return report for diag in diagnostics: scope = getattr(diag, "scope", "") message = getattr(diag, "message", str(diag)) report.warnings.append(f"{scope}: {message}" if scope else message) report.add("portable manifest", True, "plugin.json parses (Agent Plugins v1)") name = str(manifest.get("name") or "").strip() report.add( "manifest fields", bool(name), "name present" if name else "plugin.json missing required 'name'", ) for server_name, server_decl in package.server_declarations.items(): result = availability(server_decl.declaration) detail = result.state if result.version: detail += f", version {result.version}" if result.path: detail += f", path {result.path}" report.add(f"server availability: {server_name}", True, detail) _check_security_scan(report, plugin_dir) check_desktop_surface(report, plugin_dir) return report