Merge 43a2886c6e into 657aa5cbe6

strix/config/settings.py

@@ -76,6 +76,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."""
 
@@ -143,6 +169,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)

strix/report/dep_verify.py

@@ -0,0 +1,220 @@
+"""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: ...
+
+    def knows_package(self, pkg: str, ecosystem: str) -> bool | None:
+        """Does the provider have ANY advisory for this package (any version)?
+        Lets the caller tell a CONFIDENT out-of-range (provider knows the package
+        but nothing affects the installed version) from a COVERAGE GAP (provider
+        doesn't know the package at all → fail open). None = can't determine."""
+        ...
+
+
+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.
+
+    The caller runs this off the event loop (asyncio.to_thread), but keep the
+    per-request timeout short so a stalled provider fails open quickly rather than
+    holding a worker thread — this is a bounded advisory lookup, not bulk I/O."""
+
+    _TIMEOUT_S = 10
+
+    def __init__(self, url: str) -> None:
+        self.url = url
+
+    def _query(self, payload: dict[str, Any]) -> list[dict] | None:
+        try:
+            ca = os.environ.get("REQUESTS_CA_BUNDLE")
+            resp = requests.post(self.url, timeout=self._TIMEOUT_S, json=payload,
+                                 verify=ca if ca else True)
+            if resp.status_code != 200:
+                logger.info("dep-verify: OSV %s for %s; can't answer",
+                            resp.status_code, payload.get("package"))
+                return None
+            return 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; can't answer",
+                        payload.get("package"), exc_info=True)
+            return None
+
+    def affecting(self, pkg: str, ecosystem: str, version: str) -> set[str] | None:
+        vulns = self._query(
+            {"package": {"name": pkg, "ecosystem": ecosystem}, "version": version})
+        if vulns is None:
+            return None
+        affecting: set[str] = set()
+        for v in vulns:
+            affecting |= _ids_for(v)
+        return affecting
+
+    def knows_package(self, pkg: str, ecosystem: str) -> bool | None:
+        # version-less query returns every advisory for the package (any version).
+        vulns = self._query({"package": {"name": pkg, "ecosystem": ecosystem}})
+        if vulns is None:
+            return None
+        return len(vulns) > 0
+
+
+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:
+        logger.info("dep-verify: %s confirmed in range for %s@%s; emitting (real)",
+                    ident, pkg, version)
+        return None  # confirmed: installed version IS in the CVE's range → emit (real)
+    if not affecting:
+        # Nothing affects the installed version. Disambiguate:
+        #   - provider KNOWS the package (has advisories for OTHER versions) but
+        #     none hit this version → CONFIDENT out-of-range → reject.
+        #   - provider doesn't know the package at all (private/vendored, lag) →
+        #     genuine coverage gap → fail open, emit.
+        knows = None
+        try:
+            knows = provider.knows_package(pkg, eco)
+        except Exception:  # noqa: BLE001 — advisory
+            knows = None
+        if not knows:  # False (unknown pkg) or None (can't tell) → fail open
+            logger.info("dep-verify: provider has no advisories for %s at all; "
+                        "emitting (coverage gap, not a confident FP)", pkg)
+            return None
+        # else: package is known, this version is clean → fall through to reject.
+
+    # Provider knows advisories for this version but NOT the cited one (or knows
+    # the package and NOTHING affects this version) → 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) or 'none'}. If you "
+            f"believe the version IS vulnerable, cite the exact affected range."
+        ),
+        "verify_rejected": True,
+        "dep_version_out_of_range": True,
+        "installed_version": version,
+        "cited_advisory": ident,
+    }

strix/tools/reporting/tool.py

@@ -890,6 +890,30 @@ 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
+
+            # verify_dependency does blocking HTTP (advisory provider). Run it in a
+            # worker thread so a slow/unreachable provider can't stall the shared
+            # async agent loop (it would otherwise block every concurrent agent for
+            # up to the provider timeout, x2 on the clean-version path).
+            dep_verdict = await asyncio.to_thread(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,

tests/test_dep_verify.py

@@ -0,0 +1,178 @@
+"""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 (None for can't-answer). `knows` is what
+    knows_package returns (True=package known, False=unknown, None=can't tell)."""
+    def __init__(self, affecting, knows=None):
+        self._a = affecting
+        self._knows = knows
+
+    def affecting(self, _pkg, _ecosystem, _version):
+        return self._a
+
+    def knows_package(self, _pkg, _ecosystem):
+        return self._knows
+
+
+def _prov(affecting, knows=None):
+    return _FakeProvider(affecting, knows)
+
+
+# --- 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_unknown_package_fails_open():
+    # empty affecting + provider doesn't know the package (knows=False) ->
+    # genuine coverage gap (private/vendored) -> emit.
+    assert dv.verify_dependency(_cand(), _prov(set(), knows=False)) is None
+
+
+def test_empty_set_cant_tell_fails_open():
+    # empty affecting + knows_package can't answer (None) -> fail open, emit.
+    assert dv.verify_dependency(_cand(), _prov(set(), knows=None)) is None
+
+
+def test_empty_set_known_package_rejects():
+    # empty affecting BUT provider KNOWS the package (advisories on other versions)
+    # -> confident out-of-range -> REJECT. The minimist@1.2.6 case a live scan of a
+    # weak model exposed: OSV knows minimist but 0 advisories affect 1.2.6.
+    out = dv.verify_dependency(_cand(), _prov(set(), knows=True))
+    assert out is not None
+    assert out["verify_rejected"] is True
+    assert out["dep_version_out_of_range"] is True
+
+
+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

tests/test_reporting_fields.py

@@ -2,10 +2,14 @@
 
 from __future__ import annotations
 
+import threading
+import time
 from typing import TYPE_CHECKING
 
 import pytest
 
+import strix.report.dep_verify as depv
+from strix.config import loader
 from strix.report.dedupe import (
     _check_dependency_duplicate,
     _prepare_report_for_comparison,
@@ -255,6 +259,43 @@ async def test_dependency_report_with_zero_cvss_remains_low_severity(
     assert report["cvss"] == 0.0
 
 
+async def test_dependency_verify_runs_off_the_event_loop(
+    report_state: ReportState,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The verifier does blocking HTTP; it must be offloaded (asyncio.to_thread)
+    so a slow/unreachable provider can't stall the shared async agent loop. Stub
+    verify_dependency with a blocking call that records its thread + rejects, and
+    assert (a) the reject propagates and (b) it ran on a NON-main thread."""
+    monkeypatch.setenv("STRIX_DEP_VERIFY", "1")
+    loader._cached = None
+
+    main_thread = threading.current_thread()
+    ran_on: dict[str, object] = {}
+
+    def _blocking_verify(_candidate: dict) -> dict:
+        ran_on["thread"] = threading.current_thread()
+        time.sleep(0.05)  # simulate a slow provider (blocking)
+        return {"success": False, "verify_rejected": True,
+                "dep_version_out_of_range": True, "error": "out of range"}
+
+    monkeypatch.setattr(depv, "verify_dependency", _blocking_verify)
+
+    result = await _do_create_dependency(
+        title="CVE-2021-44906 in minimist 1.2.6",
+        description="Prototype pollution.", target="repo/package.json",
+        cve="CVE-2021-44906", package_name="minimist", installed_version="1.2.6",
+        impact="x", remediation_steps="upgrade", assumptions="a",
+        package_ecosystem="npm", fixed_version=None, cwe=None,
+        advisory_cvss=7.5, technical_analysis=None, fix_effort="low",
+    )
+    # reject propagated + finding NOT persisted
+    assert result["verify_rejected"] is True
+    assert not report_state.vulnerability_reports
+    # ran off the event loop (a worker thread, not the main thread)
+    assert ran_on["thread"] is not main_thread
+
+
 async def test_dependency_report_requires_advisory_cvss(report_state: ReportState) -> None:
     result = await _do_create_dependency(
         title="CVE-2024-0001 in sample 1.0.0",