diff --git a/hermes_cli/security_audit.py b/hermes_cli/security_audit.py index f29c61f333240..78d8a8b7e0486 100644 --- a/hermes_cli/security_audit.py +++ b/hermes_cli/security_audit.py @@ -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) diff --git a/tests/hermes_cli/test_security_audit.py b/tests/hermes_cli/test_security_audit.py index 0643265c5c039..c993277db6fbc 100644 --- a/tests/hermes_cli/test_security_audit.py +++ b/tests/hermes_cli/test_security_audit.py @@ -153,6 +153,25 @@ class TestExitCodes: defaults.update(kwargs) return argparse.Namespace(**defaults) + def test_discovery_runs_once_per_audit(self, tmp_path: Path, monkeypatch, capsys): + """cmd_security_audit must not scan the venv/plugins/MCP config twice. + + Regression for the double-scan noted in #75485: the component count + and the audit each ran full discovery independently. + """ + monkeypatch.setattr(sa, "get_hermes_home", lambda: str(tmp_path)) + calls = {"venv": 0} + + def counting_discover_venv(): + calls["venv"] += 1 + return [sa.Component(name="pkg", version="1.0", ecosystem="PyPI", source="venv")] + + monkeypatch.setattr(sa, "_discover_venv", counting_discover_venv) + monkeypatch.setattr(sa, "_osv_query_batch", lambda comps: {}) + sa.cmd_security_audit(self._build_args(skip_venv=False)) + capsys.readouterr() + assert calls["venv"] == 1 + diff --git a/tests/tools/test_osv_check.py b/tests/tools/test_osv_check.py index a8ed186273741..72e81058e6678 100644 --- a/tests/tools/test_osv_check.py +++ b/tests/tools/test_osv_check.py @@ -65,6 +65,15 @@ class TestParsePackageFromArgs: class TestCheckPackageForMalware: + @pytest.fixture(autouse=True) + def _fresh_cache(self): + from tools import osv_check + with osv_check._cache_lock: + osv_check._cache.clear() + yield + with osv_check._cache_lock: + osv_check._cache.clear() + def test_clean_package(self): """Clean package returns None (allow).""" mock_response = MagicMock() @@ -110,6 +119,76 @@ class TestCheckPackageForMalware: assert call_data["package"]["ecosystem"] == "PyPI" assert call_data["package"]["name"] == "mcp-server-fetch" + def test_repeat_checks_hit_cache_not_network(self): + """Same package re-checked (MCP revival loops) must not re-query OSV. + + Regression for #75485: watchdog revival loops re-ran the preflight + every spawn attempt, producing 779K api.osv.dev DNS queries in 16h. + """ + mock_response = MagicMock() + mock_response.read.return_value = json.dumps({"vulns": []}).encode() + mock_response.__enter__ = lambda s: s + mock_response.__exit__ = MagicMock(return_value=False) + + with patch("tools.osv_check.urllib.request.urlopen", return_value=mock_response) as mock_url: + for _ in range(50): + assert check_package_for_malware("uvx", ["mcp-server-fetch"]) is None + assert mock_url.call_count == 1 + + def test_blocked_verdict_is_cached(self): + """A malware verdict is served from cache on re-check too.""" + mock_response = MagicMock() + mock_response.read.return_value = json.dumps( + {"vulns": [{"id": "MAL-2023-1", "summary": "bad"}]} + ).encode() + mock_response.__enter__ = lambda s: s + mock_response.__exit__ = MagicMock(return_value=False) + + with patch("tools.osv_check.urllib.request.urlopen", return_value=mock_response) as mock_url: + first = check_package_for_malware("npx", ["evil-pkg"]) + second = check_package_for_malware("npx", ["evil-pkg"]) + assert first is not None and "BLOCKED" in first + assert second == first + assert mock_url.call_count == 1 + + def test_network_failure_not_cached(self): + """Fail-open results must not be cached — retry once network is back.""" + mock_response = MagicMock() + mock_response.read.return_value = json.dumps({"vulns": []}).encode() + mock_response.__enter__ = lambda s: s + mock_response.__exit__ = MagicMock(return_value=False) + + with patch( + "tools.osv_check.urllib.request.urlopen", + side_effect=OSError("network down"), + ): + assert check_package_for_malware("uvx", ["mcp-server-time"]) is None + # Network is back: the next check must hit OSV, not a cached fail-open. + with patch( + "tools.osv_check.urllib.request.urlopen", return_value=mock_response + ) as mock_url: + assert check_package_for_malware("uvx", ["mcp-server-time"]) is None + assert mock_url.call_count == 1 + + def test_cache_expiry_requeries(self, monkeypatch): + """Expired entries re-query instead of serving stale verdicts.""" + from tools import osv_check + + mock_response = MagicMock() + mock_response.read.return_value = json.dumps({"vulns": []}).encode() + mock_response.__enter__ = lambda s: s + mock_response.__exit__ = MagicMock(return_value=False) + + with patch("tools.osv_check.urllib.request.urlopen", return_value=mock_response) as mock_url: + check_package_for_malware("uvx", ["mcp-server-fetch"]) + # Force-expire the entry. + with osv_check._cache_lock: + key = next(iter(osv_check._cache)) + _, result = osv_check._cache[key] + osv_check._cache[key] = (0.0, result) + check_package_for_malware("uvx", ["mcp-server-fetch"]) + assert mock_url.call_count == 2 + class TestLiveOsvQuery: """Live integration test against the real OSV API. Skipped if offline.""" diff --git a/tools/osv_check.py b/tools/osv_check.py index 1f8a986e11b69..b543dd7f33199 100644 --- a/tools/osv_check.py +++ b/tools/osv_check.py @@ -14,6 +14,8 @@ import json import logging import os import re +import threading +import time import urllib.request from typing import Optional, Tuple @@ -22,6 +24,44 @@ logger = logging.getLogger(__name__) _OSV_ENDPOINT = os.getenv("OSV_ENDPOINT", "https://api.osv.dev/v1/query") _TIMEOUT = 10 # seconds +# Result cache: (ecosystem, package, version) -> (expiry_monotonic, result). +# MCP reconnect ladders, stdio recycles, and parked-server self-probes re-run +# the preflight for the SAME package on every spawn attempt. Without a cache, +# a flapping server turns into a sustained OSV query/DNS stream — the #75485 +# incident logged 779K api.osv.dev DNS queries in 16h from revival loops. +# Malware advisories don't appear or vanish on second-to-second timescales, +# so a successful verdict (clean OR blocked) is reusable. Network failures +# are NOT cached: fail-open already covers them, and caching a failure could +# mask a real advisory once connectivity returns. +_CACHE_TTL_S = float(os.getenv("OSV_CHECK_CACHE_TTL", "3600")) +_CACHE_MAX_ENTRIES = 256 +_cache: dict = {} +_cache_lock = threading.Lock() + + +def _cache_get(key) -> Tuple[bool, Optional[str]]: + """Return (hit, result) for a fresh cache entry.""" + with _cache_lock: + entry = _cache.get(key) + if entry is None: + return False, None + expiry, result = entry + if time.monotonic() >= expiry: + del _cache[key] + return False, None + return True, result + + +def _cache_put(key, result: Optional[str]) -> None: + with _cache_lock: + if len(_cache) >= _CACHE_MAX_ENTRIES: + now = time.monotonic() + for k in [k for k, (exp, _) in _cache.items() if exp <= now]: + del _cache[k] + if len(_cache) >= _CACHE_MAX_ENTRIES: + _cache.clear() # tiny working set in practice; safe reset + _cache[key] = (time.monotonic() + _CACHE_TTL_S, result) + def check_package_for_malware( command: str, args: list @@ -43,10 +83,16 @@ def check_package_for_malware( if not package: return None + cache_key = (ecosystem, package, version) + hit, cached = _cache_get(cache_key) + if hit: + return cached + try: malware = _query_osv(package, ecosystem, version) except Exception as exc: - # Fail-open: network errors, timeouts, parse failures → allow + # Fail-open: network errors, timeouts, parse failures → allow. + # Deliberately NOT cached — see _CACHE_TTL_S comment. logger.debug("OSV check failed for %s/%s (allowing): %s", ecosystem, package, exc) return None @@ -55,11 +101,14 @@ def check_package_for_malware( summaries = "; ".join( m.get("summary", m["id"])[:100] for m in malware[:3] ) - return ( + result = ( f"BLOCKED: Package '{package}' ({ecosystem}) has known malware " f"advisories: {ids}. Details: {summaries}" ) - return None + else: + result = None + _cache_put(cache_key, result) + return result def _infer_ecosystem(command: str) -> Optional[str]: