423 lines
20 KiB
Python
423 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
"""Generate the deterministic K8-REL-05 post-release audit.
|
|
|
|
The audit is deliberately additive. It reads the K8-REL-04 gate and its
|
|
inputs, compares every recorded artifact hash with the current workspace,
|
|
and records drift as a blocker. It never rewrites release evidence.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
CONTEXT = Path("context-lock.json")
|
|
GATE = Path("coverage/release-candidate-gate.report.json")
|
|
AUDIT = Path("coverage/post-release-audit.report.json")
|
|
|
|
BLOCKER_METADATA: dict[str, dict[str, Any]] = {
|
|
"manifest_seed_unassigned": {
|
|
"owner": "manifest-feature-ownership",
|
|
"review_owner": "release-closure-owner",
|
|
"evidence": [
|
|
"coverage/release-evidence-index.json",
|
|
"coverage/manifest.jsonl",
|
|
],
|
|
"remediation": "assign an accountable feature owner to every needs_owner manifest seed and regenerate the evidence index",
|
|
"review_trigger": "manifest mapping changes or the release closure is regenerated",
|
|
},
|
|
"accepted_entries_missing": {
|
|
"owner": "release-acceptance-owner",
|
|
"review_owner": "release-gate-owner",
|
|
"evidence": [
|
|
"coverage/release-evidence-index.json",
|
|
"coverage/release-candidate-gate.report.json",
|
|
],
|
|
"remediation": "complete acceptance evidence for eligible entries; do not promote blocked or derived entries by relabeling",
|
|
"review_trigger": "any entry changes to accepted or the acceptance policy changes",
|
|
},
|
|
"official_oracle_missing": {
|
|
"owner": "upstream-oracle-owner",
|
|
"review_owner": "kinematics-recovery-reviewer",
|
|
"evidence": [
|
|
"coverage/compatibility-matrix.report.json",
|
|
"coverage/kinematics-oracle.report.json",
|
|
"coverage/workspace-recovery.report.json",
|
|
],
|
|
"remediation": "add the locked official kinematics and recovery oracle inputs, then rerun the native/WASM comparison",
|
|
"review_trigger": "the locked source gains or changes an official oracle root",
|
|
},
|
|
"configuration_resources_missing": {
|
|
"owner": "configuration-closure-owner",
|
|
"review_owner": "configuration-regression-reviewer",
|
|
"evidence": [
|
|
"coverage/compatibility-matrix.report.json",
|
|
"coverage/configuration-regression.report.json",
|
|
],
|
|
"remediation": "resolve missing and unindexed configuration resources and unregistered startup references without dropping rows",
|
|
"review_trigger": "configuration graph or startup preflight counts change",
|
|
},
|
|
"browser_matrix_unverified": {
|
|
"owner": "browser-compatibility-owner",
|
|
"review_owner": "release-gate-owner",
|
|
"evidence": [
|
|
"coverage/compatibility-matrix.report.json",
|
|
"context-lock.json",
|
|
],
|
|
"remediation": "pin minimum Chromium, Firefox and Safari versions in the locked browser matrix and execute the required checks",
|
|
"review_trigger": "browser minimums or cross-origin isolation policy changes",
|
|
},
|
|
"license_closure_incomplete": {
|
|
"owner": "implementation-license-owner",
|
|
"review_owner": "license-audit-reviewer",
|
|
"evidence": [
|
|
"coverage/license-closure.report.json",
|
|
"coverage/license-source-index.jsonl",
|
|
"docs/linuxcnc-web-wasm-license-decision.md",
|
|
],
|
|
"remediation": "approve SPDX/NOTICE metadata for all 57 Web/WASM implementation entries; upstream declarations do not substitute for it",
|
|
"review_trigger": "implementation license metadata or source-root ownership changes",
|
|
},
|
|
}
|
|
|
|
REQUIRED_CONTEXT_REPORTS = (
|
|
GATE,
|
|
Path("coverage/release-closure.report.json"),
|
|
Path("coverage/release-evidence-index.json"),
|
|
Path("coverage/compatibility-matrix.report.json"),
|
|
Path("coverage/license-closure.report.json"),
|
|
Path("coverage/workspace-recovery.report.json"),
|
|
Path("coverage/workspace-migration.report.json"),
|
|
)
|
|
|
|
|
|
def canonical(value: Any) -> str:
|
|
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
|
|
|
|
|
def digest(value: Any) -> str:
|
|
return hashlib.sha256(canonical(value).encode("utf-8")).hexdigest()
|
|
|
|
|
|
def file_digest(path: Path) -> str | None:
|
|
if not path.is_file():
|
|
return None
|
|
handle = hashlib.sha256()
|
|
with path.open("rb") as stream:
|
|
for block in iter(lambda: stream.read(1024 * 1024), b""):
|
|
handle.update(block)
|
|
return handle.hexdigest()
|
|
|
|
|
|
def read_json(root: Path, path: Path) -> dict[str, Any]:
|
|
value = json.loads((root / path).read_text(encoding="utf-8"))
|
|
if not isinstance(value, dict):
|
|
raise ValueError(f"{path} must contain an object")
|
|
return value
|
|
|
|
|
|
def try_read_json(root: Path, path: Path) -> tuple[dict[str, Any], str | None]:
|
|
try:
|
|
return read_json(root, path), None
|
|
except (OSError, UnicodeError, json.JSONDecodeError, ValueError) as exc:
|
|
return {}, str(exc)
|
|
|
|
|
|
def artifact(root: Path, path: Path, role: str) -> dict[str, Any]:
|
|
full = root / path
|
|
actual = file_digest(full)
|
|
if actual is None:
|
|
return {"path": path.as_posix(), "role": role, "status": "missing"}
|
|
return {
|
|
"path": path.as_posix(),
|
|
"role": role,
|
|
"status": "present",
|
|
"bytes": full.stat().st_size,
|
|
"sha256": actual,
|
|
}
|
|
|
|
|
|
def report_hash_check(path: Path, report: dict[str, Any], error: str | None) -> dict[str, Any]:
|
|
declared = report.get("report_sha256")
|
|
if error is None and declared is None and report:
|
|
return {
|
|
"path": path.as_posix(),
|
|
"declared": None,
|
|
"actual": None,
|
|
"status": "verified",
|
|
"hash_policy": "not_declared_by_legacy_schema",
|
|
}
|
|
actual = digest({key: value for key, value in report.items() if key != "report_sha256"}) if report else None
|
|
status = "verified" if error is None and declared and declared == actual else "blocked"
|
|
return {
|
|
"path": path.as_posix(),
|
|
"declared": declared,
|
|
"actual": actual,
|
|
"status": status,
|
|
**({"error": error} if error else {}),
|
|
}
|
|
|
|
|
|
def compare_gate_inputs(root: Path, gate: dict[str, Any]) -> tuple[list[dict[str, Any]], bool]:
|
|
if not gate or not isinstance(gate.get("inputs"), list):
|
|
return [], False
|
|
comparisons: list[dict[str, Any]] = []
|
|
all_match = True
|
|
for expected in gate.get("inputs", []):
|
|
path_text = expected.get("path")
|
|
if not isinstance(path_text, str):
|
|
comparisons.append({"path": None, "status": "blocked", "reason": "gate input has no path"})
|
|
all_match = False
|
|
continue
|
|
path = Path(path_text)
|
|
full = root / path
|
|
actual_sha = file_digest(full)
|
|
actual_bytes = full.stat().st_size if full.is_file() else None
|
|
matches = (
|
|
actual_sha is not None
|
|
and actual_sha == expected.get("sha256")
|
|
and actual_bytes == expected.get("bytes")
|
|
)
|
|
comparison = {
|
|
"path": path.as_posix(),
|
|
"role": expected.get("role"),
|
|
"expected_sha256": expected.get("sha256"),
|
|
"actual_sha256": actual_sha,
|
|
"expected_bytes": expected.get("bytes"),
|
|
"actual_bytes": actual_bytes,
|
|
"status": "verified" if matches else "blocked",
|
|
}
|
|
comparisons.append(comparison)
|
|
all_match = all_match and matches
|
|
return comparisons, all_match
|
|
|
|
|
|
def context_check(root: Path, lock: dict[str, Any], reports: dict[Path, dict[str, Any]]) -> dict[str, Any]:
|
|
context_id = lock.get("context_id")
|
|
observations = []
|
|
all_match = bool(context_id)
|
|
for path in REQUIRED_CONTEXT_REPORTS:
|
|
report = reports.get(path, {})
|
|
observed = report.get("context_id")
|
|
declared = observed is not None
|
|
matches = not declared or observed == context_id
|
|
observations.append({
|
|
"path": path.as_posix(),
|
|
"context_id": observed,
|
|
"context_declared": declared,
|
|
"status": "verified" if matches else "blocked",
|
|
})
|
|
all_match = all_match and matches
|
|
return {
|
|
"context_id": context_id,
|
|
"reports": observations,
|
|
"status": "verified" if all_match else "blocked",
|
|
"all_declared_context_ids_match": all_match,
|
|
"absent_context_id_policy": "allowed only for legacy reports covered by the K8-REL-04 gate",
|
|
}
|
|
|
|
|
|
def cross_reference_check(
|
|
root: Path,
|
|
lock: dict[str, Any],
|
|
gate: dict[str, Any],
|
|
closure: dict[str, Any],
|
|
index: dict[str, Any],
|
|
compatibility: dict[str, Any],
|
|
license_report: dict[str, Any],
|
|
migration: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
checks: list[dict[str, Any]] = []
|
|
|
|
def check(name: str, expected: Any, actual: Any, evidence: str) -> None:
|
|
checks.append({
|
|
"name": name,
|
|
"expected": expected,
|
|
"actual": actual,
|
|
"evidence": evidence,
|
|
"status": "verified" if expected == actual else "blocked",
|
|
})
|
|
|
|
check("gate_context_id", lock.get("context_id"), gate.get("context_id"), GATE.as_posix())
|
|
check("closure_context_id", lock.get("context_id"), closure.get("context_id"), "coverage/release-closure.report.json")
|
|
check("evidence_index_context_id", lock.get("context_id"), index.get("context_id"), "coverage/release-evidence-index.json")
|
|
check("compatibility_context_id", lock.get("context_id"), compatibility.get("context_id"), "coverage/compatibility-matrix.report.json")
|
|
check("license_context_id", lock.get("context_id"), license_report.get("context_id"), "coverage/license-closure.report.json")
|
|
check("migration_context_id", lock.get("context_id"), migration.get("context_id"), "coverage/workspace-migration.report.json")
|
|
check("context_lock_hash", file_digest(root / CONTEXT), index.get("context_lock", {}).get("sha256"), "coverage/release-evidence-index.json")
|
|
check("closure_compatibility_report_hash", compatibility.get("report_sha256"), closure.get("compatibility", {}).get("sha256"), "coverage/release-closure.report.json")
|
|
check("closure_license_report_hash", license_report.get("report_sha256"), closure.get("license", {}).get("sha256"), "coverage/release-closure.report.json")
|
|
check("closure_evidence_index_report_hash", index.get("report_sha256"), closure.get("evidence_index", {}).get("sha256"), "coverage/release-closure.report.json")
|
|
check("index_compatibility_report_hash", compatibility.get("report_sha256"), index.get("compatibility_report", {}).get("sha256"), "coverage/release-evidence-index.json")
|
|
check("index_license_report_hash", license_report.get("report_sha256"), index.get("license_report", {}).get("sha256"), "coverage/release-evidence-index.json")
|
|
check("index_migration_report_hash", file_digest(root / Path("coverage/workspace-migration.report.json")), index.get("migration_report", {}).get("sha256"), "coverage/release-evidence-index.json")
|
|
check("recovery_hash_lock", lock.get("workspace_recovery_report_sha256"), file_digest(root / Path("coverage/workspace-recovery.report.json")), GATE.as_posix())
|
|
check("gate_release_action", "STOP_RELEASE", gate.get("release_action"), GATE.as_posix())
|
|
check("gate_candidate_eligible", False, gate.get("candidate_eligible"), GATE.as_posix())
|
|
check("gate_release_decision", "blocked_not_release_candidate", gate.get("release_decision"), GATE.as_posix())
|
|
check("gate_blocked_count", 6, gate.get("blocked_count"), GATE.as_posix())
|
|
status = "verified" if all(item["status"] == "verified" for item in checks) else "blocked"
|
|
return {"status": status, "checks": checks}
|
|
|
|
|
|
def blocker_review(root: Path, gate: dict[str, Any]) -> tuple[list[dict[str, Any]], dict[str, Any]]:
|
|
blockers = []
|
|
all_locatable = True
|
|
gate_blockers = gate.get("blockers", [])
|
|
for blocker in gate_blockers:
|
|
blocker_id = blocker.get("id")
|
|
metadata = BLOCKER_METADATA.get(blocker_id, {})
|
|
evidence = metadata.get("evidence", [])
|
|
evidence_status = all((root / Path(path)).is_file() for path in evidence)
|
|
row = {
|
|
"id": blocker_id,
|
|
"count": blocker.get("count"),
|
|
"status": blocker.get("status"),
|
|
"reason": blocker.get("reason"),
|
|
"owner": metadata.get("owner"),
|
|
"review_owner": metadata.get("review_owner"),
|
|
"evidence": [
|
|
{"path": path, "status": "present" if (root / Path(path)).is_file() else "missing"}
|
|
for path in evidence
|
|
],
|
|
"remediation": metadata.get("remediation"),
|
|
"review_trigger": metadata.get("review_trigger"),
|
|
"status_check": "verified" if blocker.get("status") == "blocked" else "blocked",
|
|
"evidence_check": "verified" if evidence_status else "blocked",
|
|
}
|
|
blockers.append(row)
|
|
all_locatable = all_locatable and bool(blocker_id) and blocker.get("status") == "blocked" and bool(metadata) and evidence_status
|
|
expected_ids = set(BLOCKER_METADATA)
|
|
actual_ids = {row.get("id") for row in gate_blockers}
|
|
identity_preserved = len(gate_blockers) == 6 and actual_ids == expected_ids
|
|
return blockers, {
|
|
"status": "verified" if all_locatable and identity_preserved else "blocked",
|
|
"all_blockers_locatable": all_locatable,
|
|
"identity_preserved": identity_preserved,
|
|
"expected_count": 6,
|
|
"observed_count": len(gate_blockers),
|
|
"observed_ids": sorted(str(value) for value in actual_ids),
|
|
}
|
|
|
|
|
|
def generate(root: Path, output: Path = AUDIT) -> dict[str, Any]:
|
|
root = root.resolve()
|
|
gate, gate_error = try_read_json(root, GATE)
|
|
lock, lock_error = try_read_json(root, CONTEXT)
|
|
reports: dict[Path, dict[str, Any]] = {}
|
|
report_errors: dict[Path, str | None] = {}
|
|
for path in REQUIRED_CONTEXT_REPORTS:
|
|
reports[path], report_errors[path] = try_read_json(root, path)
|
|
closure = reports.get(Path("coverage/release-closure.report.json"), {})
|
|
index = reports.get(Path("coverage/release-evidence-index.json"), {})
|
|
compatibility = reports.get(Path("coverage/compatibility-matrix.report.json"), {})
|
|
license_report = reports.get(Path("coverage/license-closure.report.json"), {})
|
|
migration = reports.get(Path("coverage/workspace-migration.report.json"), {})
|
|
|
|
input_comparisons, inputs_match = compare_gate_inputs(root, gate)
|
|
context = context_check(root, lock, reports)
|
|
cross_refs = cross_reference_check(root, lock, gate, closure, index, compatibility, license_report, migration)
|
|
blocker_rows, blocker_checks = blocker_review(root, gate)
|
|
report_hashes = [
|
|
report_hash_check(path, reports[path], report_errors[path])
|
|
for path in REQUIRED_CONTEXT_REPORTS
|
|
] + [report_hash_check(GATE, gate, gate_error)]
|
|
report_hashes_ok = all(item["status"] == "verified" for item in report_hashes)
|
|
|
|
gate_checks = gate.get("checks", {}) if gate else {}
|
|
gate_check_integrity = {
|
|
"context_consistency": gate_checks.get("context_consistency", {}).get("status") == "verified",
|
|
"locked_schema_contracts": gate_checks.get("locked_schema_contracts", {}).get("status") == "verified",
|
|
"evidence_integrity": gate_checks.get("evidence_integrity", {}).get("status") == "verified",
|
|
"license_source_index": gate_checks.get("license_source_index", {}).get("status") == "verified",
|
|
"migration_rollback_offline": gate_checks.get("migration_rollback_offline", {}).get("status") == "verified",
|
|
"release_prerequisites_blocked": gate_checks.get("release_prerequisites", {}).get("status") == "blocked",
|
|
}
|
|
gate_check_status = "verified" if all(gate_check_integrity.values()) else "blocked"
|
|
generated_at = lock.get("generated_at") if lock else None
|
|
checks = {
|
|
"input_hashes": {
|
|
"status": "verified" if inputs_match else "blocked",
|
|
"all_match": inputs_match,
|
|
"comparisons": input_comparisons,
|
|
},
|
|
"report_self_hashes": {
|
|
"status": "verified" if report_hashes_ok else "blocked",
|
|
"reports": report_hashes,
|
|
},
|
|
"context_consistency": context,
|
|
"cross_references": cross_refs,
|
|
"gate_checks": {
|
|
"status": gate_check_status,
|
|
"checks": gate_check_integrity,
|
|
},
|
|
"blocker_review": blocker_checks,
|
|
}
|
|
check_statuses = [value.get("status") for value in checks.values()]
|
|
audit_checks_pass = not any(value != "verified" for value in check_statuses)
|
|
gate_stop = gate.get("release_action") == "STOP_RELEASE" or gate.get("candidate_eligible") is False
|
|
status = "blocked" if not audit_checks_pass or gate_stop else "passed"
|
|
report: dict[str, Any] = {
|
|
"schema": "post-release-audit.v1",
|
|
"context_id": lock.get("context_id"),
|
|
"audit_id": "K8-REL-05",
|
|
"generated_at": generated_at,
|
|
"status": status,
|
|
"release_decision": "blocked_not_release_candidate",
|
|
"release_action": "STOP_RELEASE",
|
|
"candidate_eligible": False,
|
|
"source_baseline": {
|
|
"linuxcnc_version": lock.get("linuxcnc_version"),
|
|
"linuxcnc_source_sha": lock.get("linuxcnc_source_sha"),
|
|
"context_lock": artifact(root, CONTEXT, "context_lock"),
|
|
},
|
|
"inputs": [artifact(root, GATE, "release_candidate_gate"), artifact(root, CONTEXT, "context_lock")],
|
|
"input_hash_comparison": input_comparisons,
|
|
"checks": checks,
|
|
"blockers": blocker_rows,
|
|
"blocked_count": len(blocker_rows),
|
|
"rebuild": {
|
|
"commands": [
|
|
"python3 scripts/generate_post_release_audit.py",
|
|
"python3 scripts/generate_release_closure.py",
|
|
"node web/worker/test_replay.mjs",
|
|
"node web/workspace/test_recovery.mjs",
|
|
"node web/workspace/test_migration.mjs",
|
|
"python3 -m unittest discover -s scripts -p 'test_*.py' -v",
|
|
],
|
|
"deterministic_inputs": [path.as_posix() for path in REQUIRED_CONTEXT_REPORTS]
|
|
+ [GATE.as_posix(), CONTEXT.as_posix(), "coverage/license-source-index.jsonl", "coverage/workspace-rollback.package.json", "coverage/offline-cache.package.json", "docs/linuxcnc-web-wasm-license-decision.md"],
|
|
},
|
|
"failure_policy": {
|
|
"on_missing_input": "STOP_RELEASE",
|
|
"on_hash_mismatch": "STOP_RELEASE",
|
|
"on_context_drift": "STOP_RELEASE",
|
|
"forbidden": ["delete_samples", "lower_checks", "reclassify_as_REAL-EXCLUDED", "promote_blocked_to_accepted"],
|
|
},
|
|
"errors": {path.as_posix(): error for path, error in [(GATE, gate_error), (CONTEXT, lock_error), *report_errors.items()] if error},
|
|
"report_sha256": None,
|
|
}
|
|
report["report_sha256"] = digest({key: value for key, value in report.items() if key != "report_sha256"})
|
|
destination = output if output.is_absolute() else root / output
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
destination.write_text(json.dumps(report, ensure_ascii=False, sort_keys=True, indent=2) + "\n", encoding="utf-8")
|
|
return report
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1])
|
|
parser.add_argument("--output", type=Path, default=AUDIT)
|
|
args = parser.parse_args()
|
|
report = generate(args.repo_root, args.output)
|
|
print("post_release_audit=" + report["status"] + " action=" + report["release_action"] + " blockers=" + str(report["blocked_count"]))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|