Files
hermes-agent/hermes_cli/plugin_validate.py
Siddharth Balyan adefd99671 Portable plugins declare the application each MCP server needs; install refuses on an unsupported host (NS-942) (#119049)
* feat: enforce portable server declarations

* test: cover portable plugin host gates
2026-09-22 17:54:41 +05:30

629 lines
23 KiB
Python

"""``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_desktop import check_desktop_surface
_UPPER_SNAKE_RE = re.compile(r"^[A-Z][A-Z0-9_]*$")
_CONFIG_TYPES = {
"str", "string", "int", "integer", "float", "number",
"bool", "boolean", "list", "array", "dict", "mapping", "map",
}
_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,
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 yaml
try:
manifest = yaml.safe_load(
manifest_file.read_text(encoding="utf-8")
)
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)
_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_desktop_surface(report, plugin_dir)
return report
_LOADABLE_ENTRYPOINTS = ("__init__.py", "desktop/plugin.js", "plugin.json")
def _check_loadable(report: ValidationReport, plugin_dir: Path) -> None:
"""A plugin.yaml with nothing beside it that Hermes can load (no ``register()`` module, no
desktop bundle, no portable manifest) 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()]
report.add(
"loadable", bool(present),
f"entry: {', '.join(present)}" if present else
"nothing to load: no __init__.py, desktop/plugin.js or plugin.json 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 hermes_cli.plugin_python_deps import read_declaration
try:
decl = read_declaration(plugin_dir)
except Exception as exc:
report.add("python dependencies", False, f"declaration invalid: {exc}")
return
if decl.external:
report.add("python dependencies", True, "external runtime (plugin manages its own)")
return
from hermes_cli.plugin_python_deps import applicable_specs, unsupported_specs
urls = unsupported_specs(decl.specs)
if urls:
report.warn("python dependencies: direct URL requirement(s) are never auto-installed, users must "
f"install them by hand: {', '.join(urls)}")
installable = applicable_specs(decl.specs)
rejected = [s for s in installable if not _spec_is_safe(s)]
detail = f"{len(installable)} installable from {decl.source}" if decl.source else "none declared"
report.add("python dependencies", not rejected,
f"unsafe spec(s): {', '.join(rejected)}" if rejected else detail)
def _spec_is_safe(spec: str) -> bool:
from tools.lazy_deps import _spec_is_safe as safe
return safe(spec)
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