Fix SARIF review edge cases

strix/interface/main.py

@@ -500,13 +500,18 @@ def display_completion_message(args: argparse.Namespace, results_path: Path) ->
 
 
 def write_requested_sarif_output(args: argparse.Namespace, results_path: Path) -> Path | None:
+    """Write SARIF output when requested and report write failures without crashing."""
     if not args.sarif and not args.sarif_output:
         return None
 
     output_path = Path(args.sarif_output) if args.sarif_output else results_path / "results.sarif"
     tracer = get_global_tracer()
     vulnerability_reports = tracer.vulnerability_reports if tracer else []
-    write_sarif_report(output_path, vulnerability_reports, tool_version=get_version())
+    try:
+        write_sarif_report(output_path, vulnerability_reports, tool_version=get_version())
+    except OSError as error:
+        Console().print(f"[yellow]Failed to write SARIF:[/] {error}")
+        return None
     return output_path
 
 

strix/telemetry/sarif.py

@@ -1,3 +1,5 @@
+"""Build GitHub-compatible SARIF output from Strix vulnerability reports."""
+
 from __future__ import annotations
 
 import json
@@ -20,6 +22,7 @@ def build_sarif_report(
     *,
     tool_version: str | None = None,
 ) -> dict[str, Any]:
+    """Return a SARIF 2.1.0 document for findings with safe source locations."""
     rules_by_id: dict[str, dict[str, Any]] = {}
     results: list[dict[str, Any]] = []
     locationless_findings: list[dict[str, Any]] = []
