diff --git a/strix/report/state.py b/strix/report/state.py index fd0ebd9449..dc52e0f83c 100644 --- a/strix/report/state.py +++ b/strix/report/state.py @@ -1,3 +1,4 @@ +import asyncio import json import logging import re @@ -210,6 +211,9 @@ def __init__(self, run_name: str | None = None): self.end_time: str | None = None self.vulnerability_reports: list[dict[str, Any]] = [] + # Held while a new finding is checked against the stored ones and then added, + # so two agents filing the same finding at once cannot both pass the check. + self.dedupe_lock = asyncio.Lock() self.final_scan_result: str | None = None self.scan_results: dict[str, Any] | None = None diff --git a/strix/tools/reporting/tool.py b/strix/tools/reporting/tool.py index 791bae7961..ab83a26b38 100644 --- a/strix/tools/reporting/tool.py +++ b/strix/tools/reporting/tool.py @@ -872,7 +872,6 @@ async def _do_create( from strix.report.dedupe import check_duplicate - existing = report_state.get_existing_vulnerabilities() candidate = { "title": title, "description": description, @@ -913,30 +912,35 @@ async def _do_create( "http_exchange_ids": normalized_http_exchange_ids, } - dedupe = await check_duplicate(candidate, existing) - if dedupe.get("is_duplicate"): - duplicate_id = str(dedupe.get("duplicate_id") or "") - duplicate_title = next( - (r.get("title", "Unknown") for r in existing if r.get("id") == duplicate_id), - "", + # The duplicate check awaits an LLM call. Without a lock, two agents filing + # the same finding at once each compare against a list the other has not + # joined yet and both store it, so read, check and store as one step. + async with report_state.dedupe_lock: + existing = report_state.get_existing_vulnerabilities() + dedupe = await check_duplicate(candidate, existing) + if dedupe.get("is_duplicate"): + duplicate_id = str(dedupe.get("duplicate_id") or "") + duplicate_title = next( + (r.get("title", "Unknown") for r in existing if r.get("id") == duplicate_id), + "", + ) + return { + "success": False, + "error": ( + f"Potential duplicate of '{duplicate_title}' " + f"(id={duplicate_id[:8]}...) — do not re-report the same vulnerability" + ), + "duplicate_of": duplicate_id, + "duplicate_title": duplicate_title, + "confidence": dedupe.get("confidence", 0.0), + "reason": dedupe.get("reason", ""), + } + + report_id = report_state.add_vulnerability_report( + **report_fields, + agent_id=agent_id if isinstance(agent_id, str) else None, + agent_name=agent_name if isinstance(agent_name, str) else None, ) - return { - "success": False, - "error": ( - f"Potential duplicate of '{duplicate_title}' " - f"(id={duplicate_id[:8]}...) — do not re-report the same vulnerability" - ), - "duplicate_of": duplicate_id, - "duplicate_title": duplicate_title, - "confidence": dedupe.get("confidence", 0.0), - "reason": dedupe.get("reason", ""), - } - - report_id = report_state.add_vulnerability_report( - **report_fields, - agent_id=agent_id if isinstance(agent_id, str) else None, - agent_name=agent_name if isinstance(agent_name, str) else None, - ) except Exception as e: logger.exception("create_vulnerability_report persistence failed") return { @@ -2024,7 +2028,6 @@ async def _do_create_dependency( # noqa: PLR0912, PLR0915 from strix.report.dedupe import check_duplicate - existing = report_state.get_existing_vulnerabilities() candidate = { "title": title, "description": description, @@ -2033,41 +2036,45 @@ async def _do_create_dependency( # noqa: PLR0912, PLR0915 "dependency_metadata": dependency_metadata, "technical_analysis": technical_analysis, } - dedupe = await check_duplicate(candidate, existing) - if dedupe.get("is_duplicate"): - duplicate_id = dedupe.get("duplicate_id", "") - return { - "success": False, - "error": ( - f"Potential duplicate (id={duplicate_id[:8]}...) — " - "do not re-report the same dependency finding" - ), - "duplicate_of": duplicate_id, - "confidence": dedupe.get("confidence", 0.0), - "reason": dedupe.get("reason", ""), - } - - report_id = report_state.add_vulnerability_report( - title=title, - description=description, - severity=severity, - impact=impact, - target=target, - technical_analysis=technical_analysis, - remediation_steps=remediation_steps, - evidence=evidence, - assumptions=assumptions, - fix_effort=fix_effort, - cvss=cvss_score if advisory_cvss is not None else None, - cve=parsed_cve, - cwe=cwe, - code_locations=parsed_locations, - fix_verification=fix_verification, - finding_class="dependency_cve", - dependency_metadata=dependency_metadata, - agent_id=agent_id if isinstance(agent_id, str) else None, - agent_name=agent_name if isinstance(agent_name, str) else None, - ) + # A same-CVE report without package metadata can send dependency + # deduplication through the model too, so it shares the filing lock. + async with report_state.dedupe_lock: + existing = report_state.get_existing_vulnerabilities() + dedupe = await check_duplicate(candidate, existing) + if dedupe.get("is_duplicate"): + duplicate_id = dedupe.get("duplicate_id", "") + return { + "success": False, + "error": ( + f"Potential duplicate (id={duplicate_id[:8]}...) — " + "do not re-report the same dependency finding" + ), + "duplicate_of": duplicate_id, + "confidence": dedupe.get("confidence", 0.0), + "reason": dedupe.get("reason", ""), + } + + report_id = report_state.add_vulnerability_report( + title=title, + description=description, + severity=severity, + impact=impact, + target=target, + technical_analysis=technical_analysis, + remediation_steps=remediation_steps, + evidence=evidence, + assumptions=assumptions, + fix_effort=fix_effort, + cvss=cvss_score if advisory_cvss is not None else None, + cve=parsed_cve, + cwe=cwe, + code_locations=parsed_locations, + fix_verification=fix_verification, + finding_class="dependency_cve", + dependency_metadata=dependency_metadata, + agent_id=agent_id if isinstance(agent_id, str) else None, + agent_name=agent_name if isinstance(agent_name, str) else None, + ) except Exception as e: logger.exception("create_dependency_report persistence failed") return { diff --git a/tests/test_reporting_dedupe_race.py b/tests/test_reporting_dedupe_race.py new file mode 100644 index 0000000000..69dd3e7109 --- /dev/null +++ b/tests/test_reporting_dedupe_race.py @@ -0,0 +1,237 @@ +"""Findings filed at the same moment must go through the duplicate check one at a time. + +The duplicate check awaits a model call and compares the candidate with the findings +stored when it started. Two agents filing the same finding concurrently used to both +pass it, because neither saw the other's report yet, and the run stored it twice. + +These tests drive the real ``check_duplicate``. It only reaches the model when something +is already stored (an empty store returns at once, with nothing to await), so every test +seeds an unrelated finding first, and the model is a stand-in that pauses like a real +call and judges the candidate against the reports in the request it is handed. +""" + +from __future__ import annotations + +import asyncio +import json +from typing import TYPE_CHECKING, Any + +import pytest +from agents.items import ModelResponse +from agents.usage import Usage +from openai.types.responses import ResponseOutputMessage, ResponseOutputText + +from strix.config import loader +from strix.report import dedupe +from strix.report.state import ReportState, set_global_report_state +from strix.tools.reporting.tool import _do_create, _do_create_dependency + + +if TYPE_CHECKING: + from collections.abc import AsyncIterator + from pathlib import Path + + +_CVSS = { + "attack_vector": "N", + "attack_complexity": "L", + "privileges_required": "N", + "user_interaction": "N", + "scope": "U", + "confidentiality": "H", + "integrity": "H", + "availability": "H", +} + +_PROMPT_HEAD = "Compare this candidate vulnerability against existing reports:\n\n" +_PROMPT_TAIL = "\n\nRespond with ONLY the JSON object" + + +class _FakeDedupeModel: + """A dedupe model that pauses like a real call and judges by title.""" + + def __init__(self) -> None: + self.calls = 0 + + async def get_response(self, *, input: str, **_kwargs: Any) -> ModelResponse: # noqa: A002 + self.calls += 1 + await asyncio.sleep(0.05) + payload = json.loads(input.split(_PROMPT_HEAD, 1)[1].split(_PROMPT_TAIL, 1)[0]) + verdict = {"is_duplicate": False, "duplicate_id": "", "confidence": 0.9, "reason": "new"} + for report in payload["existing_reports"]: + if report.get("title") == payload["candidate"].get("title"): + verdict = { + "is_duplicate": True, + "duplicate_id": report["id"], + "confidence": 0.99, + "reason": "same title", + } + message = ResponseOutputMessage( + id="msg", + role="assistant", + status="completed", + type="message", + content=[ + ResponseOutputText(type="output_text", text=json.dumps(verdict), annotations=[]) + ], + ) + return ModelResponse(output=[message], usage=Usage(), response_id=None) + + +def _vulnerability(title: str = "SQL injection in /login") -> dict[str, Any]: + return { + "title": title, + "description": "d", + "impact": "i", + "target": "http://target.example", + "technical_analysis": "ta", + "poc_description": "pd", + "poc_script_code": "curl http://target.example/login", + "remediation_steps": "rs", + "evidence": "ev", + "assumptions": "as", + "counterevidence": "none found", + "confidence": "high", + "severity_change_conditions": "none", + "fix_effort": "low", + "cvss_breakdown": _CVSS, + "endpoint": "/login", + "method": "POST", + "cve": None, + "cwe": None, + "code_locations": None, + } + + +def _dependency() -> dict[str, Any]: + return { + "title": "Vulnerable example-package dependency", + "description": "The package is pinned to an affected version.", + "target": "package-lock.json", + "cve": "CVE-2024-0001", + "package_name": "example-package", + "installed_version": "1.0.0", + "impact": "The dependency exposes the vulnerable code path.", + "remediation_steps": "Upgrade to the fixed version.", + "assumptions": "The lockfile describes the deployed application.", + "package_ecosystem": "npm", + "manifest_path": "package-lock.json", + "fixed_version": "1.0.1", + "cwe": None, + "advisory_cvss": 7.5, + "technical_analysis": "The package is present in the lockfile.", + "fix_effort": "low", + "reachability_evidence": "The package is installed; call sites have not been traced.", + "contextual_cvss_breakdown": _CVSS, + "contextual_cvss_reasoning": "The fixture retains the advisory impact assumptions.", + } + + +@pytest.fixture +def model(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> _FakeDedupeModel: + fake = _FakeDedupeModel() + # Settings are memoized and fall back to the developer's stored config, so point + # them at an empty file and a model name instead of whatever ran before. + monkeypatch.setattr(loader, "_override", tmp_path / "no-config.json") + monkeypatch.setattr(loader, "_cached", None) + monkeypatch.delenv("STRIX_DEDUPE_MODEL", raising=False) + monkeypatch.setenv("STRIX_LLM", "openai/fake-dedupe") + monkeypatch.setattr("strix.report.dedupe.configure_sdk_model_defaults", lambda _settings: None) + monkeypatch.setattr("strix.report.dedupe.resolve_dedupe_model", lambda *_args: fake) + return fake + + +@pytest.fixture +async def state( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, model: _FakeDedupeModel +) -> AsyncIterator[ReportState]: + monkeypatch.chdir(tmp_path) + report_state = ReportState(run_name="dedupe-race") + set_global_report_state(report_state) + # With nothing stored the real check returns before it awaits anything, so store one + # unrelated finding: from here on every check goes to the model. + seeded = await _do_create(**_vulnerability("Open redirect in /logout")) + assert seeded["success"] is True + assert model.calls == 0 + yield report_state + set_global_report_state(None) + + +async def test_same_finding_filed_concurrently_is_stored_once( + state: ReportState, model: _FakeDedupeModel +) -> None: + first, second = await asyncio.gather( + _do_create(**_vulnerability()), _do_create(**_vulnerability()) + ) + + assert sorted([first["success"], second["success"]]) == [False, True] + rejected = first if not first["success"] else second + stored = [r for r in state.vulnerability_reports if r["title"] == "SQL injection in /login"] + assert rejected["duplicate_of"] == stored[0]["id"] + assert len(stored) == 1 + assert len(state.vulnerability_reports) == 2 + assert model.calls == 2 # both went through the model, one after the other + + +async def test_distinct_findings_filed_concurrently_are_all_stored(state: ReportState) -> None: + results = await asyncio.gather( + _do_create(**_vulnerability("SQL injection in /login")), + _do_create(**_vulnerability("Reflected XSS in /search")), + _do_create(**_vulnerability("IDOR in /orders")), + ) + + assert all(result["success"] for result in results) + assert len(state.vulnerability_reports) == 4 + + +async def test_dependency_model_fallback_serializes_concurrent_filings( + state: ReportState, model: _FakeDedupeModel +) -> None: + # A finding with the same CVE but no package identity makes the real + # dependency check fall back to the model instead of returning synchronously. + legacy = await _do_create( + **{**_vulnerability("Existing application finding"), "cve": "CVE-2024-0001"} + ) + assert legacy["success"] is True + calls_before = model.calls + + first, second = await asyncio.gather( + _do_create_dependency(**_dependency()), _do_create_dependency(**_dependency()) + ) + + assert sorted([first["success"], second["success"]]) == [False, True] + stored = [r for r in state.vulnerability_reports if r.get("finding_class") == "dependency_cve"] + assert len(stored) == 1 + rejected = first if not first["success"] else second + assert rejected["duplicate_of"] == stored[0]["id"] + # The first filing calls the model; the next sees its package identity. + assert model.calls - calls_before == 1 + + +async def test_dependency_identity_fast_path_still_needs_no_model( + state: ReportState, model: _FakeDedupeModel +) -> None: + first, second = await asyncio.gather( + _do_create_dependency(**_dependency()), _do_create_dependency(**_dependency()) + ) + + assert sorted([first["success"], second["success"]]) == [False, True] + assert len(state.vulnerability_reports) == 2 + assert model.calls == 0 + + +async def test_a_failing_check_does_not_leave_the_lock_held( + state: ReportState, monkeypatch: pytest.MonkeyPatch +) -> None: + async def broken(*_args: object) -> dict[str, Any]: + raise RuntimeError("dedupe model unreachable") + + real_check = dedupe.check_duplicate + monkeypatch.setattr(dedupe, "check_duplicate", broken) + failed = await _do_create(**_vulnerability()) + assert failed["success"] is False + assert not state.dedupe_lock.locked() + + monkeypatch.setattr(dedupe, "check_duplicate", real_check) + retried = await asyncio.wait_for(_do_create(**_vulnerability()), timeout=5) + assert retried["success"] is True