diff --git a/strix/config/settings.py b/strix/config/settings.py index e53d125ce..9c00a46b3 100644 --- a/strix/config/settings.py +++ b/strix/config/settings.py @@ -71,6 +71,32 @@ class DedupeSettings(BaseSettings): ) +class DepVerifySettings(BaseSettings): + """Deterministic dependency version-range verification (report/dep_verify.py). + + A dependency-CVE false positive is a FACTUAL question — is the installed + version actually in the advisory's affected range? — not a code-reasoning one. + So this is a deterministic check (no LLM): before a dependency report is + persisted, ask an advisory provider which advisories affect the exact + installed version; if the cited CVE/GHSA isn't among them, the finding is out + of range and gets rejected. OFF by default; opt-in. + + PROVIDER-PLUGGABLE — not everyone can/will call a hosted advisory API + (air-gapped scans, data-residency rules, private advisory DBs). ``provider``: + "osv" -> query an OSV-schema API (default https://api.osv.dev; override + ``osv_url`` for a self-hosted OSV mirror — same /v1/query contract). + "none" -> disabled (== enabled=False). + Fail-open throughout: any uncertainty emits the finding (never suppress a real + one on a provider hiccup / coverage gap). + """ + + model_config = _BASE_CONFIG + + enabled: bool = Field(default=False, alias="STRIX_DEP_VERIFY") + provider: str = Field(default="osv", alias="STRIX_DEP_VERIFY_PROVIDER") + osv_url: str = Field(default="https://api.osv.dev/v1/query", alias="STRIX_OSV_URL") + + class ContextSettings(BaseSettings): """Context-window management: per-tool-output caps and history compaction.""" @@ -134,6 +160,7 @@ class Settings(BaseSettings): llm: LlmSettings = Field(default_factory=LlmSettings) dedupe: DedupeSettings = Field(default_factory=DedupeSettings) + dep_verify: DepVerifySettings = Field(default_factory=DepVerifySettings) runtime: RuntimeSettings = Field(default_factory=RuntimeSettings) context: ContextSettings = Field(default_factory=ContextSettings) telemetry: TelemetrySettings = Field(default_factory=TelemetrySettings) diff --git a/strix/report/dep_verify.py b/strix/report/dep_verify.py new file mode 100644 index 000000000..48c207fc7 --- /dev/null +++ b/strix/report/dep_verify.py @@ -0,0 +1,184 @@ +"""Deterministic version-range verify for dependency-CVE findings. + +The code-sink verifier (report/verify.py) answers a REASONING question — is the +sink reachable + the control complete. A dependency-CVE false positive is a +different, FACTUAL question: is the installed version actually in the advisory's +vulnerable range? That's deterministic — a range check, no LLM — and more reliable +than any model for this sub-class (version-range containment is exact). This is +the "promote to a deterministic control" pattern. + +FP shape this catches: an agent files "CVE-XXXX in pkg@1.2.3" but 1.2.3 is OUT of +the advisory's affected range (already patched, or the CVE was mis-attributed to +the package). The provider is asked "which advisories affect pkg@version"; if the +finding's CVE/GHSA isn't among them, the installed version is not vulnerable → reject. + +PROVIDER-PLUGGABLE (Strix is global FOSS — not everyone can/will call a hosted +advisory API: air-gapped scans, data-residency rules, orgs with their own advisory +DB). The provider is selected by DepVerifySettings.provider: + "osv" -> query an OSV-schema endpoint (default api.osv.dev; override + STRIX_OSV_URL for a self-hosted OSV mirror — identical /v1/query + contract, so no code change, just the URL). + "none" -> disabled. +An AdvisoryProvider returns the set of advisory ids/aliases affecting a given +package@version, or None when it cannot answer (→ fail-open, emit). + +SAFETY (fail-open, asymmetric — never suppress a real dep finding): + - reject ONLY when the provider gives a definitive, non-empty answer AND the + finding's CVE/GHSA is provably absent from it; + - any uncertainty EMITS: provider disabled/unreachable/errors, package/version/ + ecosystem unparseable, non-CVE/GHSA id, empty result (coverage gap), or the + provider lists the CVE → return None (emit). + +No LLM. No extra package (OSV provider uses `requests`, already a core dep). +""" + +from __future__ import annotations + +import logging +import os +from typing import Any, Protocol + +import requests + +from strix.config import load_settings + + +logger = logging.getLogger(__name__) + +# Map common ecosystem strings (as agents/trivy emit them) to OSV's canonical set. +_ECOSYSTEM = { + "npm": "npm", "node": "npm", "javascript": "npm", "yarn": "npm", + "pypi": "PyPI", "pip": "PyPI", "python": "PyPI", + "go": "Go", "golang": "Go", "gomod": "Go", + "maven": "Maven", "java": "Maven", "gradle": "Maven", + "cargo": "crates.io", "rust": "crates.io", "crates.io": "crates.io", + "rubygems": "RubyGems", "ruby": "RubyGems", "gem": "RubyGems", + "nuget": "NuGet", "composer": "Packagist", "packagist": "Packagist", +} + + +def _norm_ecosystem(eco: str | None) -> str | None: + return _ECOSYSTEM.get((eco or "").strip().lower()) + + +def _ids_for(vuln: dict) -> set[str]: + """All identifiers an OSV advisory is known by (its id + aliases), upper-cased.""" + ids = {str(vuln.get("id", "")).upper()} + ids.update(str(a).upper() for a in (vuln.get("aliases") or [])) + return {i for i in ids if i} + + +class AdvisoryProvider(Protocol): + """Given a package coordinate, return the set of advisory ids/aliases (upper- + cased, e.g. {'CVE-...','GHSA-...'}) affecting THAT exact version — or None when + the provider cannot answer (→ caller fails open and emits the finding). + + An empty set means a definitive "no advisories affect this version"; None means + "don't know" (unreachable / error / not covered). The distinction matters: the + caller only rejects on a definitive, non-empty answer that omits the cited id.""" + + def affecting(self, pkg: str, ecosystem: str, version: str) -> set[str] | None: ... + + +class OsvProvider: + """AdvisoryProvider backed by an OSV-schema /v1/query endpoint. Default + api.osv.dev; point `url` at a self-hosted OSV mirror (identical contract) for + air-gapped / data-residency deployments. Uses `requests` (a core dep) — no + extra package.""" + + def __init__(self, url: str) -> None: + self.url = url + + def affecting(self, pkg: str, ecosystem: str, version: str) -> set[str] | None: + try: + ca = os.environ.get("REQUESTS_CA_BUNDLE") + resp = requests.post( + self.url, timeout=20, + json={"package": {"name": pkg, "ecosystem": ecosystem}, "version": version}, + verify=ca if ca else True, + ) + if resp.status_code != 200: + logger.info("dep-verify: OSV %s for %s@%s; can't answer", + resp.status_code, pkg, version) + return None + vulns = resp.json().get("vulns", []) or [] + except Exception: # noqa: BLE001 — advisory; a hiccup must never suppress a finding + logger.info("dep-verify: OSV query failed for %s@%s; can't answer", + pkg, version, exc_info=True) + return None + affecting: set[str] = set() + for v in vulns: + affecting |= _ids_for(v) + return affecting + + +def _resolve_provider(dep_settings: Any) -> AdvisoryProvider | None: + """Build the AdvisoryProvider from DepVerifySettings. None => disabled/unknown + provider => the check is a no-op (fail-open).""" + provider = (getattr(dep_settings, "provider", "osv") or "").strip().lower() + if provider in ("", "none", "off"): + return None + if provider == "osv": + return OsvProvider(getattr(dep_settings, "osv_url", "https://api.osv.dev/v1/query")) + logger.info("dep-verify: unknown provider %r; check disabled", provider) + return None + + +def verify_dependency( # noqa: PLR0911 — the returns are deliberate fail-open guard clauses + candidate: dict[str, Any], provider: AdvisoryProvider | None = None +) -> dict[str, Any] | None: + """Deterministically check a dep-CVE finding's version is in the CVE's range. + + candidate needs: cve (or ghsa), package_name, installed_version, package_ecosystem. + `provider` overrides the settings-resolved one (used by tests); when omitted it + is built from DepVerifySettings. Returns a REJECT dict when the installed + version is provably NOT affected by the cited advisory; None (emit) on any + uncertainty (fail-open).""" + cve = str(candidate.get("cve") or "").strip().upper() + ghsa = str(candidate.get("ghsa") or "").strip().upper() + ident = cve or ghsa + pkg = str(candidate.get("package_name") or "").strip() + version = str(candidate.get("installed_version") or "").strip() + eco = _norm_ecosystem(candidate.get("package_ecosystem")) + + # Need all four to make a definitive call; otherwise emit. + if not (ident and pkg and version and eco): + return None + # Only CVE/GHSA identifiers are resolvable by advisory providers. + if not ident.startswith(("CVE-", "GHSA-")): + return None + + if provider is None: + provider = _resolve_provider(load_settings().dep_verify) + if provider is None: + return None # disabled / unknown provider → emit + + affecting = provider.affecting(pkg, eco, version) + if affecting is None: + return None # provider couldn't answer → fail open, emit + if ident in affecting: + return None # confirmed: installed version IS in the CVE's range → emit (real) + if not affecting: + # definitive "no advisories affect this version" is still a coverage-gap + # risk (private/vendored pkg, provider lag) → fail open, emit. + logger.info("dep-verify: provider lists NO advisories for %s@%s; emitting " + "(coverage gap, not a confident FP)", pkg, version) + return None + + # Provider knows advisories for this version but NOT the cited one → the + # installed version is out of the cited CVE's range → false positive. + return { + "success": False, + "error": ( + f"Version-range verify rejected this dependency finding: {ident} does " + f"not affect {pkg}@{version} per the advisory provider (installed " + f"version is outside the advisory's vulnerable range — likely already " + f"patched or the CVE is mis-attributed to this package). Advisories " + f"affecting {pkg}@{version}: {sorted(affecting)}. If you believe the " + f"version IS vulnerable, cite the exact affected range." + ), + "verify_rejected": True, + "dep_version_out_of_range": True, + "installed_version": version, + "cited_advisory": ident, + } diff --git a/strix/tools/reporting/tool.py b/strix/tools/reporting/tool.py index b7bb18850..11f93b5ac 100644 --- a/strix/tools/reporting/tool.py +++ b/strix/tools/reporting/tool.py @@ -820,6 +820,26 @@ async def _do_create_dependency( # noqa: PLR0912 "reason": dedupe.get("reason", ""), } + # Deterministic version-range verify (opt-in, STRIX_DEP_VERIFY). A + # dependency-CVE false positive is a factual question — is the installed + # version actually in the advisory's affected range? — so it's checked + # against an advisory provider (OSV by default), no LLM. Fail-open: only + # rejects when the provider proves the version is OUT of the cited CVE's + # range. Sibling of the dedup-reject above. + from strix.config import load_settings + if load_settings().dep_verify.enabled: + from strix.report.dep_verify import verify_dependency + + dep_verdict = verify_dependency({ + "cve": parsed_cve, "package_name": package_name, + "installed_version": installed_version, + "package_ecosystem": package_ecosystem, + }) + if dep_verdict is not None: + logger.info("dep-verify rejected %s %s@%s (version out of range)", + parsed_cve, package_name, installed_version) + return dep_verdict + report_id = report_state.add_vulnerability_report( title=title, description=description, diff --git a/tests/test_dep_verify.py b/tests/test_dep_verify.py new file mode 100644 index 000000000..fd875fcc4 --- /dev/null +++ b/tests/test_dep_verify.py @@ -0,0 +1,157 @@ +"""Deterministic dependency version-range verify (provider-pluggable, OSV default). + +Load-bearing behaviours (a SAFE FP reducer for dep-CVE findings): + 1. In-range -> emit (real). Provider lists the CVE for the installed version. + 2. Out-of-range -> REJECT. Provider knows advisories for the version but NOT the + cited CVE (installed version outside the range / mis-attributed). + 3. Fail-open everywhere: provider can't answer (None), empty set (coverage gap), + missing fields, non-CVE/GHSA id, unknown ecosystem, disabled provider -> emit. + Never suppress a real dep finding on uncertainty. + 4. Provider-pluggable: OSV default, custom URL for mirrors, 'none' disables. +""" + +from __future__ import annotations + +import requests + +from strix.report import dep_verify as dv + + +def _cand(**kw): + base = {"cve": "CVE-2021-23337", "package_name": "lodash", + "installed_version": "4.17.20", "package_ecosystem": "npm"} + base.update(kw) + return base + + +class _FakeProvider: + """Returns a fixed affecting-set (or None for can't-answer).""" + def __init__(self, affecting): + self._a = affecting + + def affecting(self, _pkg, _ecosystem, _version): + return self._a + + +def _prov(affecting): + return _FakeProvider(affecting) + + +# --- verdict logic (explicit provider, no network) --- + +def test_in_range_emits(): + # provider lists the cited CVE as affecting this version -> real, emit. + assert dv.verify_dependency(_cand(), _prov({"GHSA-X", "CVE-2021-23337"})) is None + + +def test_out_of_range_rejects(): + # provider knows advisories for this version but NOT the cited CVE -> out of range. + out = dv.verify_dependency(_cand(), _prov({"GHSA-OTHER", "CVE-2020-0000"})) + assert out is not None + assert out["verify_rejected"] is True + assert out["dep_version_out_of_range"] is True + assert "CVE-2021-23337" in out["error"] + + +def test_empty_set_fails_open(): + # definitive "no advisories affect this version" -> coverage-gap risk -> emit. + assert dv.verify_dependency(_cand(), _prov(set())) is None + + +def test_provider_cant_answer_fails_open(): + # None = unreachable/error/not-covered -> emit. + assert dv.verify_dependency(_cand(), _prov(None)) is None + + +def test_matches_via_alias_or_id(): + assert dv.verify_dependency(_cand(cve="", ghsa="GHSA-JF85-CPCP-J695"), + _prov({"GHSA-JF85-CPCP-J695"})) is None + + +def test_missing_fields_emit(): + for miss in ("cve", "package_name", "installed_version", "package_ecosystem"): + c = _cand() + c[miss] = "" + if miss == "cve": + c["ghsa"] = "" + # provider that WOULD reject, to prove the guard short-circuits before it + assert dv.verify_dependency(c, _prov({"CVE-9999-0000"})) is None, miss + + +def test_unknown_ecosystem_emits(): + assert dv.verify_dependency(_cand(package_ecosystem="cocoapods-weird"), + _prov({"CVE-9999-0000"})) is None + + +def test_non_cve_identifier_emits(): + assert dv.verify_dependency(_cand(cve="RUSTSEC-2021-0001"), + _prov({"CVE-9999-0000"})) is None + + +# --- ecosystem normalisation --- + +def test_ecosystem_normalization(): + assert dv._norm_ecosystem("PyPI") == "PyPI" + assert dv._norm_ecosystem("pip") == "PyPI" + assert dv._norm_ecosystem("golang") == "Go" + assert dv._norm_ecosystem("node") == "npm" + assert dv._norm_ecosystem("bogus") is None + + +# --- provider resolution (the pluggable toggle) --- + +class _S: + def __init__(self, provider="osv", osv_url="https://api.osv.dev/v1/query"): + self.provider = provider + self.osv_url = osv_url + + +def test_resolve_provider_osv_default(): + p = dv._resolve_provider(_S()) + assert isinstance(p, dv.OsvProvider) + assert p.url == "https://api.osv.dev/v1/query" + + +def test_resolve_provider_custom_mirror_url(): + p = dv._resolve_provider(_S(osv_url="https://osv.internal.corp/v1/query")) + assert isinstance(p, dv.OsvProvider) + assert p.url == "https://osv.internal.corp/v1/query" + + +def test_resolve_provider_none_disables(): + assert dv._resolve_provider(_S(provider="none")) is None + assert dv._resolve_provider(_S(provider="")) is None + + +def test_resolve_provider_unknown_disables(): + assert dv._resolve_provider(_S(provider="snyk")) is None + + +def test_osv_provider_parses_query(monkeypatch): + # OsvProvider.affecting: mock requests.post, assert it flattens ids+aliases. + class _R: + status_code = 200 + + def json(self): + return {"vulns": [{"id": "GHSA-a", "aliases": ["CVE-2021-23337"]}, + {"id": "GHSA-b"}]} + monkeypatch.setattr(requests, "post", lambda *_a, **_k: _R()) + got = dv.OsvProvider("https://api.osv.dev/v1/query").affecting("lodash", "npm", "4.17.20") + assert got == {"GHSA-A", "CVE-2021-23337", "GHSA-B"} + + +def test_osv_provider_non200_returns_none(monkeypatch): + class _R: + status_code = 503 + + def json(self): + return {} + monkeypatch.setattr(requests, "post", lambda *_a, **_k: _R()) + assert dv.OsvProvider("u").affecting("p", "npm", "1.0") is None + + +def test_osv_provider_error_returns_none(monkeypatch): + def _boom(*_a, **_k): + raise requests.RequestException("timeout") + monkeypatch.setattr(requests, "post", _boom) + assert dv.OsvProvider("u").affecting("p", "npm", "1.0") is None