strix/tests/test_vulnerability_update_tool.py

"""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",
        code_locations=[
            {
                "file": "src/redirect.py",
                "start_line": 10,
                "end_line": 12,
                "snippet": "return redirect(url)",
            },
        ],
    )
    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_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:
    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")
    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 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"


@pytest.mark.parametrize(
    "code_locations",
    [[], [{"file": "../invalid.py", "start_line": 1}]],
)
async def test_update_rejects_empty_code_locations(
    report_state: ReportState,
    code_locations: list[dict[str, Any]],
) -> None:
    original_locations = report_state.vulnerability_reports[0]["code_locations"]

    result = await _do_update(
        report_id="vuln-0001",
        update_reason="The source review did not provide a valid location.",
        code_locations=code_locations,
    )

    assert result["success"] is False
    assert any("code_locations" in error for error in result["errors"])
    assert report_state.vulnerability_reports[0]["code_locations"] == original_locations


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"}