84746e060b
Backlog #7. Repro showed chromium finding LOCATIONS (rule/uri/line/ col/id) were byte-identical across runs while the summary finding_fingerprint drifted (064e60c9 -> 4573bb71): summarize_sarif_findings hashed the full canonical finding record, which carries Z3-timing- volatile fields (quality metrics, uncertainty reasons, confidence, score). The stability gate already used identity-only keys, so the fingerprint contradicted it on Z3-heavy corpora. finding_identity_key now = {finding_id, rule_id, uri, line, backend} (measurement artifacts excluded), and the fingerprint is the sha256 of the sorted identity keys — consistent with how change-time drift and dispositions already define 'a finding'. Verified: chromium fingerprint STABLE across identical runs (f5629f0f). Repinned the four baselines to the identity-based fingerprint (fold 6fe221a0, yams c9ad7d60, sage 3da6ecfa, chromium f5629f0f).
180 lines
7.1 KiB
Python
180 lines
7.1 KiB
Python
"""Canonical SARIF observation extraction for benchmark identity and drift checks."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from collections import Counter
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
def _integer(value: Any) -> int:
|
|
return value if isinstance(value, int) and not isinstance(value, bool) else 0
|
|
|
|
|
|
def _number(value: Any) -> float:
|
|
return (
|
|
value * 1.0
|
|
if isinstance(value, (int, float)) and not isinstance(value, bool)
|
|
else 0.0
|
|
)
|
|
|
|
|
|
def sarif_results(report: dict[str, Any]) -> list[dict[str, Any]]:
|
|
runs = report.get("runs", [])
|
|
if not isinstance(runs, list):
|
|
return []
|
|
findings: list[dict[str, Any]] = []
|
|
for run in runs:
|
|
if not isinstance(run, dict):
|
|
continue
|
|
results = run.get("results", [])
|
|
if isinstance(results, list):
|
|
findings.extend(item for item in results if isinstance(item, dict))
|
|
return findings
|
|
|
|
|
|
def normalize_uri(uri: str, project_root: Path, artifact_base: Path) -> str:
|
|
path = Path(uri.removeprefix("file://"))
|
|
# Resolve all three paths. On macOS /Users is commonly a symlink into
|
|
# /private, so resolving only the finding URI makes a project-relative
|
|
# location appear external and breaks reviewed-disposition matching.
|
|
root = project_root.resolve()
|
|
base = artifact_base.resolve()
|
|
resolved = path.resolve() if path.is_absolute() else (base / path).resolve()
|
|
try:
|
|
return resolved.relative_to(root).as_posix()
|
|
except ValueError:
|
|
return resolved.as_posix()
|
|
|
|
|
|
def _message(finding: dict[str, Any]) -> str:
|
|
value = finding.get("message", {})
|
|
return str(value.get("text", "")) if isinstance(value, dict) else str(value)
|
|
|
|
|
|
def canonical_findings(
|
|
report: dict[str, Any], project_root: Path, artifact_base: Path
|
|
) -> tuple[list[dict[str, Any]], int, int]:
|
|
canonical: list[dict[str, Any]] = []
|
|
missing_location = 0
|
|
missing_backend = 0
|
|
for finding in sarif_results(report):
|
|
properties = finding.get("properties", {})
|
|
properties = properties if isinstance(properties, dict) else {}
|
|
backend = str(properties.get("backend", ""))
|
|
if not backend:
|
|
missing_backend += 1
|
|
|
|
uri = ""
|
|
line = 0
|
|
function = ""
|
|
locations = finding.get("locations", [])
|
|
if isinstance(locations, list) and locations and isinstance(locations[0], dict):
|
|
first = locations[0]
|
|
physical = first.get("physicalLocation", {})
|
|
if isinstance(physical, dict):
|
|
artifact = physical.get("artifactLocation", {})
|
|
region = physical.get("region", {})
|
|
if isinstance(artifact, dict):
|
|
uri = str(artifact.get("uri", ""))
|
|
if isinstance(region, dict):
|
|
line = _integer(region.get("startLine", 0))
|
|
logical = first.get("logicalLocations", [])
|
|
if isinstance(logical, list) and logical and isinstance(logical[0], dict):
|
|
function = str(logical[0].get("fullyQualifiedName", ""))
|
|
if not uri or line <= 0:
|
|
missing_location += 1
|
|
|
|
quality = properties.get("qualityEvidence", {})
|
|
quality = quality if isinstance(quality, dict) else {}
|
|
raw_metrics = quality.get("metrics", [])
|
|
metrics: dict[str, float] = {}
|
|
if isinstance(raw_metrics, list):
|
|
for metric in raw_metrics:
|
|
if not isinstance(metric, dict) or not isinstance(
|
|
metric.get("name"), str
|
|
):
|
|
continue
|
|
value = metric.get("value", 0)
|
|
if isinstance(value, (int, float)) and not isinstance(value, bool):
|
|
metrics[metric["name"]] = _number(value)
|
|
raw_members = quality.get("member_ids", [])
|
|
member_ids = (
|
|
[str(member) for member in raw_members]
|
|
if isinstance(raw_members, list)
|
|
else []
|
|
)
|
|
|
|
canonical.append(
|
|
{
|
|
"finding_id": str(properties.get("id", "")),
|
|
"rule_id": str(finding.get("ruleId", "")),
|
|
"backend": backend,
|
|
"confidence": str(properties.get("confidence", "")),
|
|
"score": _number(properties.get("score", 0)),
|
|
"summary_backed": bool(properties.get("summaryBacked", False)),
|
|
"cross_tu_corroboration": bool(
|
|
properties.get("crossTuCorroboration", False)
|
|
),
|
|
"quality_classification": str(quality.get("classification", "")),
|
|
"quality_representative_id": str(quality.get("representative_id", "")),
|
|
"quality_member_ids": member_ids,
|
|
"quality_metrics": metrics,
|
|
"quality_incomplete": bool(quality.get("incomplete", False)),
|
|
"quality_uncertainty_reasons": [
|
|
str(reason) for reason in quality.get("uncertainty_reasons", [])
|
|
]
|
|
if isinstance(quality.get("uncertainty_reasons", []), list)
|
|
else [],
|
|
"uri": normalize_uri(uri, project_root, artifact_base) if uri else "",
|
|
"line": line,
|
|
"function": function,
|
|
"message": _message(finding),
|
|
}
|
|
)
|
|
canonical.sort(key=lambda item: json.dumps(item, sort_keys=True))
|
|
return canonical, missing_location, missing_backend
|
|
|
|
|
|
def finding_key(finding: dict[str, Any]) -> str:
|
|
return json.dumps(finding, sort_keys=True, separators=(",", ":"))
|
|
|
|
|
|
def finding_identity_key(finding: dict[str, Any]) -> str:
|
|
# Identity = the location/rule/backend + the detector's own finding id.
|
|
# Excludes measurement artifacts (message, confidence, score, quality
|
|
# metrics, uncertainty reasons, corroboration flags) that can vary with
|
|
# solver timing even when the finding LOCATION is unchanged — otherwise
|
|
# the fingerprint drifts on Z3-heavy corpora (chromium) and fake
|
|
# churn/spurious 'new findings' fire. See backlog #7.
|
|
identity = {
|
|
key: finding.get(key)
|
|
for key in ("finding_id", "rule_id", "uri", "line", "backend")
|
|
}
|
|
return finding_key(identity)
|
|
|
|
|
|
def summarize_sarif_findings(
|
|
report: dict[str, Any], project_root: Path, artifact_base: Path
|
|
) -> dict[str, Any]:
|
|
canonical, missing_location, missing_backend = canonical_findings(
|
|
report, project_root, artifact_base
|
|
)
|
|
keys = [finding_identity_key(finding) for finding in canonical]
|
|
fingerprint = hashlib.sha256(
|
|
json.dumps(sorted(keys), separators=(",", ":")).encode()
|
|
).hexdigest()
|
|
counts = Counter(str(finding["rule_id"]) for finding in canonical)
|
|
return {
|
|
"count": len(canonical),
|
|
"by_rule": dict(sorted(counts.items())),
|
|
"missing_location": missing_location,
|
|
"missing_backend": missing_backend,
|
|
"fingerprint": fingerprint,
|
|
"records": canonical,
|
|
# Kept in the runner's in-memory observation, not emitted in the public
|
|
# summary. It enables meaningful partial-drift metrics beyond one hash.
|
|
"identity_keys": keys,
|
|
}
|