fix(security): cache OSV malware preflight verdicts and stop double component discovery (#75485)

Two amplifiers behind the 779K api.osv.dev DNS queries/16h report:

1. tools/osv_check.py: check_package_for_malware() hit OSV on EVERY
   call. MCP reconnect ladders, stdio recycles, and parked-server
   self-probes re-run the preflight for the same package on every spawn
   attempt, so a flapping server became a sustained OSV query/DNS
   stream. Verdicts (clean or blocked) are now cached for 1h
   (OSV_CHECK_CACHE_TTL to tune); network failures stay uncached so
   fail-open never masks a real advisory once connectivity returns.

2. hermes_cli/security_audit.py: cmd_security_audit() ran full
   component discovery twice per audit (_count_components + run_audit).
   Discovery now runs once via _discover_components() and run_audit()
   accepts the pre-discovered list.

Both regression tests fail against the previous code (verified via
sabotage run).
This commit is contained in:
Teknium
2026-07-31 23:30:25 -07:00
parent cb1e059a98
commit 5eeafc8d25
4 changed files with 180 additions and 20 deletions
+30 -17
View File
@@ -411,14 +411,14 @@ def _osv_fetch_details(vuln_ids: Iterable[str]) -> dict[str, Vulnerability]:
# ─── Orchestration ────────────────────────────────────────────────────────────
def run_audit(
def _discover_components(
*,
skip_venv: bool = False,
skip_plugins: bool = False,
skip_mcp: bool = False,
hermes_home: Optional[Path] = None,
) -> list[Finding]:
"""Discover components, query OSV, return findings sorted by severity desc."""
) -> list[Component]:
"""Discover all scannable components across the enabled sources."""
home = hermes_home or Path(get_hermes_home())
components: list[Component] = []
if not skip_venv:
@@ -427,6 +427,30 @@ def run_audit(
components.extend(_discover_plugins(home))
if not skip_mcp:
components.extend(_discover_mcp())
return components
def run_audit(
*,
skip_venv: bool = False,
skip_plugins: bool = False,
skip_mcp: bool = False,
hermes_home: Optional[Path] = None,
components: Optional[list[Component]] = None,
) -> list[Finding]:
"""Query OSV for the given (or freshly discovered) components.
``components`` lets callers that already ran discovery (e.g. for a
component count) reuse it instead of scanning the venv/plugins/MCP
config a second time.
"""
if components is None:
components = _discover_components(
skip_venv=skip_venv,
skip_plugins=skip_plugins,
skip_mcp=skip_mcp,
hermes_home=hermes_home,
)
if not components:
return []
@@ -509,19 +533,6 @@ def _render_json(findings: list[Finding], total_components: int) -> str:
return json.dumps(payload, indent=2)
def _count_components(
*, skip_venv: bool, skip_plugins: bool, skip_mcp: bool, hermes_home: Path
) -> int:
total = 0
if not skip_venv:
total += len(_discover_venv())
if not skip_plugins:
total += len(_discover_plugins(hermes_home))
if not skip_mcp:
total += len(_discover_mcp())
return total
# ─── CLI entrypoint ───────────────────────────────────────────────────────────
@@ -541,9 +552,10 @@ def cmd_security_audit(args: argparse.Namespace) -> int:
)
return 2
total = _count_components(
components = _discover_components(
skip_venv=skip_venv, skip_plugins=skip_plugins, skip_mcp=skip_mcp, hermes_home=home
)
total = len(components)
if total == 0:
msg = "No components discovered (everything skipped, or empty environment)."
if output_json:
@@ -558,6 +570,7 @@ def cmd_security_audit(args: argparse.Namespace) -> int:
skip_plugins=skip_plugins,
skip_mcp=skip_mcp,
hermes_home=home,
components=components,
)
except RuntimeError as exc:
print(f"audit failed: {exc}", file=sys.stderr)