"""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, }