scriptoriumadmin/strix
1
0
Fork 0
mirror of https://github.com/usestrix/strix.git synced 2026-10-11 03:37:54 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1
strix/tests/test_reporting_fields.py
Ahmed Allam c0258b25fa fix(finish_scan): stop asking for a section heading in every report field 
Every renderer of the final report already titles each section, so the
heading the docstring asked for printed twice. Describe the fields as
section bodies and drop the example headings.
2026-10-03 23:07:56 +03:00

2250 lines
81 KiB
Python
Raw Blame History

"""Tests for restored report fields, SCA tool, and report formatting guidance."""
from __future__ import annotations
import json
from typing import TYPE_CHECKING, Any, cast
import pytest
from agents.tool_context import ToolContext
from strix.report.dedupe import (
_check_dependency_duplicate,
_prepare_report_for_comparison,
check_duplicate,
)
from strix.report.state import ReportRollbackError, ReportState, set_global_report_state
from strix.report.writer import write_run_record
from strix.tools.finish.tool import finish_scan
from strix.tools.reporting import tool as reporting_tool
from strix.tools.reporting.tool import (
_do_create,
_do_create_dependency,
_do_delete,
_do_update,
_normalize_http_exchange_ids,
_verify_http_exchange_ids,
create_dependency_report,
create_vulnerability_report,
delete_vulnerability_report,
update_vulnerability_report,
)
if TYPE_CHECKING:
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",
}
_DEP_CONTEXT = {
"attack_vector": "N",
"attack_complexity": "L",
"privileges_required": "N",
"user_interaction": "N",
"scope": "U",
"confidentiality": "N",
"integrity": "N",
"availability": "H",
}
_DEP_CONTEXT_VECTOR = "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:N/I:N/A:H"
_DEP_EVIDENCE = "src/render.ts:14 imports the package."
_DEP_REASONING = "Only scripts/import.py reaches the sink, so the impact is availability only."
@pytest.fixture
def report_state(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> ReportState:
monkeypatch.chdir(tmp_path)
state = ReportState(run_name="test-run")
set_global_report_state(state)
return state
def test_record_mcp_connection_status_persists_and_dedupes(
report_state: ReportState, monkeypatch: pytest.MonkeyPatch
) -> None:
"""The roster lands on the run record so run.json carries it for the viewer,
and an unchanged re-write is a no-op (it does not re-save)."""
roster = [{"name": "local_fs", "provider": None, "tool_count": 3, "dead": False}]
report_state.record_mcp_connection_status(roster)
assert report_state.run_record["mcp_connection_status"] == roster
saves = 0
original_save = report_state.save_run_data
def _counting_save(*args: Any, **kwargs: Any) -> None:
nonlocal saves
saves += 1
original_save(*args, **kwargs)
monkeypatch.setattr(report_state, "save_run_data", _counting_save)
report_state.record_mcp_connection_status(roster)
assert saves == 0, "an identical roster must not trigger another save"
report_state.record_mcp_connection_status(
[{"name": "local_fs", "provider": None, "tool_count": 3, "dead": True}]
)
assert saves == 1
assert report_state.run_record["mcp_connection_status"][0]["dead"] is True
async def test_create_report_persists_new_fields(report_state: ReportState) -> None:
result = await _do_create(
title="Reflected XSS in search",
description="q reflects unencoded input.",
impact="Session theft.",
target="https://app.example.com",
technical_analysis="Input interpolated into HTML.",
poc_description="1. open /search?q=<payload>",
poc_script_code="GET /search?q=<script>alert(1)</script>",
remediation_steps="Context-encode output.",
evidence="Response echoes the payload verbatim.",
assumptions="Assumes a victim opens a crafted link.",
counterevidence="No output encoding or CSP observed on this response.",
confidence="HIGH",
severity_change_conditions="A strict CSP would lower the severity.",
fix_effort="LOW",
cvss_breakdown=_CVSS,
endpoint="/search",
method="GET",
cve=None,
cwe="CWE-79",
code_locations=None,
http_exchange_ids=["1042", "1042", "1088"],
fix_pr_body="## Fix\nEncode output.",
)
assert result["success"] is True
report = report_state.vulnerability_reports[0]
assert report["evidence"] == "Response echoes the payload verbatim."
assert report["assumptions"] == "Assumes a victim opens a crafted link."
assert report["fix_effort"] == "low"
assert report["fix_pr_body"] == "## Fix\nEncode output."
assert report["finding_class"] == "dynamic"
assert report["counterevidence"] == "No output encoding or CSP observed on this response."
assert report["confidence"] == "high"
assert report["severity_change_conditions"] == "A strict CSP would lower the severity."
assert report["http_exchange_ids"] == ["1042", "1088"]
def test_create_report_does_not_commit_when_callback_fails(
report_state: ReportState,
) -> None:
def fail_persistence(_report: dict[str, Any]) -> None:
raise RuntimeError("persistence failed")
report_state.vulnerability_found_callback = fail_persistence
with pytest.raises(RuntimeError, match="persistence failed"):
report_state.add_vulnerability_report(
title="Unstored finding",
severity="high",
http_exchange_ids=["1042"],
)
assert report_state.vulnerability_reports == []
def test_failed_revision_keeps_old_evidence_and_can_be_retried(
report_state: ReportState,
) -> None:
report_id = report_state.add_vulnerability_report(
title="Original finding", severity="high", http_exchange_ids=["1042"]
)
original = dict(report_state.vulnerability_reports[0])
def fail_persistence(revised: dict[str, Any]) -> None:
assert revised["http_exchange_ids"] == ["1088"]
assert report_state.vulnerability_reports[0] == original
raise RuntimeError("persistence failed")
report_state.vulnerability_updated_callback = fail_persistence
changes = {"title": "Revised finding", "http_exchange_ids": ["1088"]}
with pytest.raises(RuntimeError, match="persistence failed"):
report_state.update_vulnerability_report(report_id, changes)
assert report_state.vulnerability_reports[0] == original
report_state.vulnerability_updated_callback = None
revised = report_state.update_vulnerability_report(report_id, changes)
assert revised is not None
assert revised["http_exchange_ids"] == ["1088"]
assert len(revised["update_history"]) == 1
async def test_create_report_requires_evidence_and_assumptions(
report_state: ReportState,
) -> None:
result = await _do_create(
title="X",
description="d",
impact="i",
target="t",
technical_analysis="ta",
poc_description="p",
poc_script_code="c",
remediation_steps="r",
evidence=" ",
assumptions="",
counterevidence="none found",
confidence="high",
severity_change_conditions="n/a",
fix_effort="low",
cvss_breakdown=_CVSS,
endpoint=None,
method=None,
cve=None,
cwe=None,
code_locations=None,
)
assert result["success"] is False
joined = " ".join(result["errors"])
assert "Evidence" in joined
assert "Assumptions" in joined
assert not report_state.vulnerability_reports
async def test_create_report_rejects_invalid_fix_effort(report_state: ReportState) -> None:
result = await _do_create(
title="X",
description="d",
impact="i",
target="t",
technical_analysis="ta",
poc_description="p",
poc_script_code="c",
remediation_steps="r",
evidence="e",
assumptions="a",
counterevidence="none found",
confidence="high",
severity_change_conditions="n/a",
fix_effort="enormous",
cvss_breakdown=_CVSS,
endpoint=None,
method=None,
cve=None,
cwe=None,
code_locations=None,
)
assert result["success"] is False
assert any("fix_effort" in e for e in result["errors"])
assert not report_state.vulnerability_reports
async def _create_with(report_state: ReportState, **overrides: object) -> dict[str, Any]:
kwargs: dict[str, object] = {
"title": "X",
"description": "d",
"impact": "i",
"target": "t",
"technical_analysis": "ta",
"poc_description": "p",
"poc_script_code": "c",
"remediation_steps": "r",
"evidence": "e",
"assumptions": "a",
"counterevidence": "No guard found on this path.",
"confidence": "high",
"severity_change_conditions": "Proof of internet exposure would raise it.",
"fix_effort": "low",
"cvss_breakdown": _CVSS,
"endpoint": None,
"method": None,
"cve": None,
"cwe": None,
"code_locations": None,
}
kwargs.update(overrides)
assert report_state is not None
return await _do_create(**kwargs) # type: ignore[arg-type]
async def test_create_report_requires_counterevidence(report_state: ReportState) -> None:
result = await _create_with(report_state, counterevidence=" ")
assert result["success"] is False
assert any("Counterevidence" in e for e in result["errors"])
assert not report_state.vulnerability_reports
async def test_create_report_requires_severity_change_conditions(
report_state: ReportState,
) -> None:
result = await _create_with(report_state, severity_change_conditions="")
assert result["success"] is False
assert any("severity_change_conditions" in e for e in result["errors"])
assert not report_state.vulnerability_reports
async def test_create_report_rejects_invalid_confidence(report_state: ReportState) -> None:
result = await _create_with(report_state, confidence="pretty sure")
assert result["success"] is False
assert any("confidence" in e for e in result["errors"])
assert not report_state.vulnerability_reports
async def test_create_report_requires_rationale_when_confidence_not_high(
report_state: ReportState,
) -> None:
result = await _create_with(report_state, confidence="medium")
assert result["success"] is False
assert any("confidence_rationale" in e for e in result["errors"])
assert not report_state.vulnerability_reports
async def test_create_report_accepts_medium_confidence_with_rationale(
report_state: ReportState,
) -> None:
result = await _create_with(
report_state,
confidence="medium",
confidence_rationale="Static-only trace; could not stand up the service.",
)
assert result["success"] is True
report = report_state.vulnerability_reports[0]
assert report["confidence"] == "medium"
assert report["confidence_rationale"] == "Static-only trace; could not stand up the service."
async def test_dependency_report_sets_class_and_metadata(report_state: ReportState) -> None:
result = await _do_create_dependency(
title="CVE-2021-23337 in lodash 4.17.20",
description="Command injection via template.",
target="repo/package.json",
cve="CVE-2021-23337",
package_name="lodash",
installed_version="4.17.20",
impact="Arbitrary command execution.",
remediation_steps="Upgrade to 4.17.21.",
assumptions="Assumes the template sink is reachable.",
package_ecosystem="npm",
manifest_path="package-lock.json",
fixed_version="4.17.21",
cwe="CWE-94",
advisory_cvss=7.2,
technical_analysis=None,
fix_effort="trivial",
reachability="imported",
reachability_evidence=_DEP_EVIDENCE,
contextual_cvss_breakdown=_DEP_CONTEXT,
contextual_cvss_reasoning=_DEP_REASONING,
)
assert result["success"] is True
report = report_state.vulnerability_reports[0]
assert report["finding_class"] == "dependency_cve"
assert report["cve"] == "CVE-2021-23337"
assert report["severity"] == "high"
assert report["evidence"].startswith(
"**Advisory evidence:** `CVE-2021-23337` applies to `lodash` "
"at installed version `4.17.20`. The advisory is fixed in `4.17.21`."
)
assert report["dependency_metadata"] == {
"package_name": "lodash",
"installed_version": "4.17.20",
"advisory_cvss": 7.2,
"package_ecosystem": "npm",
"manifest_path": "package-lock.json",
"fixed_version": "4.17.21",
"reachability": "imported",
"reachability_evidence": _DEP_EVIDENCE,
"contextual_cvss_breakdown": _DEP_CONTEXT,
"contextual_cvss_score": pytest.approx(7.5, abs=0.05),
"contextual_cvss_vector": _DEP_CONTEXT_VECTOR,
"contextual_cvss_reasoning": _DEP_REASONING,
}
async def test_dependency_report_records_transitive_chain(report_state: ReportState) -> None:
result = await _do_create_dependency(
title="CVE-2022-24999 in qs 6.10.2",
description="Prototype pollution in qs parsing.",
target="repo/package.json",
cve="CVE-2022-24999",
package_name="qs",
installed_version="6.10.2",
impact="Denial of service via crafted query strings.",
remediation_steps="Upgrade express to 4.18.2, which resolves qs 6.11.0.",
assumptions="qs parses all incoming query strings by default.",
package_ecosystem="npm",
manifest_path="package-lock.json",
fixed_version="6.10.3",
cwe="CWE-1321",
advisory_cvss=7.5,
technical_analysis=None,
fix_effort="trivial",
introduced_by="express@4.18.1",
dependency_path="express@4.18.1 > body-parser@1.20.0 > qs@6.10.2",
reachability="imported",
reachability_evidence=_DEP_EVIDENCE,
contextual_cvss_breakdown=_DEP_CONTEXT,
contextual_cvss_reasoning=_DEP_REASONING,
)
assert result["success"] is True
report = report_state.vulnerability_reports[0]
assert report["dependency_metadata"]["introduced_by"] == "express@4.18.1"
assert (
report["dependency_metadata"]["dependency_path"]
== "express@4.18.1 > body-parser@1.20.0 > qs@6.10.2"
)
assert (
"**Transitive dependency:** introduced by the direct dependency `express@4.18.1`."
in report["evidence"]
)
assert (
"**Dependency chain:** `express@4.18.1 > body-parser@1.20.0 > qs@6.10.2`"
in report["evidence"]
)
async def test_dependency_report_omits_blank_chain_fields(report_state: ReportState) -> None:
result = await _do_create_dependency(
title="CVE-2024-0001 in sample 1.0.0",
description="Published advisory affects the pinned version.",
target="repo/package.json",
cve="CVE-2024-0001",
package_name="sample",
installed_version="1.0.0",
impact="Impact.",
remediation_steps="Upgrade.",
assumptions="Assumptions.",
package_ecosystem="npm",
manifest_path="package-lock.json",
fixed_version=None,
cwe=None,
advisory_cvss=5.0,
technical_analysis=None,
fix_effort="trivial",
introduced_by=" ",
dependency_path=None,
reachability="imported",
reachability_evidence=_DEP_EVIDENCE,
contextual_cvss_breakdown=_DEP_CONTEXT,
contextual_cvss_reasoning=_DEP_REASONING,
)
assert result["success"] is True
report = report_state.vulnerability_reports[0]
assert "introduced_by" not in report["dependency_metadata"]
assert "dependency_path" not in report["dependency_metadata"]
async def test_dependency_report_with_no_contextual_impact_is_info(
report_state: ReportState,
) -> None:
result = await _do_create_dependency(
title="CVE-2024-0001 in sample 1.0.0",
description="Published advisory affects the pinned version.",
target="repo/package.json",
cve="CVE-2024-0001",
package_name="sample",
installed_version="1.0.0",
impact="Low-impact dependency advisory.",
remediation_steps="Upgrade to 1.0.1.",
assumptions="Assumes the package is included in deployed builds.",
package_ecosystem="npm",
manifest_path="package-lock.json",
fixed_version="1.0.1",
cwe=None,
advisory_cvss=0.0,
technical_analysis=None,
fix_effort="low",
reachability="not_imported",
reachability_evidence="No file imports the package.",
contextual_cvss_breakdown={**_DEP_CONTEXT, "availability": "N"},
contextual_cvss_reasoning="No application code imports the package.",
)
assert result["success"] is True
assert result["severity"] == "info"
report = report_state.vulnerability_reports[0]
assert report["severity"] == "info"
assert report["cvss"] == 0.0
async def test_dependency_report_records_reachability(report_state: ReportState) -> None:
result = await _do_create_dependency(
title="CVE-2021-23337 in lodash 4.17.20",
description="Command injection via template.",
target="repo/package.json",
cve="CVE-2021-23337",
package_name="lodash",
installed_version="4.17.20",
impact="Command injection where template is used.",
remediation_steps="Upgrade to 4.17.21.",
assumptions="Assumes the template sink is reachable.",
package_ecosystem="npm",
manifest_path="package-lock.json",
fixed_version="4.17.21",
cwe=None,
advisory_cvss=7.2,
technical_analysis=None,
fix_effort="low",
reachability="vulnerable_symbol_used",
reachability_evidence="src/render.ts:14 calls `_.template()`.",
contextual_cvss_breakdown=_DEP_CONTEXT,
contextual_cvss_reasoning=_DEP_REASONING,
)
assert result["success"] is True
report = report_state.vulnerability_reports[0]
assert report["dependency_metadata"]["reachability"] == "vulnerable_symbol_used"
assert (
report["dependency_metadata"]["reachability_evidence"]
== "src/render.ts:14 calls `_.template()`."
)
assert "**Usage analysis:**" in report["evidence"]
assert "not a proof of exploitability or of safety" in report["evidence"]
# The level must never influence the rating — that comes from the contextual
# breakdown, or from advisory_cvss when no breakdown applies.
assert report["severity"] == "high"
async def test_dependency_report_rejects_reachability_without_evidence(
report_state: ReportState,
) -> None:
result = await _do_create_dependency(
title="CVE-2024-0001 in sample 1.0.0",
description="Published advisory affects the pinned version.",
target="repo/package.json",
cve="CVE-2024-0001",
package_name="sample",
installed_version="1.0.0",
impact="Impact.",
remediation_steps="Upgrade.",
assumptions="Assumptions.",
package_ecosystem="npm",
manifest_path="package-lock.json",
fixed_version="1.0.1",
cwe=None,
advisory_cvss=5.0,
technical_analysis=None,
fix_effort="low",
reachability="not_imported",
)
assert result["success"] is False
assert any("reachability_evidence is required" in e for e in result["errors"])
assert not report_state.vulnerability_reports
async def test_dependency_report_rejects_unknown_reachability_level(
report_state: ReportState,
) -> None:
result = await _do_create_dependency(
title="CVE-2024-0001 in sample 1.0.0",
description="Published advisory affects the pinned version.",
target="repo/package.json",
cve="CVE-2024-0001",
package_name="sample",
installed_version="1.0.0",
impact="Impact.",
remediation_steps="Upgrade.",
assumptions="Assumptions.",
package_ecosystem="npm",
manifest_path="package-lock.json",
fixed_version="1.0.1",
cwe=None,
advisory_cvss=5.0,
technical_analysis=None,
fix_effort="low",
reachability="not_exploitable",
reachability_evidence="vibes",
)
assert result["success"] is False
assert any("Invalid reachability" in e for e in result["errors"])
assert not report_state.vulnerability_reports
async def test_dependency_report_records_unknown_reachability(report_state: ReportState) -> None:
result = await _do_create_dependency(
title="CVE-2024-0001 in sample 1.0.0",
description="Published advisory affects the pinned version.",
target="repo/package.json",
cve="CVE-2024-0001",
package_name="sample",
installed_version="1.0.0",
impact="Impact.",
remediation_steps="Upgrade.",
assumptions="Analysis was inconclusive.",
package_ecosystem="npm",
manifest_path="package-lock.json",
fixed_version="1.0.1",
cwe=None,
advisory_cvss=5.0,
technical_analysis=None,
fix_effort="low",
reachability_evidence="Grep for the package found no import.",
contextual_cvss_breakdown=_DEP_CONTEXT,
contextual_cvss_reasoning=_DEP_REASONING,
)
assert result["success"] is True, result
metadata = report_state.vulnerability_reports[0]["dependency_metadata"]
assert metadata["reachability"] == "unknown"
assert metadata["reachability_evidence"] == "Grep for the package found no import."
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",
description="Published advisory affects the pinned version.",
target="repo/package.json",
cve="CVE-2024-0001",
package_name="sample",
installed_version="1.0.0",
impact="Some impact.",
remediation_steps="Upgrade to 1.0.1.",
assumptions="Assumes the package ships in deployed builds.",
package_ecosystem="npm",
manifest_path="package-lock.json",
fixed_version="1.0.1",
cwe=None,
advisory_cvss=None,
technical_analysis=None,
fix_effort="low",
)
assert result["success"] is False
assert any("advisory_cvss is required" in e for e in result["errors"])
assert not report_state.vulnerability_reports
async def test_dependency_report_dedupe_candidate_includes_dependency_metadata(
report_state: ReportState,
monkeypatch: pytest.MonkeyPatch,
) -> None:
captured: dict[str, object] = {}
async def fake_check_duplicate(
candidate: dict[str, object],
existing: list[dict[str, object]],
) -> dict[str, object]:
captured["candidate"] = candidate
captured["existing"] = existing
return {"is_duplicate": False}
monkeypatch.setattr("strix.report.dedupe.check_duplicate", fake_check_duplicate)
report_state.vulnerability_reports.append(
{
"id": "vuln-0001",
"title": "CVE-2024-0001 in other 1.0.0",
"severity": "low",
"timestamp": "2026-01-01 00:00:00 UTC",
"description": "Existing dependency finding.",
"target": "repo/package.json",
"cve": "CVE-2024-0001",
"dependency_metadata": {
"package_name": "other",
"installed_version": "1.0.0",
"package_ecosystem": "npm",
},
}
)
result = await _do_create_dependency(
title="CVE-2024-0001 in sample 1.0.0",
description="Published advisory affects the pinned version.",
target="repo/package.json",
cve="CVE-2024-0001",
package_name="sample",
installed_version="1.0.0",
impact="Low-impact dependency advisory.",
remediation_steps="Upgrade to 1.0.1.",
assumptions="Assumes the package is included in deployed builds.",
package_ecosystem="npm",
manifest_path="package-lock.json",
fixed_version="1.0.1",
cwe=None,
advisory_cvss=0.0,
technical_analysis=None,
fix_effort="low",
reachability="imported",
reachability_evidence=_DEP_EVIDENCE,
contextual_cvss_breakdown=_DEP_CONTEXT,
contextual_cvss_reasoning=_DEP_REASONING,
)
assert result["success"] is True
assert captured["candidate"] == {
"title": "CVE-2024-0001 in sample 1.0.0",
"description": "Published advisory affects the pinned version.",
"target": "repo/package.json",
"cve": "CVE-2024-0001",
"dependency_metadata": {
"package_name": "sample",
"installed_version": "1.0.0",
"advisory_cvss": 0.0,
"package_ecosystem": "npm",
"manifest_path": "package-lock.json",
"fixed_version": "1.0.1",
"reachability": "imported",
"reachability_evidence": _DEP_EVIDENCE,
"contextual_cvss_breakdown": _DEP_CONTEXT,
"contextual_cvss_score": pytest.approx(7.5, abs=0.05),
"contextual_cvss_vector": _DEP_CONTEXT_VECTOR,
"contextual_cvss_reasoning": _DEP_REASONING,
},
"technical_analysis": None,
}
async def test_dependency_report_rejects_bad_cve(report_state: ReportState) -> None:
result = await _do_create_dependency(
title="bad",
description="d",
target="t",
cve="not-a-cve",
package_name="pkg",
installed_version="1.0.0",
impact="i",
remediation_steps="r",
assumptions="a",
package_ecosystem="npm",
manifest_path="package-lock.json",
fixed_version=None,
cwe=None,
advisory_cvss=None,
technical_analysis=None,
fix_effort="low",
)
assert result["success"] is False
assert not report_state.vulnerability_reports
async def test_dependency_report_requires_ecosystem(report_state: ReportState) -> None:
result = await _do_create_dependency(
title="CVE-2024-0001 in sample 1.0.0",
description="Published advisory affects the pinned version.",
target="repo/package.json",
cve="CVE-2024-0001",
package_name="sample",
installed_version="1.0.0",
impact="Low-impact dependency advisory.",
remediation_steps="Upgrade to 1.0.1.",
assumptions="Assumes the package is included in deployed builds.",
package_ecosystem="",
manifest_path="package-lock.json",
fixed_version="1.0.1",
cwe=None,
advisory_cvss=0.0,
technical_analysis=None,
fix_effort="low",
)
assert result["success"] is False
assert any("package_ecosystem" in error for error in result["errors"])
assert not report_state.vulnerability_reports
async def test_dependency_report_requires_manifest_path(report_state: ReportState) -> None:
result = await _do_create_dependency(
title="CVE-2024-0001 in sample 1.0.0",
description="Published advisory affects the pinned version.",
target="repo/package.json",
cve="CVE-2024-0001",
package_name="sample",
installed_version="1.0.0",
impact="Low-impact dependency advisory.",
remediation_steps="Upgrade to 1.0.1.",
assumptions="Assumes the package is included in deployed builds.",
package_ecosystem="npm",
manifest_path=None,
fixed_version="1.0.1",
cwe=None,
advisory_cvss=5.0,
technical_analysis=None,
fix_effort="low",
)
assert result["success"] is False
assert any("manifest_path is required" in error for error in result["errors"])
assert not report_state.vulnerability_reports
@pytest.mark.parametrize(
"bad_path",
["/etc/passwd", "..\\pom.xml", "services/../pom.xml", "./package.json", "C:/repo/pom.xml"],
)
async def test_dependency_report_rejects_unsafe_manifest_path(
report_state: ReportState, bad_path: str
) -> None:
result = await _do_create_dependency(
title="CVE-2024-0001 in sample 1.0.0",
description="Published advisory affects the pinned version.",
target="repo/package.json",
cve="CVE-2024-0001",
package_name="sample",
installed_version="1.0.0",
impact="Low-impact dependency advisory.",
remediation_steps="Upgrade to 1.0.1.",
assumptions="Assumes the package is included in deployed builds.",
package_ecosystem="npm",
manifest_path=bad_path,
fixed_version="1.0.1",
cwe=None,
advisory_cvss=5.0,
technical_analysis=None,
fix_effort="low",
)
assert result["success"] is False
assert any("manifest_path" in error for error in result["errors"])
assert not report_state.vulnerability_reports
def test_dedupe_comparison_preserves_cve_identity() -> None:
cleaned = _prepare_report_for_comparison(
{
"title": "CVE-2021-23337 in lodash",
"description": "Pinned vulnerable dependency.",
"target": "repo/package.json",
"cve": "CVE-2021-23337",
"dependency_metadata": {"package_name": "lodash"},
}
)
assert cleaned["cve"] == "CVE-2021-23337"
assert cleaned["dependency_metadata"] == {"package_name": "lodash"}
async def test_dependency_dedupe_uses_cve_package_identity() -> None:
existing = [
{
"id": "vuln-0001",
"title": "CVE-2024-0001 in other",
"cve": "CVE-2024-0001",
"dependency_metadata": {
"package_name": "other",
"installed_version": "1.0.0",
"package_ecosystem": "npm",
},
}
]
candidate = {
"title": "CVE-2024-0001 in sample",
"description": "Similar advisory prose.",
"target": "repo/package.json",
"cve": "CVE-2024-0001",
"dependency_metadata": {
"package_name": "sample",
"installed_version": "1.0.0",
"package_ecosystem": "npm",
},
}
result = await check_duplicate(candidate, existing)
assert result["is_duplicate"] is False
assert result["confidence"] == 1.0
async def test_dependency_dedupe_rejects_same_cve_package_identity() -> None:
existing = [
{
"id": "vuln-0001",
"title": "CVE-2024-0001 in sample",
"cve": "CVE-2024-0001",
"dependency_metadata": {
"package_name": "sample",
"installed_version": "1.0.0",
"package_ecosystem": "npm",
},
}
]
candidate = {
"title": "CVE-2024-0001 in sample with different prose",
"description": "Different prose for the same dependency identity.",
"target": "repo/package.json",
"cve": "CVE-2024-0001",
"dependency_metadata": {
"package_name": "sample",
"installed_version": "1.0.1",
"package_ecosystem": "npm",
},
}
result = await check_duplicate(candidate, existing)
assert result["is_duplicate"] is True
assert result["duplicate_id"] == "vuln-0001"
assert result["confidence"] == 1.0
async def test_dependency_dedupe_keeps_findings_from_distinct_manifests() -> None:
existing = [
{
"id": "vuln-0001",
"title": "CVE-2024-0001 in sample",
"cve": "CVE-2024-0001",
"dependency_metadata": {
"package_name": "sample",
"installed_version": "1.0.0",
"package_ecosystem": "npm",
"manifest_path": "services/api/package-lock.json",
},
}
]
candidate = {
"title": "CVE-2024-0001 in sample (web)",
"description": "Same advisory observed in a second workspace.",
"target": "repo/package.json",
"cve": "CVE-2024-0001",
"dependency_metadata": {
"package_name": "sample",
"installed_version": "1.0.0",
"package_ecosystem": "npm",
"manifest_path": "services/web/package-lock.json",
},
}
result = await check_duplicate(candidate, existing)
assert result["is_duplicate"] is False
assert result["confidence"] == 1.0
async def test_dependency_dedupe_rejects_same_manifest_identity() -> None:
existing = [
{
"id": "vuln-0001",
"title": "CVE-2024-0001 in sample",
"cve": "CVE-2024-0001",
"dependency_metadata": {
"package_name": "sample",
"installed_version": "1.0.0",
"package_ecosystem": "npm",
"manifest_path": "services/api/package-lock.json",
},
}
]
candidate = {
"title": "CVE-2024-0001 in sample re-reported",
"description": "Same advisory, same manifest.",
"target": "repo/package.json",
"cve": "CVE-2024-0001",
"dependency_metadata": {
"package_name": "sample",
"installed_version": "1.0.0",
"package_ecosystem": "npm",
"manifest_path": "services/api/package-lock.json",
},
}
result = await check_duplicate(candidate, existing)
assert result["is_duplicate"] is True
assert result["duplicate_id"] == "vuln-0001"
async def test_dependency_dedupe_detects_legacy_same_cve_package() -> None:
existing = [
{
"id": "vuln-0001",
"title": "CVE-2024-0001 in npm sample package",
"description": "Legacy dependency finding without structured metadata.",
"cve": "CVE-2024-0001",
}
]
candidate = {
"title": "CVE-2024-0001 in sample",
"description": "Different prose for the same dependency identity.",
"target": "repo/package.json",
"cve": "CVE-2024-0001",
"dependency_metadata": {
"package_name": "sample",
"installed_version": "1.0.1",
"package_ecosystem": "npm",
},
}
result = await check_duplicate(candidate, existing)
assert result["is_duplicate"] is True
assert result["duplicate_id"] == "vuln-0001"
assert result["confidence"] == 1.0
def test_dependency_dedupe_defers_unclear_legacy_same_cve() -> None:
existing = [
{
"id": "vuln-0001",
"title": "CVE-2024-0001 dependency finding",
"description": "Legacy dependency finding without package identity.",
"cve": "CVE-2024-0001",
}
]
candidate = {
"title": "CVE-2024-0001 in sample",
"description": "Candidate dependency finding.",
"target": "repo/package.json",
"cve": "CVE-2024-0001",
"dependency_metadata": {
"package_name": "sample",
"installed_version": "1.0.1",
"package_ecosystem": "npm",
},
}
assert _check_dependency_duplicate(candidate, existing) is None
def test_dependency_dedupe_defers_legacy_package_substring_match() -> None:
existing = [
{
"id": "vuln-0001",
"title": "CVE-2024-0001 in sample-package",
"description": "Legacy dependency finding for a different package.",
"cve": "CVE-2024-0001",
}
]
candidate = {
"title": "CVE-2024-0001 in sample",
"description": "Candidate dependency finding.",
"target": "repo/package.json",
"cve": "CVE-2024-0001",
"dependency_metadata": {
"package_name": "sample",
"installed_version": "1.0.1",
"package_ecosystem": "npm",
},
}
assert _check_dependency_duplicate(candidate, existing) is None
def test_dependency_dedupe_defers_legacy_ecosystem_mismatch() -> None:
existing = [
{
"id": "vuln-0001",
"title": "CVE-2024-0001 in npm sample",
"description": "Legacy dependency finding for a different ecosystem.",
"cve": "CVE-2024-0001",
}
]
candidate = {
"title": "CVE-2024-0001 in sample",
"description": "Candidate dependency finding.",
"target": "repo/requirements.txt",
"cve": "CVE-2024-0001",
"dependency_metadata": {
"package_name": "sample",
"installed_version": "1.0.1",
"package_ecosystem": "pypi",
},
}
assert _check_dependency_duplicate(candidate, existing) is None
def test_dependency_dedupe_matches_structured_missing_ecosystem() -> None:
existing = [
{
"id": "vuln-0001",
"title": "CVE-2024-0001 in sample",
"cve": "CVE-2024-0001",
"dependency_metadata": {
"package_name": "sample",
"installed_version": "1.0.0",
},
}
]
candidate = {
"title": "CVE-2024-0001 in sample",
"description": "Candidate dependency finding.",
"target": "repo/package.json",
"cve": "CVE-2024-0001",
"dependency_metadata": {
"package_name": "sample",
"installed_version": "1.0.1",
"package_ecosystem": "npm",
},
}
result = _check_dependency_duplicate(candidate, existing)
assert result is not None
assert result["is_duplicate"] is True
assert result["duplicate_id"] == "vuln-0001"
def test_tool_descriptions_include_formatting_guidance() -> None:
vuln_desc = create_vulnerability_report.description
assert "markdown" in vuln_desc.lower()
assert "fenced code" in vuln_desc.lower()
finish_desc = finish_scan.description
assert "markdown" in finish_desc.lower()
assert "section body" in finish_desc.lower()
assert "do not start a field with a heading" in finish_desc.lower()
assert "\n # Executive Summary" not in finish_desc
dep_desc = create_dependency_report.description
assert "cve" in dep_desc.lower()
assert "reachab" in dep_desc.lower()
def test_git_blame_hint_is_a_trailing_note() -> None:
desc = create_vulnerability_report.description
assert desc.count("blame") == 1
tail = desc[desc.index("Nice to have:") :]
assert "git blame" in tail
assert "technical_analysis" in tail
assert "Example" not in tail
assert "blame" not in update_vulnerability_report.description
def test_vuln_tool_exposes_new_params() -> None:
props = create_vulnerability_report.params_json_schema["properties"]
for field in (
"evidence",
"assumptions",
"fix_effort",
"fix_pr_body",
"http_exchange_ids",
):
assert field in props
dep_props = create_dependency_report.params_json_schema["properties"]
for field in ("package_name", "installed_version", "cve", "advisory_cvss"):
assert field in dep_props
for field in ("reachability", "reachability_evidence", "manifest_path"):
assert field in dep_props
dep_required = create_dependency_report.params_json_schema["required"]
assert "package_ecosystem" in dep_required
assert "advisory_cvss" in dep_required
_FIX_LOCATION = {
"file": "app/views.py",
"start_line": 10,
"end_line": 12,
"fix_before": 'query = f"SELECT * FROM t WHERE id={uid}"',
"fix_after": 'query = "SELECT * FROM t WHERE id=%s"',
}
_INFO_LOCATION = {
"file": "app/views.py",
"start_line": 10,
"end_line": 12,
"snippet": 'query = f"SELECT * FROM t WHERE id={uid}"',
}
async def test_fix_after_requires_verification(report_state: ReportState) -> None:
result = await _create_with(report_state, code_locations=[_FIX_LOCATION])
assert result["success"] is False
assert any("fix_verification" in e for e in result["errors"])
assert not report_state.vulnerability_reports
async def test_fix_after_with_verification_persists(report_state: ReportState) -> None:
verification = (
"Re-ran the PoC against the patched handler: the payload is now bound as a "
"parameter and returns no extra rows. Checked the two sibling call sites of "
"the same helper and the admin export path; both already parameterized. "
"Legitimate numeric ids still resolve and the 404 path is unchanged. "
"Ran the focused view tests and ruff."
)
result = await _create_with(
report_state,
code_locations=[_FIX_LOCATION],
fix_verification=verification,
)
assert result["success"] is True
assert report_state.vulnerability_reports[0]["fix_verification"] == verification
async def test_informational_location_needs_no_verification(report_state: ReportState) -> None:
result = await _create_with(report_state, code_locations=[_INFO_LOCATION])
assert result["success"] is True
assert "fix_verification" not in report_state.vulnerability_reports[0]
def test_vuln_tool_exposes_fix_verification() -> None:
assert "fix_verification" in create_vulnerability_report.params_json_schema["properties"]
def test_dep_tool_exposes_contextual_cvss_params() -> None:
dep_props = create_dependency_report.params_json_schema["properties"]
for field in (
"contextual_cvss_breakdown",
"contextual_cvss_reasoning",
):
assert field in dep_props
assert "source-to-sink" in dep_props["contextual_cvss_breakdown"]["description"].lower()
assert "source-to-sink" in dep_props["reachability_evidence"]["description"].lower()
assert "file:line" in dep_props["contextual_cvss_reasoning"]["description"].lower()
_CONTEXTUAL_BREAKDOWN = {
"attack_vector": "L",
"attack_complexity": "H",
"privileges_required": "H",
"user_interaction": "N",
"scope": "U",
"confidentiality": "L",
"integrity": "L",
"availability": "N",
}
@pytest.mark.asyncio
async def test_dependency_report_computes_contextual_cvss(
report_state: ReportState,
) -> None:
result = await _do_create_dependency(
title="CVE-2021-23337 in lodash 4.17.20",
description="Command injection via template.",
target="repo/package.json",
cve="CVE-2021-23337",
package_name="lodash",
installed_version="4.17.20",
impact="Arbitrary command execution.",
remediation_steps="Upgrade to 4.17.21.",
assumptions="Assumes the template sink is reachable.",
package_ecosystem="npm",
advisory_cvss=7.2,
technical_analysis=None,
fixed_version="4.17.21",
cwe="CWE-94",
fix_effort="trivial",
manifest_path="package-lock.json",
reachability="vulnerable_symbol_used",
reachability_evidence="scripts/import.py:88 calls `_.template()`.",
contextual_cvss_breakdown=_CONTEXTUAL_BREAKDOWN,
contextual_cvss_reasoning="Only scripts/import.py reaches the sink.",
)
assert result["success"] is True, result
report = report_state.vulnerability_reports[0]
metadata = report["dependency_metadata"]
assert metadata["advisory_cvss"] == 7.2
assert metadata["contextual_cvss_breakdown"] == _CONTEXTUAL_BREAKDOWN
assert metadata["contextual_cvss_vector"] == ("CVSS:3.1/AV:L/AC:H/PR:H/UI:N/S:U/C:L/I:L/A:N")
assert metadata["contextual_cvss_score"] == pytest.approx(3.0, abs=0.05)
assert metadata["contextual_cvss_reasoning"] == "Only scripts/import.py reaches the sink."
# The contextual rating determines the finding's score/severity, exactly
# like a normal finding's cvss_breakdown.
assert report["cvss"] == metadata["contextual_cvss_score"]
assert report["severity"] == "low"
@pytest.mark.asyncio
async def test_dependency_report_requires_contextual_breakdown(
report_state: ReportState,
) -> None:
result = await _do_create_dependency(
title="CVE-2021-23337 in lodash 4.17.20",
description="Command injection via template.",
target="repo/package.json",
cve="CVE-2021-23337",
package_name="lodash",
installed_version="4.17.20",
impact="Arbitrary command execution.",
remediation_steps="Upgrade to 4.17.21.",
assumptions="Assumes the template sink is reachable.",
package_ecosystem="npm",
advisory_cvss=7.2,
technical_analysis=None,
fixed_version="4.17.21",
cwe="CWE-94",
fix_effort="trivial",
manifest_path="package-lock.json",
reachability="imported",
reachability_evidence=_DEP_EVIDENCE,
)
assert result["success"] is False
assert any("contextual_cvss_breakdown is required" in error for error in result["errors"])
assert report_state.vulnerability_reports == []
@pytest.mark.asyncio
async def test_dependency_report_rejects_incomplete_contextual_breakdown(
report_state: ReportState,
) -> None:
result = await _do_create_dependency(
title="CVE-2021-23337 in lodash 4.17.20",
description="Command injection via template.",
target="repo/package.json",
cve="CVE-2021-23337",
package_name="lodash",
installed_version="4.17.20",
impact="Arbitrary command execution.",
remediation_steps="Upgrade to 4.17.21.",
assumptions="Assumes the template sink is reachable.",
package_ecosystem="npm",
advisory_cvss=7.2,
technical_analysis=None,
fixed_version="4.17.21",
cwe="CWE-94",
fix_effort="trivial",
manifest_path="package-lock.json",
contextual_cvss_breakdown={"attack_vector": "L", "attack_complexity": "Z"},
contextual_cvss_reasoning="Only scripts/import.py reaches the sink.",
)
assert result["success"] is False
assert any("attack_complexity" in error for error in result["errors"])
assert any("privileges_required" in error for error in result["errors"])
assert report_state.vulnerability_reports == []
@pytest.mark.asyncio
async def test_dependency_report_rejects_contextual_breakdown_without_reasoning(
report_state: ReportState,
) -> None:
result = await _do_create_dependency(
title="CVE-2021-23337 in lodash 4.17.20",
description="Command injection via template.",
target="repo/package.json",
cve="CVE-2021-23337",
package_name="lodash",
installed_version="4.17.20",
impact="Arbitrary command execution.",
remediation_steps="Upgrade to 4.17.21.",
assumptions="Assumes the template sink is reachable.",
package_ecosystem="npm",
advisory_cvss=7.2,
technical_analysis=None,
fixed_version="4.17.21",
cwe="CWE-94",
fix_effort="trivial",
manifest_path="package-lock.json",
contextual_cvss_breakdown=_CONTEXTUAL_BREAKDOWN,
contextual_cvss_reasoning=" ",
)
assert result["success"] is False
assert any("contextual_cvss_reasoning is required" in error for error in result["errors"])
assert report_state.vulnerability_reports == []
_CONFIRMED_KWARGS: dict[str, Any] = {
"title": "Unauthenticated file write on /files/{id}",
"description": "A multipart PATCH writes attacker content before the permission check.",
"impact": "Any anonymous user overwrites stored files and serves attacker content.",
"target": "https://cms.example.com",
"technical_analysis": "disk.write runs before the authorization guard.",
"poc_description": "1. PATCH /files/<uuid> with a multipart body as an anonymous user.",
"poc_script_code": "PATCH /files/2f1c HTTP/1.1\n\n--x\nowned\n--x--",
"remediation_steps": "Authorize before the write.",
"evidence": "The stored file returns the injected payload after the 403 response.",
"assumptions": "Assumes the uuid of one existing file is known.",
"counterevidence": "The endpoint answers 403, yet the write already landed.",
"confidence": "HIGH",
"confidence_rationale": "The write was observed end to end against the live host.",
"severity_change_conditions": "A guard before disk.write would remove the impact.",
"fix_effort": "MEDIUM",
"cvss_breakdown": _CVSS,
"endpoint": "/files/{id}",
"method": "PATCH",
"cve": "CVE-2025-55746",
"cwe": "CWE-863",
"code_locations": None,
}
def _seed_weak_report(report_state: ReportState) -> None:
"""A version-based, unproven entry for the same issue, as an earlier agent files it."""
report_state.vulnerability_reports.append(
{
"id": "vuln-0009",
"title": "Directus 11.5.1 exposed on public host (in scope for CVE-2025-55746)",
"severity": "medium",
"timestamp": "2026-01-01 00:00:00 UTC",
"description": "The banner reports a version affected by CVE-2025-55746.",
"target": "https://cms.example.com",
"confidence": "low",
"evidence": "The version banner only.",
"cvss": 5.3,
"finding_class": "dynamic",
"agent_id": "aaaa1111",
}
)
report_state._saved_vuln_ids.add("vuln-0009")
async def test_duplicate_verdict_rejects_without_touching_the_existing_report(
report_state: ReportState, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Deduplication only answers identity. A duplicate is rejected and points at the
finding it matched; revising that finding is a separate, explicit operation."""
_seed_weak_report(report_state)
async def fake_check_duplicate(
_candidate: dict[str, Any], _existing: list[dict[str, Any]]
) -> dict[str, Any]:
return {
"is_duplicate": True,
"duplicate_id": "vuln-0009",
"confidence": 0.9,
"reason": "Same root cause on the same endpoint.",
}
monkeypatch.setattr("strix.report.dedupe.check_duplicate", fake_check_duplicate)
result = await _do_create(**_CONFIRMED_KWARGS, agent_id="834f79fb", agent_name="Validation")
assert result["success"] is False
assert result["duplicate_of"] == "vuln-0009"
assert "action" not in result
assert len(report_state.vulnerability_reports) == 1
report = report_state.vulnerability_reports[0]
assert report["severity"] == "medium", "a duplicate verdict never edits the matched finding"
assert "poc_script_code" not in report
assert "update_history" not in report
def test_update_vulnerability_report_records_chained_impact(report_state: ReportState) -> None:
"""Attack chaining raises the impact of a finding already on file."""
_seed_weak_report(report_state)
updated = report_state.update_vulnerability_report(
"vuln-0009",
{
"severity": "CRITICAL",
"cvss": 9.8,
"impact": "The overwritten file loads in an admin session and takes over the account.",
"id": "vuln-9999",
"finding_class": "static",
},
update_reason="A chained admin takeover follows the file write.",
)
assert updated is not None
assert updated["id"] == "vuln-0009", "identity fields are not updatable"
assert updated["finding_class"] == "dynamic"
assert updated["severity"] == "critical"
assert updated["updated_at"]
assert report_state.update_vulnerability_report("vuln-0404", {"severity": "high"}) is None
def test_update_replaces_http_exchange_ids(report_state: ReportState) -> None:
_seed_weak_report(report_state)
result = _do_update(
report_id="vuln-0009",
update_reason="A replay produced a clearer proving exchange.",
fields={"http_exchange_ids": ["204", "204", "205"]},
)
assert result["success"] is True
assert report_state.vulnerability_reports[0]["http_exchange_ids"] == ["204", "205"]
async def test_create_rejects_invalid_http_exchange_ids(report_state: ReportState) -> None:
result = await _do_create(
**_CONFIRMED_KWARGS,
http_exchange_ids=["ok", "contains space"],
)
assert result["success"] is False
assert any("visible ASCII" in error for error in result["errors"])
assert report_state.vulnerability_reports == []
def test_http_exchange_id_limit_applies_after_deduplication() -> None:
request_ids, errors = _normalize_http_exchange_ids(["1042"] * 11)
assert errors == []
assert request_ids == ["1042"]
async def test_http_exchange_ids_must_exist_in_current_proxy_project(
monkeypatch: pytest.MonkeyPatch,
) -> None:
async def existing_request_ids(
_ctx: Any,
_request_ids: list[str],
) -> set[str]:
return {"1042"}
monkeypatch.setattr(reporting_tool, "existing_request_ids", existing_request_ids)
request_ids, errors, warning = await _verify_http_exchange_ids(
cast("Any", object()),
["1042", "1088"],
)
assert request_ids is None
assert errors == ["http_exchange_ids do not exist in the current proxy project: 1088"]
assert warning is None
async def test_http_exchange_ids_are_dropped_when_proxy_cannot_be_queried(
monkeypatch: pytest.MonkeyPatch,
) -> None:
async def existing_request_ids(
_ctx: Any,
_request_ids: list[str],
) -> set[str]:
raise RuntimeError("Caido client is not available")
monkeypatch.setattr(reporting_tool, "existing_request_ids", existing_request_ids)
request_ids, errors, warning = await _verify_http_exchange_ids(
cast("Any", object()),
["1042", "1042", "1088"],
)
assert request_ids is None
assert errors == []
assert warning is not None
assert "not stored" in warning
assert "update_vulnerability_report" in warning
async def test_create_reports_persistence_failure_as_tool_error(
report_state: ReportState,
) -> None:
def fail_persistence(_report: dict[str, Any]) -> None:
raise RuntimeError("persistence failed")
report_state.vulnerability_found_callback = fail_persistence
result = await _do_create(**_CONFIRMED_KWARGS, http_exchange_ids=["1042"])
assert result["success"] is False
assert "persistence failed" in result["error"]
assert "file it again" in result["error"]
assert report_state.vulnerability_reports == []
async def test_evidence_only_update_reports_proxy_outage_as_retryable(
report_state: ReportState,
monkeypatch: pytest.MonkeyPatch,
) -> None:
_seed_weak_report(report_state)
original = dict(report_state.vulnerability_reports[0])
async def existing_request_ids(
_ctx: Any,
_request_ids: list[str],
) -> set[str]:
raise RuntimeError("Caido client is not available")
monkeypatch.setattr(reporting_tool, "existing_request_ids", existing_request_ids)
ctx = ToolContext(
context={"agent_id": "root"},
tool_name="update_vulnerability_report",
tool_call_id="call-1",
tool_arguments="{}",
)
raw = await update_vulnerability_report.on_invoke_tool(
ctx,
json.dumps(
{
"report_id": "vuln-0009",
"update_reason": "A replay produced a clearer proving exchange.",
"http_exchange_ids": ["204"],
}
),
)
result = json.loads(raw)
assert result["success"] is False
assert "No fields to update" not in result["error"]
assert "update_vulnerability_report" in result["error"]
assert result["report_id"] == "vuln-0009"
assert report_state.vulnerability_reports[0] == original
def test_update_reports_persistence_failure_as_tool_error(report_state: ReportState) -> None:
_seed_weak_report(report_state)
original = dict(report_state.vulnerability_reports[0])
def fail_persistence(_report: dict[str, Any]) -> None:
raise RuntimeError("persistence failed")
report_state.vulnerability_updated_callback = fail_persistence
result = _do_update(
report_id="vuln-0009",
update_reason="A replay produced a clearer proving exchange.",
fields={"http_exchange_ids": ["204"]},
)
assert result["success"] is False
assert "persistence failed" in result["error"]
assert result["report_id"] == "vuln-0009"
assert report_state.vulnerability_reports[0] == original
def test_update_vulnerability_report_ignores_identical_content(report_state: ReportState) -> None:
_seed_weak_report(report_state)
assert report_state.update_vulnerability_report("vuln-0009", {"severity": "medium"}) is None
assert "update_history" not in report_state.vulnerability_reports[0]
def test_update_drops_reasoning_left_behind_by_the_field_it_describes(
report_state: ReportState,
) -> None:
"""A rating the update replaces must not keep the rationale for the old one."""
_seed_weak_report(report_state)
report = report_state.vulnerability_reports[0]
report["confidence_rationale"] = "Nothing was executed; the version banner is the only signal."
report["cvss_breakdown"] = {"attack_vector": "network", "user_interaction": "required"}
report["severity_change_conditions"] = "Confirming the write would raise this."
updated = report_state.update_vulnerability_report(
"vuln-0009",
{
"confidence": "high",
"severity": "critical",
"cvss": 9.8,
"severity_change_conditions": "A guard before the write would remove the impact.",
},
)
assert updated is not None
assert "confidence_rationale" not in updated, "the superseded rationale must not survive"
assert "cvss_breakdown" not in updated
assert updated["severity_change_conditions"].startswith("A guard"), (
"a replacement the update supplies is kept, not dropped"
)
assert updated["update_history"][0]["dropped_fields"] == [
"confidence_rationale",
"cvss_breakdown",
]
run_dir = report_state._run_dir
assert run_dir is not None
markdown = (run_dir / "vulnerabilities" / "vuln-0009.md").read_text(encoding="utf-8")
assert "version banner is the only signal" not in markdown
assert "Dropped as superseded: confidence_rationale, cvss_breakdown" in markdown
def test_agent_revises_its_own_report_without_a_duplicate_verdict(
report_state: ReportState,
) -> None:
"""Editing a finding is its own operation: no dedupe verdict is involved."""
_seed_weak_report(report_state)
result = _do_update(
report_id="vuln-0009",
update_reason="An unauthenticated PATCH wrote the file, so the finding is confirmed.",
fields={
"poc_script_code": "PATCH /files/2f1c HTTP/1.1",
"confidence": "HIGH",
"confidence_rationale": "The write was replayed twice.",
"cvss_breakdown": _CVSS,
"severity_change_conditions": "A guard before the write would remove the impact.",
},
agent_id="834f79fb",
agent_name="Directus CVE-2025-55746 Validation Agent",
)
assert result["success"] is True
assert result["action"] == "updated"
assert result["severity"] == "critical"
assert result["cvss_score"] == pytest.approx(9.8)
assert "cvss" in result["updated_fields"], "a new vector carries its own score"
assert len(report_state.vulnerability_reports) == 1
report = report_state.vulnerability_reports[0]
assert report["id"] == "vuln-0009"
assert report["confidence"] == "high"
assert report["agent_id"] == "aaaa1111", "the original reporter stays on the finding"
history = report["update_history"]
assert history[0]["agent_name"] == "Directus CVE-2025-55746 Validation Agent"
assert history[0]["reason"].startswith("An unauthenticated PATCH")
run_dir = report_state._run_dir
assert run_dir is not None
markdown = (run_dir / "vulnerabilities" / "vuln-0009.md").read_text(encoding="utf-8")
assert "PATCH /files/2f1c" in markdown
@pytest.mark.parametrize(
("report_id", "update_reason", "fields", "expected"),
[
(" ", "reason", {"impact": "x"}, "report_id cannot be empty"),
("vuln-0009", " ", {"impact": "x"}, "update_reason cannot be empty"),
("vuln-0009", "reason", {}, "No fields to update"),
("vuln-0404", "reason", {"impact": "x"}, "not found"),
],
)
def test_update_rejects_a_call_it_cannot_act_on(
report_state: ReportState,
report_id: str,
update_reason: str,
fields: dict[str, Any],
expected: str,
) -> None:
_seed_weak_report(report_state)
result = _do_update(report_id=report_id, update_reason=update_reason, fields=fields)
assert result["success"] is False
assert expected in result["error"]
assert "update_history" not in report_state.vulnerability_reports[0]
def test_update_reports_every_invalid_field_at_once(report_state: ReportState) -> None:
_seed_weak_report(report_state)
result = _do_update(
report_id="vuln-0009",
update_reason="Raising the rating.",
fields={
"confidence": "very high",
"fix_effort": "weeks",
"cvss_breakdown": {**_CVSS, "attack_vector": "X"},
"cve": "CVE-BAD",
},
)
assert result["success"] is False
joined = " ".join(result["errors"])
assert "confidence" in joined
assert "fix_effort" in joined
assert "attack_vector" in joined
assert "CVE" in joined
assert report_state.vulnerability_reports[0]["confidence"] == "low", "nothing was applied"
def test_update_wants_verification_for_a_fix_it_would_apply(
report_state: ReportState,
) -> None:
_seed_weak_report(report_state)
result = _do_update(
report_id="vuln-0009",
update_reason="Adding the file the write lands in.",
fields={
"code_locations": [
{
"file": "api/src/controllers/files.ts",
"start_line": 42,
"fix_before": "await storage.write(id, body)",
"fix_after": "await assertPermission(req); await storage.write(id, body)",
}
]
},
)
assert result["success"] is False
assert any("fix_verification" in error for error in result["errors"])
def test_update_says_so_when_the_report_already_carries_it(
report_state: ReportState,
) -> None:
_seed_weak_report(report_state)
result = _do_update(
report_id="vuln-0009",
update_reason="Restating the severity.",
fields={"confidence": "low"},
)
assert result["success"] is False
assert "already says this" in result["error"]
assert result["report_id"] == "vuln-0009"
def test_update_tool_asks_for_the_report_and_the_reason() -> None:
schema = update_vulnerability_report.params_json_schema
assert set(schema["required"]) >= {"report_id", "update_reason"}
assert "cvss_breakdown" in schema["properties"]
assert "id" not in schema["properties"], "identity fields are not editable"
description = update_vulnerability_report.description
assert "not deduplication" in description
def test_update_keeps_an_exploit_out_of_a_dependency_finding(report_state: ReportState) -> None:
"""A dependency record is rated from its advisory, so a revision must not write a
PoC and a dynamic rating onto it. The proof belongs in its own finding."""
_seed_weak_report(report_state)
dependency_report = report_state.vulnerability_reports[0]
dependency_report["finding_class"] = "dependency_cve"
dependency_report["dependency_metadata"] = {
"package_name": "directus",
"installed_version": "11.5.1",
}
result = _do_update(
report_id="vuln-0009",
update_reason="An unauthenticated PATCH wrote the file.",
fields={
"poc_script_code": "PATCH /files/2f1c HTTP/1.1",
"endpoint": "/files/{uuid}",
"cvss_breakdown": _CVSS,
},
)
assert result["success"] is False
assert "dependency_cve" in result["error"]
assert set(result["rejected_fields"]) == {"endpoint", "poc_script_code"}
assert dependency_report["severity"] == "medium"
assert "poc_script_code" not in dependency_report
assert "update_history" not in dependency_report
def _seed_dependency_report(report_state: ReportState) -> dict[str, Any]:
_seed_weak_report(report_state)
dependency_report = report_state.vulnerability_reports[0]
dependency_report["finding_class"] = "dependency_cve"
dependency_report["dependency_metadata"] = {
"package_name": "directus",
"installed_version": "11.5.1",
"manifest_path": "package-lock.json",
"advisory_cvss": 9.8,
"contextual_cvss_breakdown": {**_CVSS, "confidentiality": "L"},
"contextual_cvss_score": 5.3,
"contextual_cvss_vector": "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:L/I:N/A:N",
"contextual_cvss_reasoning": "The vulnerable API is imported but never called.",
}
return dependency_report
def test_update_re_rates_a_dependency_finding_through_its_contextual_cvss(
report_state: ReportState,
) -> None:
"""A dependency finding is rated in the context of the codebase. A revised
breakdown replaces that contextual rating, with the reasoning a reader can
check, and leaves the package identity alone."""
dependency_report = _seed_dependency_report(report_state)
result = _do_update(
report_id="vuln-0009",
update_reason="A call path from the upload handler to the vulnerable API was found.",
fields={
"cvss_breakdown": _CVSS,
"contextual_cvss_reasoning": (
"routes/upload.ts:88 reaches the affected parser with user input."
),
},
)
assert result["success"] is True
assert result["severity"] == "critical"
assert dependency_report["severity"] == "critical"
assert dependency_report["cvss"] == 9.8
assert "cvss_breakdown" not in dependency_report
assert "contextual_cvss_reasoning" not in dependency_report
metadata = dependency_report["dependency_metadata"]
assert metadata["package_name"] == "directus"
assert metadata["installed_version"] == "11.5.1"
assert metadata["manifest_path"] == "package-lock.json"
assert metadata["advisory_cvss"] == 9.8
assert metadata["contextual_cvss_breakdown"] == _CVSS
assert metadata["contextual_cvss_score"] == 9.8
assert metadata["contextual_cvss_vector"] == "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H"
assert metadata["contextual_cvss_reasoning"].startswith("routes/upload.ts:88")
assert dependency_report["finding_class"] == "dependency_cve"
history = dependency_report["update_history"]
assert history[-1]["previous_severity"] == "medium"
assert set(history[-1]["fields"]) == {"cvss", "dependency_metadata", "severity"}
def test_update_wants_the_reasoning_behind_a_dependency_re_rating(
report_state: ReportState,
) -> None:
dependency_report = _seed_dependency_report(report_state)
result = _do_update(
report_id="vuln-0009",
update_reason="The parser is reachable.",
fields={"cvss_breakdown": _CVSS},
)
assert result["success"] is False
assert any("contextual_cvss_reasoning" in error for error in result["errors"])
assert dependency_report["severity"] == "medium"
assert dependency_report["dependency_metadata"]["contextual_cvss_score"] == 5.3
assert "update_history" not in dependency_report
def test_update_corrects_the_reasoning_behind_a_dependency_rating_alone(
report_state: ReportState,
) -> None:
"""The rating on file stays; only its explanation is replaced."""
dependency_report = _seed_dependency_report(report_state)
result = _do_update(
report_id="vuln-0009",
update_reason="The reasoning named the wrong module.",
fields={"contextual_cvss_reasoning": "lib/parser.ts imports it; no call site reaches it."},
)
assert result["success"] is True
assert result["updated_fields"] == ["dependency_metadata"]
assert dependency_report["severity"] == "medium"
assert dependency_report["cvss"] == 5.3
metadata = dependency_report["dependency_metadata"]
assert metadata["contextual_cvss_breakdown"] == {**_CVSS, "confidentiality": "L"}
assert metadata["contextual_cvss_score"] == 5.3
assert metadata["contextual_cvss_reasoning"].startswith("lib/parser.ts")
assert metadata["package_name"] == "directus"
def test_update_wants_a_rating_before_reasoning_about_one(
report_state: ReportState,
) -> None:
_seed_weak_report(report_state)
dependency_report = report_state.vulnerability_reports[0]
dependency_report["finding_class"] = "dependency_cve"
dependency_report["dependency_metadata"] = {
"package_name": "directus",
"installed_version": "11.5.1",
}
result = _do_update(
report_id="vuln-0009",
update_reason="Explaining the rating.",
fields={"contextual_cvss_reasoning": "Reachable."},
)
assert result["success"] is False
assert any("cvss_breakdown is required" in error for error in result["errors"])
assert "contextual_cvss_reasoning" not in dependency_report["dependency_metadata"]
def test_update_keeps_contextual_reasoning_off_a_dynamic_finding(
report_state: ReportState,
) -> None:
_seed_weak_report(report_state)
result = _do_update(
report_id="vuln-0009",
update_reason="Re-rating.",
fields={"cvss_breakdown": _CVSS, "contextual_cvss_reasoning": "Reachable."},
)
assert result["success"] is False
assert result["rejected_fields"] == ["contextual_cvss_reasoning"]
assert report_state.vulnerability_reports[0]["severity"] == "medium"
def test_update_reads_a_legacy_dependency_record_by_its_metadata(
report_state: ReportState,
) -> None:
"""A dependency finding filed before finding_class was persisted still carries
package metadata, so its class is read from that, not defaulted to dynamic."""
_seed_weak_report(report_state)
dependency_report = report_state.vulnerability_reports[0]
dependency_report.pop("finding_class", None)
dependency_report["dependency_metadata"] = {
"package_name": "directus",
"installed_version": "11.5.1",
}
result = _do_update(
report_id="vuln-0009",
update_reason="An unauthenticated PATCH wrote the file.",
fields={"poc_script_code": "PATCH /files/2f1c HTTP/1.1", "cvss_breakdown": _CVSS},
)
assert result["success"] is False
assert "dependency_cve" in result["error"]
assert "poc_script_code" not in dependency_report
def test_update_still_corrects_the_prose_of_a_dependency_finding(
report_state: ReportState,
) -> None:
"""Fields every class carries stay editable on a dependency record."""
_seed_weak_report(report_state)
dependency_report = report_state.vulnerability_reports[0]
dependency_report["finding_class"] = "dependency_cve"
result = _do_update(
report_id="vuln-0009",
update_reason="The advisory names a later fixed release than the report says.",
fields={"remediation_steps": "Upgrade to 11.5.2 or later."},
)
assert result["success"] is True
assert dependency_report["remediation_steps"] == "Upgrade to 11.5.2 or later."
assert dependency_report["finding_class"] == "dependency_cve"
def test_update_refuses_code_locations_it_cannot_use(report_state: ReportState) -> None:
"""A location without a usable file and line is reported, not dropped in silence."""
_seed_weak_report(report_state)
result = _do_update(
report_id="vuln-0009",
update_reason="Naming the vulnerable handler.",
fields={"code_locations": [{"label": "the file write"}]},
)
assert result["success"] is False
assert any("start_line" in error for error in result["errors"])
def _seed_saved_report(report_state: ReportState) -> Path:
"""The weak report, written to disk the way a filed finding is."""
_seed_weak_report(report_state)
report_state._saved_vuln_ids.clear()
report_state.save_run_data()
run_dir = report_state._run_dir
assert run_dir is not None
md_path = run_dir / "vulnerabilities" / "vuln-0009.md"
assert md_path.exists()
return run_dir
def test_delete_withdraws_a_disproved_report_and_its_artifacts(
report_state: ReportState,
) -> None:
"""A deleted finding leaves the run's state and every rendered artifact."""
run_dir = _seed_saved_report(report_state)
persisted: list[dict[str, Any]] = []
report_state.vulnerability_deleted_callback = persisted.append
result = _do_delete(
report_id="vuln-0009",
delete_reason="The cross-tenant read came from the harness reusing the victim's session.",
agent_id="aaaa1111",
agent_name="Recon Agent",
)
assert result["success"] is True
assert result["action"] == "deleted"
assert result["report_id"] == "vuln-0009"
assert result["title"].startswith("Directus 11.5.1")
assert report_state.vulnerability_reports == []
assert not (run_dir / "vulnerabilities" / "vuln-0009.md").exists()
assert json.loads((run_dir / "vulnerabilities.json").read_text(encoding="utf-8")) == []
assert "vuln-0009" not in (run_dir / "vulnerabilities.csv").read_text(encoding="utf-8")
assert len(persisted) == 1
assert persisted[0]["id"] == "vuln-0009"
assert persisted[0]["agent_id"] == "aaaa1111", "the callback sees the finding as filed"
deletion = persisted[0]["deletion"]
assert deletion["agent_id"] == "aaaa1111"
assert deletion["agent_name"] == "Recon Agent"
assert deletion["filed_by_agent_id"] == "aaaa1111"
assert deletion["reason"].startswith("The cross-tenant read")
run = json.loads((run_dir / "run.json").read_text(encoding="utf-8"))
history = run["deleted_vulnerability_reports"]
assert [entry["id"] for entry in history] == ["vuln-0009"]
assert history[0]["severity"] == "medium"
assert history[0]["deleted_at"]
def test_delete_never_reissues_the_withdrawn_id(report_state: ReportState) -> None:
"""The next finding must not inherit a deleted report's id, file or history."""
_seed_saved_report(report_state)
assert report_state.delete_vulnerability_report(
"vuln-0009", delete_reason="Disproved.", deleted_by_agent_id="aaaa1111"
)
new_id = report_state.add_vulnerability_report(title="A real finding", severity="high")
assert new_id == "vuln-0010"
assert [r["id"] for r in report_state.vulnerability_reports] == ["vuln-0010"]
def _assert_report_still_filed(
report_state: ReportState, run_dir: Path, original: dict[str, Any]
) -> None:
"""The report is in memory and in every artifact exactly as it was filed."""
assert report_state.vulnerability_reports == [original]
assert "vuln-0009" in report_state._saved_vuln_ids
assert (run_dir / "vulnerabilities" / "vuln-0009.md").exists()
indexed = json.loads((run_dir / "vulnerabilities.json").read_text(encoding="utf-8"))
assert [r["id"] for r in indexed] == ["vuln-0009"]
assert "vuln-0009" in (run_dir / "vulnerabilities.csv").read_text(encoding="utf-8")
run = json.loads((run_dir / "run.json").read_text(encoding="utf-8"))
assert "deleted_vulnerability_reports" not in run
assert "deleted_vulnerability_reports" not in report_state.run_record
assert report_state._next_report_id() == "vuln-0010"
def test_any_agent_in_the_run_may_delete_a_report(report_state: ReportState) -> None:
"""Agents in a run share one trust boundary: a report another agent filed can be
withdrawn once disproved, and the history keeps both identities apart."""
run_dir = _seed_saved_report(report_state)
calls: list[dict[str, Any]] = []
report_state.vulnerability_deleted_callback = calls.append
result = _do_delete(
report_id="vuln-0009",
delete_reason="Disproved.",
agent_id="834f79fb",
agent_name="Validation Agent",
)
assert result["success"] is True
assert report_state.vulnerability_reports == []
assert [c["id"] for c in calls] == ["vuln-0009"]
assert not (run_dir / "vulnerabilities" / "vuln-0009.md").exists()
(deletion,) = report_state.run_record["deleted_vulnerability_reports"]
assert deletion["filed_by_agent_id"] == "aaaa1111"
assert deletion["agent_id"] == "834f79fb"
assert deletion["agent_name"] == "Validation Agent"
def test_delete_reports_persistence_failure_and_keeps_the_report(
report_state: ReportState,
) -> None:
"""Persistence refused after the local indexes were rewritten: they are rewritten back."""
run_dir = _seed_saved_report(report_state)
original = dict(report_state.vulnerability_reports[0])
def fail_persistence(_report: dict[str, Any]) -> None:
assert report_state.vulnerability_reports == [], "indexes are rewritten first"
raise RuntimeError("persistence failed")
report_state.vulnerability_deleted_callback = fail_persistence
result = _do_delete(report_id="vuln-0009", delete_reason="Disproved.", agent_id="aaaa1111")
assert result["success"] is False
assert "persistence failed" in result["error"]
assert "still on file" in result["error"]
_assert_report_still_filed(report_state, run_dir, original)
def test_delete_reports_an_artifact_write_failure_and_keeps_the_report(
report_state: ReportState, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A disk that cannot take the rewritten run record is a failed deletion, not a
success with stale artifacts; persistence is never told about it."""
run_dir = _seed_saved_report(report_state)
original = dict(report_state.vulnerability_reports[0])
calls: list[dict[str, Any]] = []
report_state.vulnerability_deleted_callback = calls.append
attempts = 0
def write_run_record_once_failing(run_dir_: Path, record: dict[str, Any]) -> None:
nonlocal attempts
attempts += 1
if attempts == 1:
raise OSError(28, "No space left on device")
(run_dir_ / "run.json").write_text(json.dumps(record), encoding="utf-8")
monkeypatch.setattr("strix.report.state.write_run_record", write_run_record_once_failing)
result = _do_delete(report_id="vuln-0009", delete_reason="Disproved.", agent_id="aaaa1111")
assert result["success"] is False
assert "No space left on device" in result["error"]
assert "still on file" in result["error"]
assert calls == []
assert attempts == 2, "the run record is written back after the failed rewrite"
_assert_report_still_filed(report_state, run_dir, original)
def test_delete_reports_when_the_rollback_itself_cannot_be_written(
report_state: ReportState, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A disk that stays full: memory is restored, the caller is told the indexes
may be stale, and the next save rewrites them."""
run_dir = _seed_saved_report(report_state)
original = dict(report_state.vulnerability_reports[0])
calls: list[dict[str, Any]] = []
report_state.vulnerability_deleted_callback = calls.append
disk_full = True
def write_run_record_while_disk_full(run_dir_: Path, record: dict[str, Any]) -> None:
if disk_full:
raise OSError(28, "No space left on device")
write_run_record(run_dir_, record)
monkeypatch.setattr("strix.report.state.write_run_record", write_run_record_while_disk_full)
with pytest.raises(ReportRollbackError) as excinfo:
report_state.delete_vulnerability_report(
"vuln-0009", delete_reason="Disproved.", deleted_by_agent_id="aaaa1111"
)
assert isinstance(excinfo.value.cause, OSError)
assert calls == []
assert report_state.vulnerability_reports == [original]
assert "vuln-0009" in report_state._saved_vuln_ids
assert "deleted_vulnerability_reports" not in report_state.run_record
result = _do_delete(report_id="vuln-0009", delete_reason="Disproved.", agent_id="aaaa1111")
assert result["success"] is False
assert "No space left on device" in result["error"]
assert "may be stale" in result["error"]
disk_full = False
report_state.save_run_data()
_assert_report_still_filed(report_state, run_dir, original)
def test_delete_can_be_retried_after_persistence_recovers(report_state: ReportState) -> None:
run_dir = _seed_saved_report(report_state)
outcomes = iter([RuntimeError("persistence failed"), None])
def flaky_persistence(_report: dict[str, Any]) -> None:
outcome = next(outcomes)
if outcome is not None:
raise outcome
report_state.vulnerability_deleted_callback = flaky_persistence
first = _do_delete(report_id="vuln-0009", delete_reason="Disproved.", agent_id="aaaa1111")
second = _do_delete(report_id="vuln-0009", delete_reason="Disproved.", agent_id="aaaa1111")
assert first["success"] is False
assert second["success"] is True
assert report_state.vulnerability_reports == []
assert not (run_dir / "vulnerabilities" / "vuln-0009.md").exists()
assert json.loads((run_dir / "vulnerabilities.json").read_text(encoding="utf-8")) == []
run = json.loads((run_dir / "run.json").read_text(encoding="utf-8"))
assert [e["id"] for e in run["deleted_vulnerability_reports"]] == ["vuln-0009"]
@pytest.mark.parametrize(
("report_id", "delete_reason", "expected"),
[
(" ", "reason", "report_id cannot be empty"),
("vuln-0009", " ", "delete_reason cannot be empty"),
("vuln-0404", "reason", "not found"),
],
)
def test_delete_rejects_a_call_it_cannot_act_on(
report_state: ReportState,
report_id: str,
delete_reason: str,
expected: str,
) -> None:
_seed_weak_report(report_state)
calls: list[dict[str, Any]] = []
report_state.vulnerability_deleted_callback = calls.append
result = _do_delete(report_id=report_id, delete_reason=delete_reason, agent_id="aaaa1111")
assert result["success"] is False
assert expected in result["error"]
assert len(report_state.vulnerability_reports) == 1
assert calls == [], "nothing reaches persistence for a call that is refused"
async def test_delete_tool_carries_the_caller_identity(report_state: ReportState) -> None:
_seed_saved_report(report_state)
coordinator = type("Coordinator", (), {"names": {"aaaa1111": "Recon Agent"}})()
ctx = ToolContext(
context={"agent_id": "aaaa1111", "coordinator": coordinator},
tool_name="delete_vulnerability_report",
tool_call_id="call-1",
tool_arguments="{}",
)
raw = await delete_vulnerability_report.on_invoke_tool(
ctx,
json.dumps({"report_id": "vuln-0009", "delete_reason": "Disproved on re-test."}),
)
result = json.loads(raw)
assert result["success"] is True
history = report_state.run_record["deleted_vulnerability_reports"]
assert history[0]["agent_id"] == "aaaa1111"
assert history[0]["agent_name"] == "Recon Agent"
assert "update_vulnerability_report" in delete_vulnerability_report.description
Reference in a new issue View git blame Copy permalink