From 18556da0cdb263f9a82ce3798f1e9322059f37e6 Mon Sep 17 00:00:00 2001 From: Alex Schapiro Date: Wed, 19 Aug 2026 21:28:10 +0000 Subject: [PATCH] Add vulnerability report amendments --- strix/agents/factory.py | 2 + strix/agents/prompts/system_prompt.jinja | 7 +- strix/report/state.py | 54 ++++- strix/report/writer.py | 31 ++- strix/tools/finish/tool.py | 15 +- strix/tools/reporting/tool.py | 289 ++++++++++++++++++++++- tests/test_vulnerability_update_tool.py | 223 +++++++++++++++++ 7 files changed, 606 insertions(+), 15 deletions(-) create mode 100644 tests/test_vulnerability_update_tool.py diff --git a/strix/agents/factory.py b/strix/agents/factory.py index 9d599a54..9628b78f 100644 --- a/strix/agents/factory.py +++ b/strix/agents/factory.py @@ -48,6 +48,7 @@ from strix.tools.reporting.tool import ( create_vulnerability_report, get_report, list_reports, + update_vulnerability_report, ) from strix.tools.respond.tool import respond_to_user from strix.tools.thinking.tool import think @@ -500,6 +501,7 @@ _BASE_TOOLS: tuple[Tool, ...] = ( delete_note, web_search, create_vulnerability_report, + update_vulnerability_report, create_dependency_report, list_reports, get_report, diff --git a/strix/agents/prompts/system_prompt.jinja b/strix/agents/prompts/system_prompt.jinja index 23493d2d..77688fa8 100644 --- a/strix/agents/prompts/system_prompt.jinja +++ b/strix/agents/prompts/system_prompt.jinja @@ -214,11 +214,12 @@ VALIDATION REQUIREMENTS: - Treat public metadata, internal-looking identifiers, source maps without secrets, and transport/configuration hygiene as observations unless validation proves unauthorized restricted-data access, modification, or service disruption - Every non-None Confidentiality, Integrity, or Availability metric must map to explicit evidence in the report; use Scope Changed only for a demonstrated crossing of security authorities - Independent verification through subagent -- Document complete attack chain +- Document complete attack chains. Amend an existing finding when new evidence increases or decreases its impact on the same asset and root cause - Keep going until you find something that matters -- A vulnerability is ONLY considered reported when a reporting agent uses create_vulnerability_report (or create_dependency_report for known-CVE dependency/supply-chain findings) with full details. Mentions in agent_finish, finish_scan, or generic messages are NOT sufficient +- A vulnerability is ONLY considered reported when a reporting agent uses create_vulnerability_report or update_vulnerability_report (or create_dependency_report for known-CVE dependency/supply-chain findings) with full details. Mentions in agent_finish, finish_scan, or generic messages are NOT sufficient - Reporting and fixing are ONE step, not two: when source is available, the reporting agent derives the concrete fix and files it INLINE via create_vulnerability_report (`code_locations` with `fix_before`/`fix_after` + `fix_pr_body`) — the report is not complete without it. Do NOT report first and then spawn a separate downstream agent to re-derive and re-apply the same patch; that just re-does the analysis and wastes tokens. (Do not silently patch a finding WITHOUT filing a report — the report, with its embedded fix, is the deliverable.) - DEDUPLICATION: The create_vulnerability_report tool uses LLM-based deduplication. If it rejects your report as a duplicate, DO NOT attempt to re-submit the same vulnerability. Accept the rejection and move on to testing other areas. The vulnerability has already been reported by another agent +- DISTINCT ISSUE VS AMENDMENT: File a distinct issue with create_vulnerability_report. Use update_vulnerability_report when new evidence proves greater or lower impact for the same root cause on the same asset. Supply an evidence-backed update_reason and recalculate the rating from cvss_breakdown - REVIEWING FILED FINDINGS (orchestrator/root agent): use list_reports to see every vulnerability filed so far in this scan (by any agent, root or child) — metadata-first with per-severity counts — and get_report to read one finding in full by its id. These are read-only orchestration tools: the root agent uses them to track coverage, avoid dispatching work on already-covered ground, assemble the finish_scan executive summary, and reason about attack-chaining across confirmed findings. Leaf/specialist agents should NOT call them — just do your assigned testing and file findings. Each entry shows which agent filed it (agent_name), and your own entries are flagged by_you. list_notes/get_note do the same for notes. @@ -372,7 +373,7 @@ CRITICAL RULES: - **REALISTIC OUTCOMES** - Some tests find nothing, some validations fail - **ONE AGENT = ONE TASK** - Don't let agents do multiple unrelated jobs - **SPAWN REACTIVELY** - Create new agents based on what you discover -- **ONLY REPORTING AGENTS** can use create_vulnerability_report tool +- **ONLY REPORTING AGENTS** can use create_vulnerability_report or update_vulnerability_report tools - **AGENT SPECIALIZATION MANDATORY** - Each agent must be highly specialized; prefer 1–3 skills, up to 5 for complex contexts - **NO GENERIC AGENTS** - Avoid creating broad, multi-purpose agents that dilute focus diff --git a/strix/report/state.py b/strix/report/state.py index 6f111178..6bda587f 100644 --- a/strix/report/state.py +++ b/strix/report/state.py @@ -149,6 +149,7 @@ class ReportState: self.caido_url: str | None = None self.vulnerability_found_callback: Callable[[dict[str, Any]], None] | None = None + self.vulnerability_updated_callback: Callable[[dict[str, Any]], None] | None = None self._sarif_repo_ctx: dict[str, Any] | None = None self._sarif_repo_ctx_ready: bool = False @@ -315,6 +316,53 @@ class ReportState: self.save_run_data() return report_id + def update_vulnerability_report( + self, + report_id: str, + update_reason: str, + **updates: Any, + ) -> dict[str, Any] | None: + """Amend a filed vulnerability report and persist the updated state.""" + report = next( + (item for item in self.vulnerability_reports if item.get("id") == report_id), + None, + ) + if report is None: + return None + + timestamp = datetime.now(UTC).isoformat() + previous_severity = report.get("severity") + previous_cvss = report.get("cvss") + changed_fields: list[str] = [] + for field, value in updates.items(): + if report.get(field) != value: + changed_fields.append(field) + report[field] = value + + history_entry: dict[str, Any] = { + "timestamp": timestamp, + "update_reason": update_reason.strip(), + "fields_changed": changed_fields, + } + if ("severity" in updates and updates.get("severity") != previous_severity) or ( + "cvss" in updates and updates.get("cvss") != previous_cvss + ): + history_entry["previous_severity"] = previous_severity + history_entry["previous_cvss_score"] = previous_cvss + + history = report.setdefault("update_history", []) + if not isinstance(history, list): + history = [] + report["update_history"] = history + history.append(history_entry) + report["updated_at"] = timestamp + + self._saved_vuln_ids.discard(report_id) + if self.vulnerability_updated_callback: + self.vulnerability_updated_callback(report) + self.save_run_data() + return report + def get_existing_vulnerabilities(self) -> list[dict[str, Any]]: return list(self.vulnerability_reports) @@ -454,7 +502,11 @@ class ReportState: run_dir.mkdir(parents=True, exist_ok=True) if self.final_scan_result: - write_executive_report(run_dir, self.final_scan_result) + write_executive_report( + run_dir, + self.final_scan_result, + self.vulnerability_reports, + ) if self.vulnerability_reports: write_vulnerabilities(run_dir, self.vulnerability_reports, self._saved_vuln_ids) diff --git a/strix/report/writer.py b/strix/report/writer.py index 2cdbae22..3c11cea5 100644 --- a/strix/report/writer.py +++ b/strix/report/writer.py @@ -113,12 +113,20 @@ def write_run_record(run_dir: Path, run_record: dict[str, Any]) -> None: ) -def write_executive_report(run_dir: Path, final_scan_result: str) -> None: +def write_executive_report( + run_dir: Path, + final_scan_result: str, + vulnerability_reports: list[dict[str, Any]] | None = None, +) -> None: path = run_dir / "penetration_test_report.md" with path.open("w", encoding="utf-8") as f: f.write("# Security Penetration Test Report\n\n") f.write(f"**Generated:** {datetime.now(UTC).strftime('%Y-%m-%d %H:%M:%S UTC')}\n\n") f.write(f"{final_scan_result}\n") + if vulnerability_reports: + f.write("\n## Vulnerability Findings\n\n") + for report in vulnerability_reports: + f.write(f"{render_vulnerability_md(report)}\n") logger.info("Saved final penetration test report to: %s", path) @@ -197,6 +205,8 @@ def render_vulnerability_md(report: dict[str, Any]) -> str: # noqa: PLR0912, PL f"**Severity:** {report.get('severity', 'unknown').upper()}", f"**Found:** {report.get('timestamp', 'unknown')}", ] + if report.get("updated_at"): + lines.append(f"**Updated:** {report['updated_at']}") dep_meta = report.get("dependency_metadata") or {} metadata: list[tuple[str, Any]] = [ @@ -304,4 +314,23 @@ def render_vulnerability_md(report: dict[str, Any]) -> str: # noqa: PLR0912, PL lines.append(str(report["assumptions"])) lines.append("") + update_history = report.get("update_history") + if isinstance(update_history, list) and update_history: + lines.append("## Amendment History\n") + for entry in update_history: + if not isinstance(entry, dict): + continue + timestamp = entry.get("timestamp", "unknown") + reason = entry.get("update_reason", "") + fields = ", ".join(str(field) for field in entry.get("fields_changed", [])) + lines.append(f"- **{timestamp}:** {reason}") + lines.append(f" Changed fields: {fields or 'none'}") + if "previous_severity" in entry or "previous_cvss_score" in entry: + lines.append( + " Previous rating: " + f"{str(entry.get('previous_severity', 'unknown')).upper()} " + f"(CVSS {entry.get('previous_cvss_score', 'unknown')})" + ) + lines.append("") + return "\n".join(lines) diff --git a/strix/tools/finish/tool.py b/strix/tools/finish/tool.py index b9704ad1..296460c4 100644 --- a/strix/tools/finish/tool.py +++ b/strix/tools/finish/tool.py @@ -133,14 +133,15 @@ async def finish_scan( combination. You may rule out combinations you can confidently call unrelated — note why instead of padding chains. Any validated chain must already be filed via - ``create_vulnerability_report`` — a demonstrated end-to-end chain - is a PoC-backed vulnerability, so it uses that tool even when one - link is a dependency CVE (the standalone CVE stays in its own - ``create_dependency_report``) — and surfaced prominently in + ``create_vulnerability_report`` — or update the existing finding with + ``update_vulnerability_report`` when the chain amplifies that finding + on the same asset and root cause. A demonstrated new chain is a + PoC-backed vulnerability, so file it even when one link is a + dependency CVE. Keep the standalone CVE in its own + ``create_dependency_report``. Surface the result prominently in ``executive_summary`` / ``technical_analysis``. Finding no real - chain after a serious attempt is acceptable; skipping the - chaining reasoning, or ignoring a plausibly-related combination, - is not. + chain after a serious attempt is acceptable. Skipping the chaining + reasoning, or ignoring a plausibly-related combination, is not. **Calling this multiple times overwrites the previous report.** Make the single call comprehensive. diff --git a/strix/tools/reporting/tool.py b/strix/tools/reporting/tool.py index 6cae75f1..455e2d9b 100644 --- a/strix/tools/reporting/tool.py +++ b/strix/tools/reporting/tool.py @@ -1,8 +1,8 @@ -"""Reporting tools — file vuln findings (with dedup + CVSS) and read them back. +"""Reporting tools — file findings (with dedup + CVSS) and read them back. ``create_vulnerability_report`` / ``create_dependency_report`` file findings; -``list_reports`` / ``get_report`` let any agent (notably the root orchestrator) -review what's been filed so far across the whole scan. +``update_vulnerability_report`` amends a known dynamic finding; +``list_reports`` / ``get_report`` let the root orchestrator review findings. """ from __future__ import annotations @@ -162,6 +162,24 @@ _REQUIRED_FIELDS = { _VALID_FIX_EFFORT = frozenset({"trivial", "low", "medium", "high"}) +_AMENDABLE_FIELDS = ( + "title", + "description", + "impact", + "technical_analysis", + "poc_description", + "poc_script_code", + "remediation_steps", + "evidence", + "assumptions", + "fix_effort", + "cvss_breakdown", + "endpoint", + "method", + "cwe", + "code_locations", +) + async def _do_create( # noqa: PLR0912 *, @@ -329,6 +347,200 @@ async def _do_create( # noqa: PLR0912 } +async def _do_update( # noqa: PLR0911, PLR0912, PLR0915 + *, + report_id: str, + update_reason: str, + title: str | None = None, + description: str | None = None, + impact: str | None = None, + technical_analysis: str | None = None, + poc_description: str | None = None, + poc_script_code: str | None = None, + remediation_steps: str | None = None, + evidence: str | None = None, + assumptions: str | None = None, + fix_effort: str | None = None, + cvss_breakdown: dict[str, str] | None = None, + endpoint: str | None = None, + method: str | None = None, + cwe: str | None = None, + code_locations: list[dict[str, Any]] | None = None, + target: str | None = None, + cve: str | None = None, + dependency_metadata: dict[str, Any] | None = None, +) -> dict[str, Any]: + """Validate and amend one known dynamic vulnerability report.""" + immutable_attempts = { + name: value + for name, value in { + "target": target, + "cve": cve, + "dependency_metadata": dependency_metadata, + }.items() + if value is not None + } + if immutable_attempts: + fields = ", ".join(immutable_attempts) + return { + "success": False, + "error": ( + f"Cannot amend {fields}. File a new report instead because " + "the target and dependency metadata identify the finding." + ), + } + + if not report_id.strip(): + return {"success": False, "error": "report_id cannot be empty"} + if not update_reason.strip(): + return {"success": False, "error": "update_reason cannot be empty"} + + raw_updates = { + name: value + for name, value in { + "title": title, + "description": description, + "impact": impact, + "technical_analysis": technical_analysis, + "poc_description": poc_description, + "poc_script_code": poc_script_code, + "remediation_steps": remediation_steps, + "evidence": evidence, + "assumptions": assumptions, + "fix_effort": fix_effort, + "cvss_breakdown": cvss_breakdown, + "endpoint": endpoint, + "method": method, + "cwe": cwe, + "code_locations": code_locations, + }.items() + if value is not None and name in _AMENDABLE_FIELDS + } + if not raw_updates: + return { + "success": False, + "error": "At least one amendable field must be supplied", + } + + try: + from strix.report.state import get_global_report_state + + report_state = get_global_report_state() + if report_state is None: + return { + "success": False, + "error": "Report state is unavailable", + } + + existing = report_state.get_existing_vulnerabilities() + report = next((item for item in existing if item.get("id") == report_id), None) + if report is None: + valid_ids = [str(item["id"]) for item in existing if item.get("id")] + result: dict[str, Any] = { + "success": False, + "error": f"Report with id '{report_id}' was not found", + } + if valid_ids: + result["valid_report_ids"] = valid_ids + return result + + errors: list[str] = [] + updates: dict[str, Any] = {} + for field, value in raw_updates.items(): + if field in { + "title", + "description", + "impact", + "technical_analysis", + "poc_description", + "poc_script_code", + "remediation_steps", + "evidence", + "assumptions", + "endpoint", + "method", + }: + updates[field] = str(value).strip() + + if "fix_effort" in raw_updates: + normalized_fix_effort = str(fix_effort).strip().lower() + if normalized_fix_effort not in _VALID_FIX_EFFORT: + errors.append( + f"Invalid fix_effort: {normalized_fix_effort!r}. " + f"Must be one of: {sorted(_VALID_FIX_EFFORT)}" + ) + else: + updates["fix_effort"] = normalized_fix_effort + + if "cvss_breakdown" in raw_updates: + if not isinstance(cvss_breakdown, dict) or not cvss_breakdown: + errors.append("cvss_breakdown: must be an object with the 8 CVSS metrics") + else: + for name, valid in _CVSS_VALID.items(): + metric_value = cvss_breakdown.get(name) + if metric_value not in valid: + errors.append( + f"Invalid {name}: {metric_value}. Must be one of: {valid}", + ) + updates["cvss_breakdown"] = cvss_breakdown + + if "cwe" in raw_updates: + parsed_cwe = _extract_cwe(str(cwe)) + cwe_err = _validate_cwe(parsed_cwe) + if cwe_err: + errors.append(cwe_err) + else: + updates["cwe"] = parsed_cwe + + if "code_locations" in raw_updates: + if not isinstance(code_locations, list): + errors.append("code_locations must be a list") + else: + parsed_locations = _normalize_code_locations(code_locations) + if parsed_locations: + errors.extend(_validate_code_locations(parsed_locations)) + updates["code_locations"] = parsed_locations or [] + + if errors: + return {"success": False, "error": "Validation failed", "errors": errors} + + if "cvss_breakdown" in updates: + try: + cvss_score, severity, _vector = _calculate_cvss(updates["cvss_breakdown"]) + except ValueError as exc: + return {"success": False, "error": "Validation failed", "errors": [str(exc)]} + updates["cvss"] = cvss_score + updates["severity"] = severity + + updated = report_state.update_vulnerability_report( + report_id, + update_reason, + **updates, + ) + if updated is None: + return { + "success": False, + "error": f"Report with id '{report_id}' was not found", + } + except (AttributeError, KeyError, TypeError, ValueError) as exc: + logger.exception("update_vulnerability_report persistence failed") + return {"success": False, "error": f"Failed to update vulnerability report: {exc!s}"} + + logger.info( + "Vulnerability report updated: id=%s fields=%s", + report_id, + sorted(raw_updates), + ) + return { + "success": True, + "message": f"Vulnerability report '{report_id}' updated successfully", + "report_id": report_id, + "severity": updated.get("severity"), + "cvss_score": updated.get("cvss"), + "updated_at": updated.get("updated_at"), + } + + def _caller_identity(ctx: RunContextWrapper) -> tuple[str | None, str | None]: """Return the (agent_id, agent_name) of the agent invoking this tool.""" inner = ctx.context if isinstance(ctx.context, dict) else {} @@ -377,6 +589,8 @@ async def create_vulnerability_report( - Suspicions you haven't confirmed with a PoC. - Tracking multiple vulnerabilities at once — one report per vuln. - Re-reporting something you (or another agent) already filed. + - A chain that only amplifies an existing finding's impact on the same + asset and root cause. Use ``update_vulnerability_report`` instead. - Known-CVE dependency / supply-chain findings that can't be dynamically PoC'd — a vulnerable dependency version pinned in a lockfile/manifest that matches a published advisory. File those @@ -701,6 +915,75 @@ async def create_vulnerability_report( return json.dumps(result, ensure_ascii=False, default=str) +@function_tool(timeout=180, strict_mode=False) +async def update_vulnerability_report( + ctx: RunContextWrapper, + report_id: str, + update_reason: str, + title: str | None = None, + description: str | None = None, + impact: str | None = None, + technical_analysis: str | None = None, + poc_description: str | None = None, + poc_script_code: str | None = None, + remediation_steps: str | None = None, + evidence: str | None = None, + assumptions: str | None = None, + fix_effort: str | None = None, + cvss_breakdown: dict[str, str] | None = None, + endpoint: str | None = None, + method: str | None = None, + cwe: str | None = None, + code_locations: list[dict[str, Any]] | None = None, + target: str | None = None, + cve: str | None = None, + dependency_metadata: dict[str, Any] | None = None, +) -> str: + """Amend a known vulnerability report when new evidence changes its impact. + + Use ``create_vulnerability_report`` when the chain proves a distinct issue. + Use this tool when the same root cause on the same asset has greater impact. + The tool recomputes severity and score from ``cvss_breakdown``. + + Apply the same report output rules and CVSS calibration discipline as the + create tool. An amendment must use evidence from a validated result. Do + not use this tool for a speculative upgrade. + + Pass at least one amendable field: + + - ``title``, ``description``, ``impact``, ``technical_analysis``, + ``poc_description``, ``poc_script_code``, ``remediation_steps``, + ``evidence``, ``assumptions``, ``fix_effort``, ``cvss_breakdown``, + ``endpoint``, ``method``, ``cwe``, or ``code_locations``. + + Do not change ``target``, ``cve``, or dependency metadata. File a new + report instead when those values must change. + """ + result = await _do_update( + report_id=report_id, + update_reason=update_reason, + title=title, + description=description, + impact=impact, + technical_analysis=technical_analysis, + poc_description=poc_description, + poc_script_code=poc_script_code, + remediation_steps=remediation_steps, + evidence=evidence, + assumptions=assumptions, + fix_effort=fix_effort, + cvss_breakdown=cvss_breakdown, + endpoint=endpoint, + method=method, + cwe=cwe, + code_locations=code_locations, + target=target, + cve=cve, + dependency_metadata=dependency_metadata, + ) + return json.dumps(result, ensure_ascii=False, default=str) + + _DEP_SEVERITY_FROM_CVSS = { (9.0, 10.0): "critical", (7.0, 9.0): "high", diff --git a/tests/test_vulnerability_update_tool.py b/tests/test_vulnerability_update_tool.py new file mode 100644 index 00000000..b8332ea0 --- /dev/null +++ b/tests/test_vulnerability_update_tool.py @@ -0,0 +1,223 @@ +"""Tests for amending filed vulnerability reports.""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING, Any + +import pytest + +from strix.report.state import ReportState, set_global_report_state +from strix.tools.reporting.tool import _do_update + + +if TYPE_CHECKING: + from pathlib import Path + + +_LOW_CVSS = { + "attack_vector": "N", + "attack_complexity": "H", + "privileges_required": "H", + "user_interaction": "R", + "scope": "U", + "confidentiality": "L", + "integrity": "N", + "availability": "N", +} + +_CRITICAL_CVSS = { + "attack_vector": "N", + "attack_complexity": "L", + "privileges_required": "N", + "user_interaction": "N", + "scope": "U", + "confidentiality": "H", + "integrity": "H", + "availability": "H", +} + + +@pytest.fixture(autouse=True) +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) + state.add_vulnerability_report( + title="Unsafe redirect", + severity="low", + description="The redirect accepts attacker input.", + impact="Limited redirect manipulation.", + target="https://app.example.com", + technical_analysis="The handler does not validate the destination.", + poc_description="Send a crafted redirect value.", + poc_script_code="GET /redirect?url=https://example.net", + remediation_steps="Validate redirect destinations.", + evidence="The response contains the attacker-controlled destination.", + assumptions="Assumes a victim follows the link.", + fix_effort="medium", + cvss=3.1, + cvss_breakdown=_LOW_CVSS, + endpoint="/redirect", + method="GET", + cwe="CWE-601", + ) + return state + + +async def test_update_rejects_unknown_report_id() -> None: + result = await _do_update( + report_id="vuln-9999", + update_reason="The validation pass proved broader impact.", + impact="Broader impact.", + ) + + assert result["success"] is False + assert "vuln-9999" in result["error"] + assert result["valid_report_ids"] == ["vuln-0001"] + + +async def test_update_requires_an_amendable_field() -> None: + result = await _do_update( + report_id="vuln-0001", + update_reason="The validation pass found no new field to amend.", + ) + + assert result == { + "success": False, + "error": "At least one amendable field must be supplied", + } + + +async def test_update_requires_nonempty_reason() -> None: + result = await _do_update( + report_id="vuln-0001", + update_reason=" ", + impact="The impact is broader.", + ) + + assert result == {"success": False, "error": "update_reason cannot be empty"} + + +async def test_update_rejects_target_and_cve_changes() -> None: + result = await _do_update( + report_id="vuln-0001", + update_reason="The chain reached a second asset.", + target="https://other.example.com", + cve="CVE-2024-12345", + ) + + assert result["success"] is False + assert "File a new report instead" in result["error"] + assert "target" in result["error"] + assert "cve" in result["error"] + + +async def test_update_changes_impact_only(report_state: ReportState) -> None: + result = await _do_update( + report_id="vuln-0001", + update_reason="The confirmed chain exposes account data.", + impact="The chain exposes account data.", + ) + + assert result["success"] is True + report = report_state.vulnerability_reports[0] + assert report["impact"] == "The chain exposes account data." + assert report["severity"] == "low" + assert report["cvss"] == 3.1 + assert report["update_history"][0]["fields_changed"] == ["impact"] + assert "previous_severity" not in report["update_history"][0] + + +async def test_cvss_update_recomputes_score_and_severity(report_state: ReportState) -> None: + result = await _do_update( + report_id="vuln-0001", + update_reason="The exploit chain proves full account compromise.", + cvss_breakdown=_CRITICAL_CVSS, + ) + + assert result["success"] is True + report = report_state.vulnerability_reports[0] + assert report["severity"] == "critical" + assert report["cvss"] == 9.8 + history = report["update_history"][0] + assert history["fields_changed"] == ["cvss_breakdown", "cvss", "severity"] + assert history["previous_severity"] == "low" + assert history["previous_cvss_score"] == 3.1 + + +async def test_update_history_is_append_only(report_state: ReportState) -> None: + await _do_update( + report_id="vuln-0001", + update_reason="The chain proves account access.", + impact="Account access is possible.", + ) + await _do_update( + report_id="vuln-0001", + update_reason="The second proof confirms persistent access.", + evidence="The second proof confirms persistent access.", + ) + + history = report_state.vulnerability_reports[0]["update_history"] + assert len(history) == 2 + assert history[0]["update_reason"] == "The chain proves account access." + assert history[1]["update_reason"] == "The second proof confirms persistent access." + assert all("description" not in entry for entry in history) + + +async def test_update_callback_fires(report_state: ReportState) -> None: + updated: list[dict[str, Any]] = [] + report_state.vulnerability_updated_callback = updated.append + + await _do_update( + report_id="vuln-0001", + update_reason="The new proof confirms data exposure.", + evidence="The new proof confirms data exposure.", + ) + + assert len(updated) == 1 + assert updated[0] is report_state.vulnerability_reports[0] + + +async def test_update_persists_all_report_artifacts(report_state: ReportState) -> None: + report_state.final_scan_result = "# Executive Summary\n\nThe scan is complete." + report_state.save_run_data() + + result = await _do_update( + report_id="vuln-0001", + update_reason="The chain proves account takeover.", + description="The redirect reaches the account takeover flow.", + impact="An attacker can take over an account.", + cvss_breakdown=_CRITICAL_CVSS, + ) + + assert result["success"] is True + run_dir = report_state.get_run_dir() + finding_md = (run_dir / "vulnerabilities" / "vuln-0001.md").read_text(encoding="utf-8") + executive_md = (run_dir / "penetration_test_report.md").read_text(encoding="utf-8") + findings = json.loads((run_dir / "vulnerabilities.json").read_text(encoding="utf-8")) + sarif = json.loads((run_dir / "findings.sarif").read_text(encoding="utf-8")) + sarif_finding = sarif["runs"][0]["results"][0] + + assert "An attacker can take over an account." in finding_md + assert "An attacker can take over an account." in executive_md + assert findings[0]["impact"] == "An attacker can take over an account." + assert findings[0]["severity"] == "critical" + assert sarif_finding["properties"]["strix"]["impact"] == ( + "An attacker can take over an account." + ) + assert sarif_finding["properties"]["strix"]["severity"] == "critical" + + +async def test_update_fails_without_global_report_state( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr("strix.report.state._global_report_state", None) + + result = await _do_update( + report_id="vuln-0001", + update_reason="The new proof confirms broader impact.", + impact="Broader impact.", + ) + + assert result == {"success": False, "error": "Report state is unavailable"}