mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
fix(plugins): clear catalog sweep 1028 scanner false positives and close three validator gaps
Install scanner (tools/plugin_guard*, PLUGIN_SCANNER_VERSION bumped once):
- hardcoded_secret: a PostHog project key (phc_, write-only ingest key) joins the
Google installed-app secret as a public client key -> high (caution), only
when every matched literal on the line has a public prefix (telesec
telemetry.py:27 was dangerous). phx_ personal keys and any other token stay
critical.
- proc_access: reads of /proc/self|<pid>/{cgroup,mountinfo,stat,status,...}
(container placement, pid start time) land at low instead of high
(hermes-lcm-x cpu_capacity.py was caution). environ, cmdline, mem, maps, fd,
root/cwd, ns, bare /proc/self, any unlisted leaf and any write keep high.
Generalises #144304's cgroup/mountinfo allowance and keeps the finding
visible instead of dropping it.
- bytecode_file: a committed .pyc/.pyo (git ls-files; plus any bytecode outside
__pycache__) is a high structural finding, the binary_file tier (rule 17;
hermes-emotion shipped tests/__pycache__/*.pyc and scanned safe). Untracked
caches the validate run writes are not flagged.
Validator (hermes plugins validate, warnings only):
- private imports: gateway.session_context.get_session_env is the documented
route (plugins/index.md) and a plugin's fallback import of its own install
path (plugins.memory.openltm._db) is not a core import.
- new warning for private-method calls on core handles (SessionDB, AIAgent,
GatewayRunner, PluginContext) the plugin does not define itself
(opencode-sync db._execute_write, runner._evict_cached_agent).
- desktop surface review warning for raw window.hermesDesktop calls and app
markup selectors ([data-slot], [data-tree-*], [data-pane-host]) in CSS text
or closest() (squad-hub, novel-reader, search-sources).
This commit is contained in:
1 parent
1bb5b3d67b
commit
0040f02b5c
9 files changed
+348
-107
No files matched your search
@@ -45,6 +45,19 @@ _FORBIDDEN: tuple[tuple[str, re.Pattern[str]], ...] = (
|
||||
re.compile(r"\b(?:querySelector(?:All)?|closest|getElementsByTagName|createElement)\(\s*[`'\"]webview\b")),
|
||||
)
|
||||
|
||||
# Review cues, warned and never refused (catalog rule 8): the raw ``window.hermesDesktop`` native bridge
|
||||
# (files, git, REST, installs) has no folder or profile scoping — ``ctx.files`` / ``ctx.os`` /
|
||||
# ``host.request`` do — yet listed plugins still use it where the SDK has no verb (microphone access,
|
||||
# favicon lookup). And the app's markup selected from a CSS rule string, ``closest()`` or a template
|
||||
# (``[data-tree-tab=…]{display:none}``) is the reach the refused ``document.querySelector`` rule
|
||||
# names, in shapes no refusal regex can tell from a plugin styling SDK components it rendered itself.
|
||||
_REVIEW: tuple[tuple[str, re.Pattern[str]], ...] = (
|
||||
("raw bridge window.hermesDesktop",
|
||||
re.compile(r"\b(?:window|globalThis|self)\s*(?:\?\.|\.)\s*hermesDesktop\b")),
|
||||
("app markup selector",
|
||||
re.compile(r"\[data-(?:slot|tour|sidebar|testid|tree-[\w-]+|pane-host|titlebar-cluster)\b")),
|
||||
)
|
||||
|
||||
# String literals are matched first and kept, so a ``/*`` or ``//`` INSIDE a string (a glob such as
|
||||
# ``'**/*.md'``, a URL) cannot open a "comment" that blanks every line up to the next ``*/`` and
|
||||
# hides whatever forbidden construct sits there.
|
||||
@@ -93,6 +106,14 @@ def desktop_surface_findings(source: str) -> list[tuple[str, int]]:
|
||||
return sorted(findings, key=lambda f: f[1])
|
||||
|
||||
|
||||
def desktop_review_findings(source: str) -> list[tuple[str, int]]:
|
||||
"""``[(cue, line)]`` for every review cue in a plugin.js source, skipping lines a refusal covers."""
|
||||
stripped = _strip_comments(source)
|
||||
refused = {line for _rule, line in desktop_surface_findings(source)}
|
||||
found = {(cue, stripped.count("\n", 0, m.start()) + 1) for cue, pattern in _REVIEW for m in pattern.finditer(stripped)}
|
||||
return sorted((hit for hit in found if hit[1] not in refused), key=lambda h: (h[1], h[0]))
|
||||
|
||||
|
||||
# The Desktop installer's entry rules (apps/desktop/electron/desktop-plugin-install.ts
|
||||
# ``findDesktopEntry``): a repo-root ``plugin.js`` wins and the whole repo root is published as the
|
||||
# plugin folder; otherwise ``desktop/plugin.js`` and its ``desktop/`` tree. Admission must lint every
|
||||
@@ -136,8 +157,9 @@ def _desktop_surface_files(plugin_dir: Path) -> list[Path]:
|
||||
)
|
||||
|
||||
|
||||
def desktop_surface_hits(plugin_dir: Path) -> list[str]:
|
||||
"""``["<rule> (<rel>:<line>)", ...]`` over the plugin's Desktop surface files only."""
|
||||
def desktop_surface_hits(plugin_dir: Path, findings=None) -> list[str]:
|
||||
"""``["<rule> (<rel>:<line>)", ...]`` over the plugin's Desktop surface files only. *findings* is
|
||||
the per-source rule pass (``desktop_surface_findings`` refusals by default)."""
|
||||
plugin_dir = Path(plugin_dir)
|
||||
hits: list[str] = []
|
||||
for js in _desktop_surface_files(plugin_dir):
|
||||
@@ -146,7 +168,7 @@ def desktop_surface_hits(plugin_dir: Path) -> list[str]:
|
||||
source = js.read_text(encoding="utf-8-sig", errors="replace")
|
||||
except OSError:
|
||||
continue
|
||||
hits.extend(f"{rule} ({rel}:{line})" for rule, line in desktop_surface_findings(source))
|
||||
hits.extend(f"{rule} ({rel}:{line})" for rule, line in (findings or desktop_surface_findings)(source))
|
||||
return hits
|
||||
|
||||
|
||||
@@ -161,3 +183,7 @@ def check_desktop_surface(report, plugin_dir: Path) -> None:
|
||||
"; ".join(hits[:8]) + (f" (+{len(hits) - 8} more)" if len(hits) > 8 else "")
|
||||
if hits else "stays inside the plugin SDK surface",
|
||||
)
|
||||
if cues := desktop_surface_hits(plugin_dir, desktop_review_findings):
|
||||
report.warn("desktop surface review (catalog rule 8: SDK surface only): " + "; ".join(cues[:8])
|
||||
+ (f" (+{len(cues) - 8} more)" if len(cues) > 8 else "")
|
||||
+ ". Use ctx.files / ctx.os / host.request and SDK slots, or open an issue for the hook you need.")
|
||||
@@ -62,6 +62,8 @@ _PUBLIC = frozenset({
|
||||
"tools.skills_tool:find_all_skills", "tools.skills_tool:sort_skills", "tools.skills_tool:skill_search_dirs",
|
||||
"tools.skills_tool:locate_skill", "tools.skills_tool:skill_lookup_path_error",
|
||||
"tools.skills_sync:dir_hash", "tools.skill_usage:read_skill_name",
|
||||
# plugins/index.md § platform plugins: read session identity through the task-local context
|
||||
"gateway.session_context:get_session_env",
|
||||
})
|
||||
|
||||
_SHOWN = 10
|
||||
@@ -111,30 +113,52 @@ def _imported(node: ast.AST) -> list[tuple[str, str]]:
|
||||
return []
|
||||
|
||||
|
||||
@lru_cache(maxsize=512)
|
||||
def _core_module_exists(module: str) -> bool:
|
||||
path = Path(__file__).resolve().parents[1].joinpath(*module.split("."))
|
||||
return path.with_suffix(".py").is_file() or path.is_dir()
|
||||
|
||||
|
||||
def _is_own_module_path(module: str, local: set[str]) -> bool:
|
||||
"""A Hermes-shaped path that core does not ship but whose tail is one of the plugin's own modules:
|
||||
the fallback spelling of the plugin's install location (``plugins.memory.openltm._db`` beside
|
||||
``from . import _db``). A path core DOES ship stays a core import."""
|
||||
parts = module.split(".")
|
||||
return any(".".join(parts[i:]) in local for i in range(1, len(parts))) and not _core_module_exists(module)
|
||||
|
||||
|
||||
def _private_target(module: str, name: str, core: frozenset, local: set[str]) -> str:
|
||||
"""The dotted target to report, or "" when the import is public, local or not Hermes's."""
|
||||
target = f"{module}.{name}" if name else module
|
||||
hermes = module.split(".")[0] in core and module not in local
|
||||
hermes = module.split(".")[0] in core and module not in local and not _is_own_module_path(module, local)
|
||||
public = target in _PUBLIC or f"{module}:{name}" in _PUBLIC \
|
||||
or (module in _PUBLIC and not name.startswith("_")) # ``from gateway.platforms import base`` too
|
||||
return target if hermes and not public else ""
|
||||
|
||||
|
||||
def _parsed_files(plugin_dir: Path) -> list[tuple[Path, ast.AST]]:
|
||||
"""``(rel, tree)`` for every plugin module that can load inside Hermes. Hidden dirs (``.github/``
|
||||
CI helpers) never do; tests and vendored envs are skipped by ``_python_files``."""
|
||||
parsed = []
|
||||
for path, rel in _python_files(plugin_dir):
|
||||
if any(p.startswith(".") for p in rel.parts[:-1]):
|
||||
continue
|
||||
try:
|
||||
parsed.append((rel, ast.parse(path.read_text(encoding="utf-8-sig", errors="replace"))))
|
||||
except (SyntaxError, ValueError):
|
||||
continue
|
||||
return parsed
|
||||
|
||||
|
||||
def private_import_findings(plugin_dir: Path) -> list[str]:
|
||||
"""``["<rel>:<line> <module>[.<name>]", ...]`` for every import of a private Hermes core module:
|
||||
the first site of each distinct module first (so a truncated list still shows the breadth),
|
||||
then the rest, each group in file/line order."""
|
||||
plugin_dir = Path(plugin_dir)
|
||||
# Hidden dirs (``.github/`` CI helpers) never load inside Hermes.
|
||||
files = [(path, rel) for path, rel in _python_files(plugin_dir) if not any(p.startswith(".") for p in rel.parts[:-1])]
|
||||
local = _local_modules(rel for _path, rel in files)
|
||||
files = _parsed_files(Path(plugin_dir))
|
||||
local = _local_modules(rel for rel, _tree in files)
|
||||
core = _hermes_packages()
|
||||
found: set[tuple[str, int, str, str]] = set()
|
||||
for path, rel in files:
|
||||
try:
|
||||
tree = ast.parse(path.read_text(encoding="utf-8-sig", errors="replace"))
|
||||
except (SyntaxError, ValueError):
|
||||
continue
|
||||
for rel, tree in files:
|
||||
skipped = _type_checking_nodes(tree)
|
||||
for node in ast.walk(tree):
|
||||
if id(node) in skipped:
|
||||
@@ -149,13 +173,69 @@ def private_import_findings(plugin_dir: Path) -> list[str]:
|
||||
return [f"{rel}:{line} {target}" for rel, line, target, _module in first + rest]
|
||||
|
||||
|
||||
# The core objects a plugin holds a live handle to — ``SessionDB`` (``ctx`` session APIs, the
|
||||
# dashboard), ``AIAgent`` and ``GatewayRunner`` (hook kwargs), ``PluginContext`` — split across their
|
||||
# facade + sibling files. Calling one of their private methods (``db._execute_write(...)``,
|
||||
# ``runner._evict_cached_agent(...)``) couples the plugin to core internals exactly like a private
|
||||
# import, but no import names it, so the import pass above cannot see it.
|
||||
_CORE_HANDLE_SOURCES = ("hermes_state.py", "hermes_state_*.py", "run_agent.py", "agent/turn_*.py",
|
||||
"gateway/run.py", "gateway/run_*.py", "hermes_cli/plugins.py")
|
||||
|
||||
|
||||
@lru_cache(maxsize=1)
|
||||
def _core_private_methods() -> frozenset:
|
||||
root = Path(__file__).resolve().parents[1]
|
||||
names: set[str] = set()
|
||||
for pattern in _CORE_HANDLE_SOURCES:
|
||||
for path in root.glob(pattern):
|
||||
try:
|
||||
tree = ast.parse(path.read_text(encoding="utf-8-sig"))
|
||||
except (OSError, SyntaxError, ValueError):
|
||||
continue
|
||||
names |= {item.name for node in ast.walk(tree) if isinstance(node, ast.ClassDef) for item in node.body
|
||||
if isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef))
|
||||
and item.name.startswith("_") and not item.name.startswith("__")}
|
||||
return frozenset(names)
|
||||
|
||||
|
||||
def _own_names(files: list[tuple[Path, ast.AST]]) -> set[str]:
|
||||
"""Every name the plugin defines or assigns, so its own ``obj._helper()`` never reads as core."""
|
||||
own: set[str] = set()
|
||||
for _rel, tree in files:
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
|
||||
own.add(node.name)
|
||||
elif isinstance(node, ast.Attribute) and isinstance(node.ctx, ast.Store):
|
||||
own.add(node.attr)
|
||||
elif isinstance(node, ast.Name) and isinstance(node.ctx, ast.Store):
|
||||
own.add(node.id)
|
||||
return own
|
||||
|
||||
|
||||
def private_attribute_findings(plugin_dir: Path) -> list[str]:
|
||||
"""``["<rel>:<line> <expr>", ...]`` for every access to a private method of a core handle class
|
||||
that the plugin does not define itself (``self``/``cls`` receivers excluded)."""
|
||||
files = _parsed_files(Path(plugin_dir))
|
||||
private = _core_private_methods() - _own_names(files)
|
||||
found = {(rel.as_posix(), node.lineno, ast.unparse(node)[:80]) for rel, tree in files for node in ast.walk(tree)
|
||||
if isinstance(node, ast.Attribute) and node.attr in private
|
||||
and not (isinstance(node.value, ast.Name) and node.value.id in ("self", "cls"))}
|
||||
return [f"{rel}:{line} {expr}" for rel, line, expr in sorted(found)]
|
||||
|
||||
|
||||
def _warn_sites(report, label: str, hits: list[str]) -> None:
|
||||
more = f"; and {len(hits) - _SHOWN} more" if len(hits) > _SHOWN else ""
|
||||
report.warn(
|
||||
f"{label} (catalog rule 9: public surfaces only; these can break on any release): "
|
||||
+ "; ".join(hits[:_SHOWN]) + more
|
||||
+ ". Use ctx.* / documented plugin APIs, or open an issue for the seam you need."
|
||||
)
|
||||
|
||||
|
||||
def check_private_imports(report, plugin_dir: Path) -> None:
|
||||
"""Warn (never fail) when the plugin imports private Hermes core modules."""
|
||||
hits = private_import_findings(plugin_dir)
|
||||
if hits:
|
||||
more = f"; and {len(hits) - _SHOWN} more" if len(hits) > _SHOWN else ""
|
||||
report.warn(
|
||||
"private core imports (catalog rule 9: public surfaces only; these can break on any release): "
|
||||
+ "; ".join(hits[:_SHOWN]) + more
|
||||
+ ". Use ctx.* / documented plugin APIs, or open an issue for the seam you need."
|
||||
)
|
||||
"""Warn (never fail) when the plugin imports private Hermes core modules or calls private
|
||||
methods on the core objects it is handed."""
|
||||
if hits := private_import_findings(plugin_dir):
|
||||
_warn_sites(report, "private core imports", hits)
|
||||
if hits := private_attribute_findings(plugin_dir):
|
||||
_warn_sites(report, "private core attribute access", hits)
|
||||
@@ -58,8 +58,11 @@ meaningful:
|
||||
(querying `data-slot` / `data-tour` / `data-sidebar` / `data-testid`
|
||||
elements from `document`, or a `document.body` MutationObserver, to restyle,
|
||||
hide, click or rewrite core UI). `hermes plugins validate` refuses these at
|
||||
admission (`desktop surface` check); a plugin that needs a capability the
|
||||
SDK lacks asks for an SDK hook instead of patching around it.
|
||||
admission (`desktop surface` check) and warns, for the reviewer, on raw
|
||||
`window.hermesDesktop` bridge calls and on the app's markup selected any
|
||||
other way (`[data-slot]`, `[data-tree-*]` in CSS text or `closest()`); a
|
||||
plugin that needs a capability the SDK lacks asks for an SDK hook instead
|
||||
of patching around it.
|
||||
9. **No runtime overrides of Hermes core.** A listed plugin extends Hermes only
|
||||
through public surfaces: hooks, middleware, provider profiles and the
|
||||
other `register_*` APIs, and Desktop SDK slots and routes. It must not
|
||||
@@ -69,7 +72,8 @@ meaningful:
|
||||
patching the same seam silently break each other, and every core release
|
||||
can break both. `hermes plugins validate` refuses these at admission (`no
|
||||
core override` check), and warns (without failing) on each import of an
|
||||
undocumented core module, by file and line. If the hook you need does not
|
||||
undocumented core module and each call to a private method of a core
|
||||
object it is handed (`db._execute_write(...)`), by file and line. If the hook you need does not
|
||||
exist, open an issue describing it: we would rather add the seam than list
|
||||
a patch.
|
||||
10. **Dependency security policy is the plugin's.** Hermes's 3-day
|
||||
|
||||
@@ -289,6 +289,25 @@ class TestDesktopSurface:
|
||||
))
|
||||
report = validate_plugin_dir(d)
|
||||
assert any(name == "desktop surface" and ok for name, ok, _ in report.checks)
|
||||
assert not [w for w in report.warnings if w.startswith("desktop surface review")]
|
||||
|
||||
def test_raw_bridge_and_app_markup_selectors_warn_for_review(self, tmp_path):
|
||||
"""squad-hub / novel-reader shapes: the raw native bridge and the app's layout markup selected
|
||||
from CSS text or ``closest()`` pass the refusal rules, so review gets a warning (listed plugins
|
||||
still need the bridge where the SDK has no verb). A comment naming them is not a use."""
|
||||
d = self._desktop_plugin(tmp_path, (
|
||||
"// window.hermesDesktop.readDir and [data-slot=x] are what we avoid\n"
|
||||
"const git = () => window.hermesDesktop?.git || null\n"
|
||||
"rules.push('[data-tree-tab=\"bots:pane\"]{display:none!important}')\n"
|
||||
"const root = el.closest('[data-tree-split]')\n"
|
||||
))
|
||||
report = validate_plugin_dir(d)
|
||||
assert any(name == "desktop surface" and ok for name, ok, _ in report.checks)
|
||||
[warning] = [w for w in report.warnings if w.startswith("desktop surface review")]
|
||||
assert "raw bridge window.hermesDesktop (desktop/plugin.js:2)" in warning
|
||||
assert "app markup selector (desktop/plugin.js:3)" in warning
|
||||
assert "app markup selector (desktop/plugin.js:4)" in warning
|
||||
assert ":1)" not in warning
|
||||
|
||||
def test_script_regex_literal_is_not_injection_but_string_is(self, tmp_path):
|
||||
d = self._desktop_plugin(tmp_path, (
|
||||
|
||||
@@ -126,3 +126,49 @@ def test_skill_catalog_seams_are_public_and_drive_the_skills_tools(tmp_path, mon
|
||||
monkeypatch.setattr(skills_tool, "find_all_skills", lambda **_kw: [dict(marker)])
|
||||
listed = json.loads(skills_tool.skills_list())
|
||||
assert [s["name"] for s in listed["skills"] if s["name"] == "seam-probe-skill"] == ["seam-probe-skill"]
|
||||
|
||||
|
||||
def test_documented_session_helper_and_own_install_path_fallback_are_not_flagged(tmp_path):
|
||||
"""onebot-platform reads session identity through the documented ``get_session_env``; openltm
|
||||
falls back to its own install path (``plugins.memory.openltm._db``) when the relative import
|
||||
fails. Neither is a core internal. A real core path beside them still warns."""
|
||||
plugin = _plugin(tmp_path / "openltm", {
|
||||
"__init__.py": (
|
||||
"def register(ctx):\n"
|
||||
" from gateway.session_context import get_session_env\n"
|
||||
" try:\n"
|
||||
" from . import _db\n"
|
||||
" except ImportError:\n"
|
||||
" import plugins.memory.openltm._db as _db\n"
|
||||
" from hermes_cli.gateway_client import _session_ticket\n"
|
||||
" return get_session_env, _db, _session_ticket\n"
|
||||
),
|
||||
"_db.py": "",
|
||||
})
|
||||
|
||||
[warning] = _private_warnings(validate_plugin_dir(plugin))
|
||||
|
||||
sites = warning.split("release): ", 1)[1].split(". Use ctx.*")[0].split("; ")
|
||||
assert sites == ["__init__.py:7 hermes_cli.gateway_client._session_ticket"], sites
|
||||
|
||||
|
||||
def test_private_method_calls_on_core_handles_warn_but_own_helpers_do_not(tmp_path):
|
||||
"""opencode-sync calls SessionDB / GatewayRunner private methods on the objects it is handed;
|
||||
no import names them, so only the attribute pass can see it. The plugin's own ``_helper`` and
|
||||
``self._x`` calls are not core reach."""
|
||||
plugin = _plugin(tmp_path / "sync", {"engine.py": (
|
||||
"class Store:\n"
|
||||
" def _execute_write(self, fn):\n"
|
||||
" return self._execute_write(fn)\n"
|
||||
"def apply(db, runner, store, route):\n"
|
||||
" rows = db._row_to_message_dict(None)\n"
|
||||
" runner._evict_cached_agent(route)\n"
|
||||
" return store._execute_write(rows)\n"
|
||||
)})
|
||||
|
||||
report = validate_plugin_dir(plugin)
|
||||
|
||||
assert report.ok, report.failures
|
||||
[warning] = [w for w in report.warnings if w.startswith("private core attribute access")]
|
||||
sites = warning.split("release): ", 1)[1].split(". Use ctx.*")[0].split("; ")
|
||||
assert sites == ["engine.py:5 db._row_to_message_dict", "engine.py:6 runner._evict_cached_agent"], sites
|
||||
@@ -8,6 +8,7 @@ content (credential-store exfiltration, reverse shells, prompt injection
|
||||
in docs) is flagged or blocked.
|
||||
"""
|
||||
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
@@ -959,53 +960,71 @@ class TestCatalogSweepFalsePositives:
|
||||
assert hit is not None and hit.severity in ("high", "critical"), format_scan_report(result)
|
||||
assert result.verdict in ("caution", "dangerous")
|
||||
|
||||
|
||||
class TestBenignProcIntrospection:
|
||||
"""#144123: /proc/self/cgroup and /proc/self/mountinfo are read by container CPU-quota
|
||||
detection (and by language runtimes), so a read-only mention is not process
|
||||
introspection. Writes, other proc paths and a bare /proc/self keep the finding."""
|
||||
|
||||
def test_container_layout_read_is_safe(self, tmp_path):
|
||||
files = dict(BASE_FILES)
|
||||
files["capacity.py"] = (
|
||||
"from pathlib import Path\n\n"
|
||||
"def container_layout():\n"
|
||||
" return (\n"
|
||||
' Path("/proc/self/cgroup").read_text(),\n'
|
||||
' Path("/proc/self/mountinfo").read_text(),\n'
|
||||
" )\n"
|
||||
)
|
||||
result = scan_plugin(_mk_plugin(tmp_path, files), source="owner/repo")
|
||||
assert [f for f in result.findings if f.pattern_id == "proc_access"] == []
|
||||
assert result.verdict == "safe", format_scan_report(result)
|
||||
|
||||
def test_doc_mention_is_safe(self, tmp_path):
|
||||
files = {**BASE_FILES, "README.md": "# x\n\nReads /proc/self/cgroup for the CPU quota.\n"}
|
||||
result = scan_plugin(_mk_plugin(tmp_path, files), source="owner/repo")
|
||||
assert result.verdict == "safe", format_scan_report(result)
|
||||
|
||||
@pytest.mark.parametrize("line", [
|
||||
'Path("/proc/self/status").read_text()\n',
|
||||
'Path("/proc/1/cmdline").read_text()\n',
|
||||
'print(Path("/proc/self").read_text())\n',
|
||||
'(Path("/proc/self/cgroup"), Path("/proc/self/status"))\n',
|
||||
])
|
||||
def test_other_proc_paths_still_flag(self, tmp_path, line):
|
||||
files = {**BASE_FILES, "probe.py": line}
|
||||
result = scan_plugin(_mk_plugin(tmp_path, files), source="owner/repo")
|
||||
hit = next((f for f in result.findings if f.pattern_id == "proc_access"), None)
|
||||
assert hit is not None and hit.severity == "high", format_scan_report(result)
|
||||
|
||||
@pytest.mark.parametrize("line", [
|
||||
'Path("/proc/self/cgroup").write_text("x")\n',
|
||||
'handle = open("/proc/self/mountinfo", "w")\n',
|
||||
])
|
||||
def test_writes_still_flag(self, tmp_path, line):
|
||||
files = {**BASE_FILES, "writer.py": line}
|
||||
result = scan_plugin(_mk_plugin(tmp_path, files), source="owner/repo")
|
||||
hit = next((f for f in result.findings if f.pattern_id == "proc_access"), None)
|
||||
assert hit is not None and hit.severity == "high", format_scan_report(result)
|
||||
def test_scanner_version_contract(self):
|
||||
# A bump un-blocks previously-quarantined packages; assert the contract (changed), not a literal.
|
||||
assert PLUGIN_SCANNER_VERSION != "plugin-guard-v10"
|
||||
assert PLUGIN_SCANNER_VERSION != "plugin-guard-v11"
|
||||
assert PLUGIN_SCANNER_VERSION.startswith("plugin-guard-v")
|
||||
|
||||
|
||||
# Built at runtime so no full vendor-shaped key sits in the source (forge secret scanning).
|
||||
_FAKE_KEY_BODY = "Ab3" * 14 + "x"
|
||||
|
||||
|
||||
class TestProcKeyAndBytecodeShapes:
|
||||
"""Shapes real catalog pins tripped (telesec, hermes-lcm-x, hermes-emotion), each next to the
|
||||
attack twin that must keep its severity."""
|
||||
|
||||
def test_benign_proc_reads_and_public_client_key_install_without_a_block(self, tmp_path):
|
||||
# #144123 (hermes-lcm-x cpu_capacity.py): container placement reads, plus the pid-reuse
|
||||
# start-time read. Kept visible at low, never deleted.
|
||||
files = {**BASE_FILES, "capacity.py": (
|
||||
"from pathlib import Path\n"
|
||||
'groups = Path("/proc/self/cgroup").read_text()\n'
|
||||
'mounts = Path("/proc/self/mountinfo").read_text()\n'
|
||||
'container = open("/proc/1/cgroup", "rb").read()\n'
|
||||
'start = Path("/proc/self/stat").read_text().split()[21]\n'
|
||||
)}
|
||||
result = scan_plugin(_mk_plugin(tmp_path, files), source="owner/repo")
|
||||
proc = [f for f in result.findings if f.pattern_id == "proc_access"]
|
||||
assert len(proc) == 4 and {f.severity for f in proc} == {"low"}, format_scan_report(result)
|
||||
assert result.verdict == "safe", format_scan_report(result)
|
||||
|
||||
# telesec telemetry.py:27: a PostHog project key is a write-only ingest key, public by design.
|
||||
files = {**BASE_FILES, "telemetry.py": f'API_KEY = "phc_{_FAKE_KEY_BODY}" # public\n'}
|
||||
(tmp_path / "posthog").mkdir()
|
||||
result = scan_plugin(_mk_plugin(tmp_path / "posthog", files), source="owner/repo")
|
||||
assert result.verdict == "caution", format_scan_report(result)
|
||||
assert [f.severity for f in result.findings if f.pattern_id == "hardcoded_secret"] == ["high"]
|
||||
|
||||
@pytest.mark.parametrize("rel,text,pattern_id,severity", [
|
||||
("probe.py", 'Path("/proc/1/environ").read_bytes()\n', "proc_access", "high"),
|
||||
("probe.py", 'Path("/proc/self/environ").read_bytes()\n', "proc_access", "high"),
|
||||
("probe.py", 'os.listdir("/proc/self/fd")\n', "proc_access", "high"),
|
||||
("probe.py", 'open("/proc/1/root/home/u/.netrc").read()\n', "proc_access", "high"),
|
||||
("probe.py", 'print(Path("/proc/self").read_text())\n', "proc_access", "high"),
|
||||
("probe.py", '(Path("/proc/self/cgroup"), Path("/proc/1/cmdline"))\n', "proc_access", "high"),
|
||||
("writer.py", 'Path("/proc/self/cgroup").write_text("x")\n', "proc_access", "high"),
|
||||
("keys.py", f'API_KEY = "phx_{_FAKE_KEY_BODY}"\n', "hardcoded_secret", "critical"),
|
||||
("keys.py", f'api_key = "phc_{_FAKE_KEY_BODY}"; token = "a8f3k2m9q7w1z5x0c4v6b8n2"\n',
|
||||
"hardcoded_secret", "critical"),
|
||||
])
|
||||
def test_attack_twins_keep_their_severity(self, tmp_path, rel, text, pattern_id, severity):
|
||||
result = scan_plugin(_mk_plugin(tmp_path, {**BASE_FILES, rel: text}), source="owner/repo")
|
||||
hit = next((f for f in result.findings if f.pattern_id == pattern_id), None)
|
||||
assert hit is not None and hit.severity == severity, format_scan_report(result)
|
||||
|
||||
def test_committed_bytecode_flags_but_a_cache_the_validate_run_wrote_does_not(self, tmp_path):
|
||||
"""Catalog rule 17 (hermes-emotion shipped tests/__pycache__/*.pyc): tracked bytecode runs in
|
||||
place of the scanned source. ``hermes plugins validate`` imports the plugin before it scans,
|
||||
so an untracked ``__pycache__`` is that run's own output and must not read as shipped."""
|
||||
plugin = _mk_plugin(tmp_path, {**BASE_FILES, "tests/__pycache__/t.cpython-312.pyc": "x",
|
||||
"__pycache__/__init__.cpython-314.pyc": "x"})
|
||||
subprocess.run(["git", "init", "-q", str(plugin)], check=True)
|
||||
subprocess.run(["git", "-C", str(plugin), "add", "-f", "plugin.yaml", "__init__.py",
|
||||
"tests/__pycache__/t.cpython-312.pyc"], check=True)
|
||||
|
||||
result = scan_plugin(plugin, source="owner/repo")
|
||||
|
||||
assert [(f.pattern_id, f.file, f.severity) for f in result.findings] == [
|
||||
("bytecode_file", "tests/__pycache__/t.cpython-312.pyc", "high")], format_scan_report(result)
|
||||
assert result.verdict == "caution"
|
||||
+37
-23
@@ -11,7 +11,8 @@ needs confirmation, ``dangerous`` is blocked and ``--force`` does NOT override.
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
import re
|
||||
import os
|
||||
import subprocess
|
||||
from datetime import datetime, timezone, UTC
|
||||
from pathlib import Path
|
||||
from typing import Iterator, List, Optional, Tuple
|
||||
@@ -19,9 +20,9 @@ from typing import Iterator, List, Optional, Tuple
|
||||
from tools.plugin_guard_context import (
|
||||
STEP_DOWN, catalog_cap, is_agent_facing, is_authorized_keys_prose, is_base64_archive_roundtrip,
|
||||
is_base64_media, is_benign_lazy_os_import, is_ci_workflow, is_coin_name_only,
|
||||
is_disclosed_config_note, is_doc_prose, is_env_dump_in_safety_clause, is_google_installed_app_secret,
|
||||
is_hex_in_char_class, is_inert_fixture_line, is_json_prose_value, is_locale_catalog,
|
||||
is_loopback_continuation, is_loopback_only, is_pip_install_in_prose_literal,
|
||||
is_disclosed_config_note, is_doc_prose, is_env_dump_in_safety_clause, is_hex_in_char_class,
|
||||
is_inert_fixture_line, is_json_prose_value, is_locale_catalog, is_loopback_continuation,
|
||||
is_loopback_only, is_own_proc_read, is_pip_install_in_prose_literal, is_public_client_key,
|
||||
is_quoted_injection_phrase, is_regex_alternation_token, is_repo_root_traversal,
|
||||
is_self_uninstall_doc, is_sudo_inert, is_test_tree, is_trusted_installer_pipe, logical_line, prose_cap)
|
||||
from tools.skills_guard import (
|
||||
@@ -108,6 +109,11 @@ MAX_PLUGIN_FILE_COUNT = 400
|
||||
MAX_PLUGIN_TOTAL_SIZE_KB = 10 * 1024 # 10MB of scannable tree
|
||||
MAX_PLUGIN_SINGLE_FILE_KB = 1024 # 1MB single file
|
||||
|
||||
# Committed bytecode (catalog rule 17): an unchecked-hash or sourceless ``.pyc`` is imported in
|
||||
# place of the ``.py`` this scanner read, so it is the same reviewable-but-unscannable tier as a
|
||||
# bundled binary (``SEVERITY_REMAP["binary_file"]``).
|
||||
_BYTECODE_SUFFIXES = (".pyc", ".pyo")
|
||||
|
||||
|
||||
def _walk(plugin_dir: Path) -> Iterator[tuple[Path, str]]:
|
||||
"""Yield (path, "a/b/c" relative path) for every non-excluded entry under plugin_dir."""
|
||||
@@ -182,8 +188,6 @@ def _filter_findings(findings: list[Finding], rel_path: str, file_path: Path) ->
|
||||
f.severity = DOC_PROSE_DEMOTIONS[f.pattern_id]
|
||||
line = lines[f.line - 1] if 0 < f.line <= len(lines) else f.match
|
||||
joined = logical_line(lines, f.line - 1) if 0 < f.line <= len(lines) else line
|
||||
if f.pattern_id == "proc_access" and _is_benign_proc_read(joined):
|
||||
continue
|
||||
f.severity = _context_severity(f, rel_path, line, joined, doc_prose or locale_catalog, is_code,
|
||||
locale_catalog)
|
||||
if _is_defensive_documentation(f, rel_path):
|
||||
@@ -206,20 +210,6 @@ def _filter_findings(findings: list[Finding], rel_path: str, file_path: Path) ->
|
||||
|
||||
_SEVERITY_RANK = {"low": 0, "medium": 1, "high": 2, "critical": 3}
|
||||
|
||||
# /proc files every process may read about itself: cgroup membership and mount layout.
|
||||
# Container CPU-quota detection needs them to find the controller hierarchy, and language
|
||||
# runtimes read them for container resource discovery, so a read-only mention is not
|
||||
# process introspection (#144123). A write or any other /proc path keeps the finding.
|
||||
_BENIGN_PROC_PATHS = frozenset({"/proc/self/cgroup", "/proc/self/mountinfo"})
|
||||
_PROC_PATH_RE = re.compile(r"/proc/(?:self|\d+)/[\w.\-/]*")
|
||||
_PROC_WRITE_RE = re.compile(r"write_text|write_bytes|\.write\s*\(|open\s*\([^)]*['\"][wa]|>\s*/proc/")
|
||||
|
||||
|
||||
def _is_benign_proc_read(text: str) -> bool:
|
||||
"""True when every /proc path on *text* is the cgroup/mountinfo pair and none is written."""
|
||||
paths = [m.group(0).rstrip("/") for m in _PROC_PATH_RE.finditer(text)]
|
||||
return bool(paths) and all(p in _BENIGN_PROC_PATHS for p in paths) and not _PROC_WRITE_RE.search(text)
|
||||
|
||||
|
||||
def _at_most(severity: str, cap: str) -> str:
|
||||
"""Lower *severity* to *cap*; never raise it."""
|
||||
@@ -275,6 +265,7 @@ def _inert_context_cap(f: Finding, rel_path: str, line: str, joined: str, is_cod
|
||||
or is_loopback_continuation(f, line, joined) # loopback health check over a `\`-continuation
|
||||
or is_hex_in_char_class(f, line) # `[\x00-\x1F]`: a control-char filter, not a payload
|
||||
or (is_code and is_pip_install_in_prose_literal(f, line)) # "no pip install is needed" message
|
||||
or is_own_proc_read(f, joined) # `/proc/self/cgroup`: container placement, read only
|
||||
)
|
||||
if to_low:
|
||||
return "low"
|
||||
@@ -292,8 +283,8 @@ def _inert_context_cap(f: Finding, rel_path: str, line: str, joined: str, is_cod
|
||||
)
|
||||
if to_medium:
|
||||
return _at_most(sev, "medium")
|
||||
if is_google_installed_app_secret(f, line):
|
||||
return _at_most(sev, "high") # public installed-app OAuth client secret, reviewable caution
|
||||
if is_public_client_key(f, line):
|
||||
return _at_most(sev, "high") # public-by-design client key (GOCSPX-, phc_), reviewable caution
|
||||
return sev
|
||||
|
||||
|
||||
@@ -324,9 +315,32 @@ def _dangerous_findings_summary(findings: list[Finding]) -> str:
|
||||
return f"{len(critical)} critical of {len(findings)} findings{names}"
|
||||
|
||||
|
||||
def _tracked_bytecode(plugin_dir: Path) -> set[str]:
|
||||
"""``.pyc``/``.pyo`` the checkout tracks. ``hermes plugins validate`` imports the plugin before
|
||||
it scans, so an untracked ``__pycache__`` that run wrote is not shipped content; only the index
|
||||
says what the pinned commit carries. Not a checkout (or no git) → empty."""
|
||||
try:
|
||||
out = subprocess.run(
|
||||
["git", "-c", "core.fsmonitor=false", "-C", str(plugin_dir), "ls-files", "-z", "--", "*.pyc", "*.pyo"],
|
||||
capture_output=True, check=True, timeout=60, stdin=subprocess.DEVNULL,
|
||||
env={**os.environ, "GIT_OPTIONAL_LOCKS": "0"}).stdout
|
||||
except (OSError, subprocess.SubprocessError):
|
||||
return set()
|
||||
return {os.fsdecode(p) for p in out.split(b"\0") if p}
|
||||
|
||||
|
||||
def _bytecode_findings(plugin_dir: Path) -> list[Finding]:
|
||||
"""Tracked bytecode anywhere, plus bytecode outside ``__pycache__`` (CPython 3 never writes it
|
||||
there, so it was shipped even when the tree is not a checkout)."""
|
||||
loose = {rel for f, rel in _walk(plugin_dir) if rel.endswith(_BYTECODE_SUFFIXES) and f.is_file()}
|
||||
return [_finding("bytecode_file", SEVERITY_REMAP["binary_file"], "structural", rel, "committed bytecode",
|
||||
"committed Python bytecode runs in place of the scanned source (catalog rule 17)")
|
||||
for rel in sorted(loose | _tracked_bytecode(plugin_dir))]
|
||||
|
||||
|
||||
def _check_plugin_structure(plugin_dir: Path) -> list[Finding]:
|
||||
"""Structural checks sized for plugin repositories."""
|
||||
findings: list[Finding] = []
|
||||
findings: list[Finding] = _bytecode_findings(plugin_dir)
|
||||
file_count = 0
|
||||
total_size = 0
|
||||
resolved_root = plugin_dir.resolve()
|
||||
|
||||
@@ -760,16 +760,44 @@ def is_coin_name_only(finding: Finding, line: str) -> bool:
|
||||
return bool(hits) and all(_is_standalone_coin_name(line, h.start(), h.end()) for h in hits)
|
||||
|
||||
|
||||
# Google mints installed-app (desktop/CLI) OAuth clients with a ``GOCSPX-`` secret that its OAuth
|
||||
# docs say cannot be kept confidential: every copy of the app ships it (gcloud does). A plugin is
|
||||
# such an app, so the literal identifies a public client rather than leaking a credential; it stays
|
||||
# in the report for review (caution) instead of hard-blocking the install.
|
||||
_GOOGLE_PUBLIC_CLIENT_SECRET = re.compile(r'["\']GOCSPX-[A-Za-z0-9_-]{20,}["\']')
|
||||
# Key shapes the vendor documents as public, shipped in every copy of a client: a Google
|
||||
# installed-app (desktop/CLI) OAuth client secret (``GOCSPX-``, gcloud ships one) and a PostHog
|
||||
# project API key (``phc_``, a write-only ingest key every web page embeds). The literal identifies
|
||||
# a client rather than leaking a credential, so it stays in the report for review (caution) instead
|
||||
# of hard-blocking the install. Only when EVERY ``hardcoded_secret`` literal on the line has one of
|
||||
# these prefixes: a PostHog personal key (``phx_``) or any other token beside it keeps critical.
|
||||
_PUBLIC_CLIENT_KEY_PREFIXES = ("GOCSPX-", "phc_")
|
||||
_SECRET_LITERAL = re.compile(r'[=:]\s*["\']([A-Za-z0-9+/=_-]{20,})')
|
||||
|
||||
|
||||
def is_google_installed_app_secret(finding: Finding, line: str) -> bool:
|
||||
"""A ``hardcoded_secret`` hit whose literal is a Google installed-app OAuth client secret."""
|
||||
return finding.pattern_id == "hardcoded_secret" and bool(_GOOGLE_PUBLIC_CLIENT_SECRET.search(line))
|
||||
def is_public_client_key(finding: Finding, line: str) -> bool:
|
||||
"""A ``hardcoded_secret`` hit whose every matched literal is a public-by-design client key."""
|
||||
if finding.pattern_id != "hardcoded_secret":
|
||||
return False
|
||||
literals = [_SECRET_LITERAL.search(m.group(0)) for m in _PATTERN_BY_ID["hardcoded_secret"].finditer(line)]
|
||||
return bool(literals) and all(
|
||||
lit is not None and lit.group(1).startswith(_PUBLIC_CLIENT_KEY_PREFIXES) for lit in literals)
|
||||
|
||||
|
||||
# ``/proc/self/<leaf>`` and ``/proc/<pid>/<leaf>`` files that describe scheduling and container
|
||||
# placement (cgroup membership, mount layout, start time in ``stat``, limits): container CPU-quota
|
||||
# detection and pid-reuse checks read them. A read of one of these is not process introspection.
|
||||
# The leaves that leak or reach another process keep ``high``: ``environ``, ``cmdline``, ``mem``,
|
||||
# ``maps``, ``fd``, ``root``/``cwd`` (procfs escape into the host tree), ``exe``, ``ns`` and any
|
||||
# leaf not listed, as do a bare ``/proc/self`` and any write.
|
||||
_BENIGN_PROC_LEAVES = frozenset({
|
||||
"cgroup", "mountinfo", "mounts", "stat", "statm", "status", "limits", "io", "comm", "cpuset",
|
||||
"sched", "schedstat", "oom_score"})
|
||||
_PROC_PATH = re.compile(r"/proc/(?:self|\d+)(?:/([\w.-]*))?")
|
||||
_PROC_WRITE = re.compile(r"write_text|write_bytes|\.write\s*\(|open\s*\([^)]*['\"][wax+]|>\s*/proc/|\btee\b")
|
||||
|
||||
|
||||
def is_own_proc_read(finding: Finding, line: str) -> bool:
|
||||
"""A ``proc_access`` hit where every ``/proc`` path on the line is a benign leaf, read only."""
|
||||
if finding.pattern_id != "proc_access":
|
||||
return False
|
||||
leaves = [m.group(1) for m in _PROC_PATH.finditer(line)]
|
||||
return bool(leaves) and all(leaf in _BENIGN_PROC_LEAVES for leaf in leaves) and not _PROC_WRITE.search(line)
|
||||
|
||||
|
||||
__all__ = [
|
||||
@@ -787,13 +815,14 @@ __all__ = [
|
||||
"is_disclosed_config_note",
|
||||
"is_doc_prose",
|
||||
"is_env_dump_in_safety_clause",
|
||||
"is_google_installed_app_secret",
|
||||
"is_hex_in_char_class",
|
||||
"is_inert_fixture_line",
|
||||
"is_json_prose_value",
|
||||
"is_loopback_continuation",
|
||||
"is_loopback_only",
|
||||
"is_own_proc_read",
|
||||
"is_pip_install_in_prose_literal",
|
||||
"is_public_client_key",
|
||||
"is_quoted_injection_phrase",
|
||||
"is_regex_alternation_token",
|
||||
"is_repo_root_traversal",
|
||||
|
||||
@@ -116,8 +116,11 @@ the gallery. There is no separate listing to maintain.
|
||||
(querying `data-slot` / `data-tour` / `data-sidebar` / `data-testid`
|
||||
elements from `document`, or a `document.body` MutationObserver, to restyle,
|
||||
hide, click or rewrite core UI). `hermes plugins validate` refuses these at
|
||||
admission (`desktop surface` check); a plugin that needs a capability the
|
||||
SDK lacks asks for an SDK hook instead of patching around it.
|
||||
admission (`desktop surface` check) and warns, for the reviewer, on raw
|
||||
`window.hermesDesktop` bridge calls and on the app's markup selected any
|
||||
other way (`[data-slot]`, `[data-tree-*]` in CSS text or `closest()`); a
|
||||
plugin that needs a capability the SDK lacks asks for an SDK hook instead
|
||||
of patching around it.
|
||||
9. **No runtime overrides of Hermes core.** A listed plugin extends Hermes only
|
||||
through public surfaces: hooks, middleware, provider profiles and the
|
||||
other `register_*` APIs, and Desktop SDK slots and routes. It must not
|
||||
@@ -127,7 +130,8 @@ the gallery. There is no separate listing to maintain.
|
||||
patching the same seam silently break each other, and every core release
|
||||
can break both. `hermes plugins validate` refuses these at admission (`no
|
||||
core override` check), and warns (without failing) on each import of an
|
||||
undocumented core module, by file and line. If the hook you need does not
|
||||
undocumented core module and each call to a private method of a core
|
||||
object it is handed (`db._execute_write(...)`), by file and line. If the hook you need does not
|
||||
exist, open an issue describing it: we would rather add the seam than list
|
||||
a patch.
|
||||
10. **Dependency security policy is the plugin's.** Hermes's 3-day
|
||||
|
||||
Reference in new issue
Block a user