@@ -27,13 +30,13 @@ def build_sarif_report(
 
     for report in vulnerability_reports:
         locations, dropped_location_count = _build_locations(report.get("code_locations"))
+        if not locations:
+            locationless_findings.append(_locationless_summary(report))
+            continue
         if dropped_location_count:
             dropped_unsafe_location_findings.append(
                 _dropped_location_summary(report, dropped_location_count)
             )
-        if not locations:
-            locationless_findings.append(_locationless_summary(report))
-            continue
 
         rule_id = _rule_id(report)
         rules_by_id.setdefault(rule_id, _build_rule(rule_id, report))
@@ -76,6 +79,7 @@ def write_sarif_report(
     *,
     tool_version: str | None = None,
 ) -> None:
+    """Write a SARIF report to disk, creating parent directories first."""
     output_path.parent.mkdir(parents=True, exist_ok=True)
     sarif = build_sarif_report(vulnerability_reports, tool_version=tool_version)
     with output_path.open("w", encoding="utf-8") as sarif_file:
@@ -84,6 +88,7 @@ def write_sarif_report(
 
 
 def _build_rule(rule_id: str, report: dict[str, Any]) -> dict[str, Any]:
+    """Build a SARIF rule descriptor from a Strix finding."""
     title = _string_value(report.get("title")) or rule_id
     full_description = _string_value(report.get("description")) or title
     rule: dict[str, Any] = {
@@ -107,6 +112,7 @@ def _build_result(
     report: dict[str, Any],
     locations: list[dict[str, Any]],
 ) -> dict[str, Any]:
+    """Build one SARIF result using validated physical locations."""
     title = _string_value(report.get("title")) or rule_id
     return {
         "ruleId": rule_id,
@@ -118,6 +124,7 @@ def _build_result(
 
 
 def _result_properties(report: dict[str, Any]) -> dict[str, Any]:
+    """Return non-empty Strix finding metadata for SARIF properties."""
     properties: dict[str, Any] = {}
     for key in (
         "id",
@@ -138,6 +145,7 @@ def _result_properties(report: dict[str, Any]) -> dict[str, Any]:
 
 
 def _build_locations(raw_locations: Any) -> tuple[list[dict[str, Any]], int]:
+    """Return SARIF locations and a count of dropped unsafe locations."""
     if not isinstance(raw_locations, list):
         return [], 0
 
@@ -162,10 +170,8 @@ def _build_locations(raw_locations: Any) -> tuple[list[dict[str, Any]], int]:
             continue
 
         region: dict[str, Any] = {"startLine": start_line}
-        if type(end_line) is not int or end_line < start_line:
-            dropped_location_count += 1
-            continue
-        region["endLine"] = end_line
+        if type(end_line) is int and end_line >= start_line:
+            region["endLine"] = end_line
 
         snippet = _string_value(location.get("snippet"))
         if snippet:
@@ -184,6 +190,7 @@ def _build_locations(raw_locations: Any) -> tuple[list[dict[str, Any]], int]:
 
 
 def _rule_id(report: dict[str, Any]) -> str:
+    """Choose a stable SARIF rule id from CWE, CVE, id, or title."""
     for key in ("cwe", "cve", "id"):
         value = _string_value(report.get(key))
         if value:
@@ -194,6 +201,7 @@ def _rule_id(report: dict[str, Any]) -> str:
 
 
 def _sarif_level(severity: Any) -> str:
+    """Map Strix severity labels to SARIF result levels."""
     normalized = (_string_value(severity) or "").lower()
     if normalized in {"critical", "high"}:
         return "error"
@@ -203,6 +211,7 @@ def _sarif_level(severity: Any) -> str:
 
 
 def _sarif_uri(file_path: str) -> str | None:
+    """Return a safe repo-relative SARIF URI, or None for unsafe paths."""
     uri = PurePosixPath(file_path.replace("\\", "/")).as_posix()
     parts = PurePosixPath(uri).parts
     if not uri or uri.startswith("/") or not parts:
@@ -213,6 +222,7 @@ def _sarif_uri(file_path: str) -> str | None:
 
 
 def _string_value(value: Any) -> str | None:
+    """Return a stripped non-empty string value, or None."""
     if isinstance(value, str):
         stripped = value.strip()
         return stripped or None
@@ -220,12 +230,14 @@ def _string_value(value: Any) -> str | None:
 
 
 def _slugify(value: str) -> str:
+    """Convert arbitrary finding text into a stable lowercase slug."""
     chars = [char.lower() if char.isalnum() else "-" for char in value]
     slug = "-".join(part for part in "".join(chars).split("-") if part)
     return slug or "strix-finding"
 
 
 def _help_text(report: dict[str, Any], fallback: str) -> str:
+    """Assemble SARIF help text from finding details and remediation."""
     sections = [
         _string_value(report.get("description")),
         _string_value(report.get("impact")),
@@ -236,6 +248,7 @@ def _help_text(report: dict[str, Any], fallback: str) -> str:
 
 
 def _locationless_summary(report: dict[str, Any]) -> dict[str, Any]:
+    """Summarize findings that cannot be emitted as code-scanning alerts."""
     summary: dict[str, Any] = {}
     for key in ("id", "title", "severity", "cwe", "cve", "target", "endpoint", "method"):
         value = report.get(key)
@@ -248,6 +261,7 @@ def _dropped_location_summary(
     report: dict[str, Any],
     dropped_location_count: int,
 ) -> dict[str, Any]:
+    """Summarize unsafe locations dropped from a partially emitted finding."""
     summary: dict[str, Any] = {"droppedLocationCount": dropped_location_count}
     for key in ("id", "title"):
         value = report.get(key)

tests/interface/test_sarif_cli.py

@@ -58,3 +58,21 @@ def test_write_requested_sarif_output_writes_before_non_interactive_exit(
 
     assert written_path == output_path
     assert output_path.exists()
+
+
+def test_write_requested_sarif_output_reports_write_errors(
+    monkeypatch,
+    capsys,
+    tmp_path: Path,
+) -> None:
+    args = Namespace(sarif=True, sarif_output=str(tmp_path / "results.sarif"))
+
+    def raise_write_error(*_args: Any, **_kwargs: Any) -> None:
+        raise OSError("disk full")
+
+    monkeypatch.setattr(main_module, "write_sarif_report", raise_write_error)
+
+    written_path = write_requested_sarif_output(args, tmp_path / "strix_runs" / "demo")
+
+    assert written_path is None
+    assert "Failed to write SARIF" in capsys.readouterr().out

tests/telemetry/test_sarif.py

@@ -102,7 +102,6 @@ def test_build_sarif_drops_unsafe_code_locations() -> None:
                     {"file": "foo:bar.py", "start_line": 5, "end_line": 5},
                     {"file": "src/app.py", "start_line": 0, "end_line": 1},
                     {"file": "src/other.py", "start_line": True, "end_line": True},
-                    {"file": "src/reversed.py", "start_line": 5, "end_line": 4},
                 ]
             )
         ]
@@ -111,6 +110,27 @@ def test_build_sarif_drops_unsafe_code_locations() -> None:
     run = sarif["runs"][0]
     assert run["results"] == []
     assert run["properties"]["locationlessFindingCount"] == 1
+    assert "droppedUnsafeLocationCount" not in run["properties"]
+    assert "droppedUnsafeLocationFindings" not in run["properties"]
+
+
+def test_build_sarif_keeps_locations_without_valid_end_line() -> None:
+    sarif = build_sarif_report(
+        [
+            _finding(
+                code_locations=[
+                    {"file": "src/app.py", "start_line": 10},
+                    {"file": "src/reversed.py", "start_line": 20, "end_line": 19},
+                ]
+            )
+        ]
+    )
+
+    regions = [
+        location["physicalLocation"]["region"]
+        for location in sarif["runs"][0]["results"][0]["locations"]
+    ]
+    assert regions == [{"startLine": 10}, {"startLine": 20}]
 
 
 def test_build_sarif_summarizes_dropped_unsafe_locations_when_safe_locations_remain() -> None: