fix(viewer): isolate invalid review data in run history

This commit is contained in:
bearsyankees
2026-09-17 10:31:29 -04:00
parent 65188b28fb
commit 327c78c2ca
8 changed files with 126 additions and 7 deletions

View File

@@ -167,7 +167,11 @@ export default function PastRunsView({
{run.open_count !== undefined && <span>· {run.open_count} open · {run.closed_count ?? 0} false positives · {run.detected_count ?? run.open_count} found</span>}
</div>
</div>
<SeverityChips counts={run.severity_counts} />
{run.severity_counts ? (
<SeverityChips counts={run.severity_counts} />
) : (
<span className="text-xs text-[#888]">Review unavailable</span>
)}
<ChevronRight className="h-4 w-4 flex-shrink-0 text-[#555] transition-colors group-hover:text-[#aaa]" aria-hidden="true" />
</button>
);

View File

@@ -210,7 +210,7 @@ export interface RunListEntry {
start_time: string | null;
end_time: string | null;
finished: boolean;
severity_counts: RunSeverityCounts;
severity_counts: RunSeverityCounts | null;
open_count?: number;
closed_count?: number;
detected_count?: number;

View File

@@ -71,9 +71,7 @@ def _iter_run_dirs(base_dir: Path) -> list[Path]:
def run_list_entry(run_dir: Path) -> dict[str, Any]:
"""Compact summary of a single run for the history list."""
record = read_run_summary(run_dir)
findings = read_triaged_vulnerabilities(run_dir)
active = [finding for finding in findings if finding.get("status") != "closed"]
return {
entry = {
"name": record.get("run_name") or run_dir.name,
"target": primary_target(record),
"scan_mode": record.get("scan_mode"),
@@ -81,6 +79,16 @@ def run_list_entry(run_dir: Path) -> dict[str, Any]:
"start_time": record.get("start_time"),
"end_time": record.get("end_time"),
"finished": bool(record.get("finished")),
}
try:
findings = read_triaged_vulnerabilities(run_dir)
except TriageError:
# Keep the run selectable without claiming its review counts are known.
# Its findings endpoint still reports the error and preserves the data.
return {**entry, "severity_counts": None}
active = [finding for finding in findings if finding.get("status") != "closed"]
return {
**entry,
"severity_counts": severity_counts(active),
"open_count": len(active),
"closed_count": len(findings) - len(active),

View File

@@ -6,7 +6,7 @@
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<meta name="color-scheme" content="dark" />
<title>Strix Results</title>
<script type="module" crossorigin src="./assets/index-CMYkx6m6.js"></script>
<script type="module" crossorigin src="./assets/index-RaZRtkBA.js"></script>
<link rel="stylesheet" crossorigin href="./assets/index-DNOJaOMX.css">
</head>
<body>

View File

@@ -109,6 +109,8 @@ def _file_lock(descriptor: int, *, unlock: bool = False) -> None:
if sys.platform == "win32":
import msvcrt # noqa: PLC0415
# The CRT permits locking beyond EOF, so even an empty file has byte zero
# available as a stable lock region. Always unlock that same region.
os.lseek(descriptor, 0, os.SEEK_SET)
msvcrt.locking(descriptor, msvcrt.LK_UNLCK if unlock else msvcrt.LK_NBLCK, 1)
else:

View File

@@ -0,0 +1,40 @@
"""Native operating-system locking invariants for the local review sidecar."""
from __future__ import annotations
import os
from typing import TYPE_CHECKING
import pytest
from strix.report.triage_store import _file_lock
if TYPE_CHECKING:
from pathlib import Path
@pytest.mark.parametrize("contents", [b"", b"existing lock file"])
def test_file_lock_excludes_other_handles_and_releases(tmp_path: Path, contents: bytes) -> None:
"""Exercise fcntl on POSIX and the real msvcrt implementation on Windows."""
path = tmp_path / "triage.lock"
path.write_bytes(contents)
first = os.open(path, os.O_RDWR)
try:
second = os.open(path, os.O_RDWR)
try:
_file_lock(first)
try:
with pytest.raises(OSError):
_file_lock(second)
# Locking/unlocking must agree on byte zero, regardless of position.
os.lseek(first, 100, os.SEEK_SET)
finally:
_file_lock(first, unlock=True)
_file_lock(second)
_file_lock(second, unlock=True)
finally:
os.close(second)
finally:
os.close(first)
assert path.read_bytes() == contents

View File

@@ -261,6 +261,71 @@ def test_corrupt_sidecar_is_a_visible_error_not_empty_findings(viewer: ViewerCli
assert read_vulnerabilities(viewer.run_dir)[0]["evidence"] == "original evidence"
@pytest.mark.parametrize("damage", ["malformed", "null", "schema", "identity", "findings"])
def test_invalid_historical_run_does_not_break_history(
viewer: ViewerClient, monkeypatch: pytest.MonkeyPatch, damage: str
) -> None:
other = viewer.run_dir.parent / "damaged"
other.mkdir()
(other / "run.json").write_text(
json.dumps({"run_name": "damaged", "status": "completed", "end_time": "2026-09-15"}),
encoding="utf-8",
)
(other / "vulnerabilities.json").write_bytes(
(viewer.run_dir / "vulnerabilities.json").read_bytes()
)
finding = read_triaged_vulnerabilities(other)[0]
triage_finding(
other,
finding["id"],
status="closed",
expected_revision=0,
reviewed_digest=finding["finding_digest"],
surface="tui",
)
sidecar = other / "triage.json"
document = json.loads(sidecar.read_text(encoding="utf-8"))
if damage == "malformed":
sidecar.write_text("{", encoding="utf-8")
elif damage == "null":
sidecar.write_text("null", encoding="utf-8")
elif damage == "findings":
(other / "vulnerabilities.json").write_text("[null]", encoding="utf-8")
else:
if damage == "schema":
document["schema_version"] = 2
else:
document["run_identity"]["run_id"] = "another-run"
sidecar.write_text(json.dumps(document), encoding="utf-8")
saved = sidecar.read_bytes()
evidence = (other / "vulnerabilities.json").read_bytes()
# The locked history must not try to read the damaged run's findings.
assert viewer.request("/api/runs") == (200, {"locked": True, "count": 2, "runs": []})
monkeypatch.setattr("strix.interface.viewer.auth.is_verified", lambda: True)
status, payload = viewer.request("/api/runs")
assert status == 200
assert payload["count"] == 2
runs = {run["name"]: run for run in payload["runs"]}
assert runs["example"]["open_count"] == 1
assert runs["example"]["severity_counts"]["high"] == 1
assert runs["damaged"]["status"] == "completed"
assert runs["damaged"]["severity_counts"] is None
assert all(
key not in runs["damaged"] for key in ("open_count", "closed_count", "detected_count")
)
assert viewer.finding()["status"] == "open"
assert viewer.request("/api/vulnerabilities?run=damaged")[0] == 409
body = {
"status": "open",
"expected_revision": 1,
"reviewed_digest": finding["finding_digest"],
}
assert viewer.request("/api/vulnerabilities/vuln-0001/triage?run=damaged", body)[0] == 409
assert sidecar.read_bytes() == saved
assert (other / "vulnerabilities.json").read_bytes() == evidence
def test_read_only_run_has_no_write_capability(viewer: ViewerClient) -> None:
viewer.run_dir.chmod(0o500)
try: