2dcbdbc75d
Preserve source, binary, and runtime evidence boundaries while making precision and performance regressions reproducible. - decompose analysis orchestration into bounded modules - add differential, metamorphic, organic, and Yams change-time gates - fix detector, P-Code, fuzz sanitizer, and report identity precision
172 lines
7 KiB
Python
172 lines
7 KiB
Python
"""Reviewed finding dispositions and change-time delta reporting."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from collections import Counter
|
|
from datetime import datetime
|
|
from typing import Any, Mapping, Sequence
|
|
|
|
|
|
DISPOSITION_SCHEMA_VERSION = 1
|
|
DISPOSITIONS = frozenset(
|
|
{"actionable", "accepted-risk", "duplicate", "false-positive", "unsupported"}
|
|
)
|
|
|
|
|
|
def configuration_fingerprint(configuration: Mapping[str, Any]) -> str:
|
|
encoded = json.dumps(configuration, sort_keys=True, separators=(",", ":")).encode()
|
|
return hashlib.sha256(encoded).hexdigest()
|
|
|
|
|
|
def validate_disposition_baseline(document: Mapping[str, Any]) -> list[str]:
|
|
errors: list[str] = []
|
|
if document.get("schema_version") != DISPOSITION_SCHEMA_VERSION:
|
|
errors.append(f"unsupported disposition schema_version: {document.get('schema_version')}")
|
|
for field in ("target_name", "reviewed_at", "reviewer"):
|
|
if not isinstance(document.get(field), str) or not str(document[field]).strip():
|
|
errors.append(f"{field} must be a non-empty string")
|
|
reviewed_at = document.get("reviewed_at")
|
|
if isinstance(reviewed_at, str):
|
|
try:
|
|
datetime.fromisoformat(reviewed_at.replace("Z", "+00:00"))
|
|
except ValueError:
|
|
errors.append("reviewed_at must be an ISO-8601 timestamp")
|
|
|
|
target = document.get("target")
|
|
if not isinstance(target, dict):
|
|
errors.append("target must be an object")
|
|
else:
|
|
for field in ("revision", "tu_set_fingerprint", "configuration_fingerprint"):
|
|
if not isinstance(target.get(field), str) or not target[field]:
|
|
errors.append(f"target.{field} must be a non-empty string")
|
|
|
|
findings = document.get("findings")
|
|
if not isinstance(findings, list):
|
|
errors.append("findings must be an array")
|
|
return errors
|
|
seen: set[str] = set()
|
|
for index, finding in enumerate(findings):
|
|
if not isinstance(finding, dict):
|
|
errors.append(f"findings[{index}] must be an object")
|
|
continue
|
|
finding_id = finding.get("id")
|
|
if not isinstance(finding_id, str) or not finding_id:
|
|
errors.append(f"findings[{index}].id must be a non-empty string")
|
|
elif finding_id in seen:
|
|
errors.append(f"duplicate disposition finding id: {finding_id}")
|
|
else:
|
|
seen.add(finding_id)
|
|
if finding.get("status") not in DISPOSITIONS:
|
|
errors.append(f"findings[{index}].status is not a recognized disposition")
|
|
for field in ("rule_id", "rationale"):
|
|
if not isinstance(finding.get(field), str) or not finding[field].strip():
|
|
errors.append(f"findings[{index}].{field} must be a non-empty string")
|
|
location = finding.get("location")
|
|
if location is not None and (
|
|
not isinstance(location, dict)
|
|
or not isinstance(location.get("uri"), str)
|
|
or not isinstance(location.get("line"), int)
|
|
or isinstance(location.get("line"), bool)
|
|
or location.get("line", 0) <= 0
|
|
):
|
|
errors.append(
|
|
f"findings[{index}].location must contain a URI and positive line"
|
|
)
|
|
return errors
|
|
|
|
|
|
def compare_dispositions(
|
|
current_findings: Sequence[Mapping[str, Any]],
|
|
baseline: Mapping[str, Any],
|
|
*,
|
|
changed_files: Sequence[str] | None = None,
|
|
scope_kloc: float = 0.0,
|
|
) -> dict[str, Any]:
|
|
errors = validate_disposition_baseline(baseline)
|
|
if errors:
|
|
raise ValueError("; ".join(errors))
|
|
current: dict[str, Mapping[str, Any]] = {}
|
|
for finding in current_findings:
|
|
finding_id = str(finding.get("finding_id", ""))
|
|
if not finding_id:
|
|
raise ValueError("current finding is missing properties.id")
|
|
if finding_id in current:
|
|
raise ValueError(f"duplicate current finding id: {finding_id}")
|
|
current[finding_id] = finding
|
|
reviewed = {str(item["id"]): item for item in baseline["findings"]}
|
|
current_ids = set(current)
|
|
reviewed_ids = set(reviewed)
|
|
new_ids = sorted(current_ids - reviewed_ids)
|
|
persisting_ids = sorted(current_ids & reviewed_ids)
|
|
resolved_ids = sorted(reviewed_ids - current_ids)
|
|
changed = set(changed_files or [])
|
|
new_in_scope = [
|
|
finding_id
|
|
for finding_id in new_ids
|
|
if not changed or str(current[finding_id].get("uri", "")) in changed
|
|
]
|
|
current_in_scope = [
|
|
finding_id
|
|
for finding_id in current_ids
|
|
if not changed or str(current[finding_id].get("uri", "")) in changed
|
|
]
|
|
current_status = Counter(str(reviewed[item]["status"]) for item in persisting_ids)
|
|
resolved_status = Counter(str(reviewed[item]["status"]) for item in resolved_ids)
|
|
new_rules = Counter(str(current[item].get("rule_id", "")) for item in new_ids)
|
|
identity_mismatches = [
|
|
{
|
|
"id": finding_id,
|
|
"reviewed_rule_id": reviewed[finding_id]["rule_id"],
|
|
"current_rule_id": current[finding_id].get("rule_id", ""),
|
|
}
|
|
for finding_id in persisting_ids
|
|
if reviewed[finding_id]["rule_id"] != current[finding_id].get("rule_id", "")
|
|
]
|
|
moved = []
|
|
for finding_id in persisting_ids:
|
|
location = reviewed[finding_id].get("location")
|
|
if not isinstance(location, dict):
|
|
continue
|
|
before = {"uri": location.get("uri", ""), "line": int(location.get("line", 0))}
|
|
after = {
|
|
"uri": current[finding_id].get("uri", ""),
|
|
"line": int(current[finding_id].get("line", 0)),
|
|
}
|
|
if before != after:
|
|
moved.append({"id": finding_id, "reviewed": before, "current": after})
|
|
current_count = len(current_ids)
|
|
return {
|
|
"new": new_ids,
|
|
"persisting": persisting_ids,
|
|
"resolved": resolved_ids,
|
|
"counts": {
|
|
"current": current_count,
|
|
"reviewed": len(reviewed_ids),
|
|
"new": len(new_ids),
|
|
"persisting": len(persisting_ids),
|
|
"resolved": len(resolved_ids),
|
|
"current_in_scope": len(current_in_scope),
|
|
"new_in_scope": len(new_in_scope),
|
|
"new_in_changed_files": len(new_in_scope) if changed else 0,
|
|
},
|
|
"new_in_changed_files": new_in_scope if changed else [],
|
|
"changed_files": sorted(changed),
|
|
"changed_scope": "explicit-files" if changed else "selected-and-finding-files",
|
|
"by_disposition": dict(sorted(current_status.items())),
|
|
"resolved_by_disposition": dict(sorted(resolved_status.items())),
|
|
"new_by_rule": dict(sorted(new_rules.items())),
|
|
"identity_mismatches": identity_mismatches,
|
|
"moved": moved,
|
|
"review_coverage": round(len(persisting_ids) / current_count, 6)
|
|
if current_count
|
|
else 1.0,
|
|
"scope_kloc": round(scope_kloc, 3),
|
|
"alert_density_per_kloc": round(len(current_in_scope) / scope_kloc, 3)
|
|
if scope_kloc > 0
|
|
else None,
|
|
"new_alert_density_per_kloc": round(len(new_in_scope) / scope_kloc, 3)
|
|
if scope_kloc > 0
|
|
else None,
|
|
}
|