K8: record blocked owner review and release evidence
This commit is contained in:
173
scripts/generate_implementation_license_remediation.py
Normal file
173
scripts/generate_implementation_license_remediation.py
Normal file
@@ -0,0 +1,173 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Generate deterministic K8-REL-13 implementation license remediation."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
CONTEXT = Path("context-lock.json")
|
||||
LICENSE_REPORT = Path("coverage/license-closure.report.json")
|
||||
LICENSE_INDEX = Path("coverage/license-source-index.jsonl")
|
||||
DECISION = Path("docs/linuxcnc-web-wasm-license-decision.md")
|
||||
GATE = Path("coverage/release-candidate-gate.report.json")
|
||||
INTAKE = Path("coverage/release-remediation-intake.report.json")
|
||||
OUTPUT = Path("coverage/implementation-license-remediation.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) -> tuple[dict[str, Any], str | None]:
|
||||
try:
|
||||
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, None
|
||||
except (OSError, UnicodeError, json.JSONDecodeError, ValueError) as exc:
|
||||
return {}, str(exc)
|
||||
|
||||
|
||||
def read_jsonl(root: Path, path: Path) -> tuple[list[dict[str, Any]], str | None]:
|
||||
try:
|
||||
rows = []
|
||||
with (root / path).open(encoding="utf-8") as stream:
|
||||
for line_number, line in enumerate(stream, 1):
|
||||
if not line.strip():
|
||||
continue
|
||||
value = json.loads(line)
|
||||
if not isinstance(value, dict):
|
||||
raise ValueError(f"{path}:{line_number} must contain an object")
|
||||
rows.append(value)
|
||||
return rows, 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
|
||||
sha = file_digest(full)
|
||||
if sha 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": sha}
|
||||
|
||||
|
||||
def source_declaration(root: Path, relative: str) -> dict[str, Any]:
|
||||
path = root / relative
|
||||
text = path.read_text(encoding="utf-8", errors="replace") if path.is_file() else ""
|
||||
spdx = sorted(set(re.findall(r"SPDX-License-Identifier:\s*([^\s*]+)", text)))
|
||||
notice = bool(re.search(r"\bNOTICE\b|copyright", text, re.IGNORECASE))
|
||||
return {"spdx_identifiers": spdx, "copyright_or_notice_text": notice, "status": "declared" if spdx else "unassigned"}
|
||||
|
||||
|
||||
def generate(root: Path, output: Path = OUTPUT) -> dict[str, Any]:
|
||||
root = root.resolve()
|
||||
lock, lock_error = read_json(root, CONTEXT)
|
||||
license_report, license_report_error = read_json(root, LICENSE_REPORT)
|
||||
gate, gate_error = read_json(root, GATE)
|
||||
intake, intake_error = read_json(root, INTAKE)
|
||||
license_rows, license_index_error = read_jsonl(root, LICENSE_INDEX)
|
||||
errors = {
|
||||
path.as_posix(): error
|
||||
for path, error in ((CONTEXT, lock_error), (LICENSE_REPORT, license_report_error), (GATE, gate_error), (INTAKE, intake_error), (LICENSE_INDEX, license_index_error))
|
||||
if error
|
||||
}
|
||||
implementation = [row for row in license_rows if row.get("source_root") in {"wasm-core", "web"}]
|
||||
declarations = {row.get("source_path"): source_declaration(root, str(row.get("source_path"))) for row in implementation}
|
||||
root_license_files = [path.relative_to(root).as_posix() for path in sorted(root.glob("LICENSE*")) if path.is_file()]
|
||||
root_notice_files = [path.relative_to(root).as_posix() for path in sorted(root.glob("NOTICE*")) if path.is_file()]
|
||||
source_hash_ok = all(row.get("source_hash_status") == "verified" for row in implementation)
|
||||
all_unassigned = all(row.get("license_status") == "blocked_unassigned" and row.get("license_spdx") is None for row in implementation)
|
||||
no_declarations = not root_license_files and not root_notice_files and all(not value["spdx_identifiers"] for value in declarations.values())
|
||||
stop_ok = (
|
||||
gate.get("release_action") == "STOP_RELEASE"
|
||||
and gate.get("candidate_eligible") is False
|
||||
and gate.get("blocked_count") == 6
|
||||
and intake.get("release_action") == "STOP_RELEASE"
|
||||
and intake.get("candidate_eligible") is False
|
||||
)
|
||||
checks = {
|
||||
"implementation_entry_coverage": {"status": "verified" if len(implementation) == 57 else "blocked", "entries": len(implementation)},
|
||||
"source_hashes": {"status": "verified" if source_hash_ok else "blocked", "verified": sum(row.get("source_hash_status") == "verified" for row in implementation)},
|
||||
"unassigned_status_preserved": {"status": "verified" if all_unassigned else "blocked", "blocked_unassigned": sum(row.get("license_status") == "blocked_unassigned" for row in implementation)},
|
||||
"no_unapproved_declaration": {"status": "verified" if no_declarations else "blocked", "root_license_files": root_license_files, "root_notice_files": root_notice_files, "files_with_spdx": sum(bool(value["spdx_identifiers"]) for value in declarations.values())},
|
||||
"upstream_not_inherited": {"status": "verified" if all(row.get("owner_source") == "implementation_license_decision" for row in implementation) else "blocked"},
|
||||
"context_consistency": {"status": "verified" if lock.get("context_id") and license_report.get("context_id") == lock.get("context_id") and gate.get("context_id") == lock.get("context_id") else "blocked", "context_id": lock.get("context_id")},
|
||||
"release_stop_preserved": {"status": "verified" if stop_ok else "blocked", "release_action": "STOP_RELEASE" if stop_ok else None},
|
||||
}
|
||||
checks_ok = not errors and all(value["status"] == "verified" for value in checks.values())
|
||||
rows = [
|
||||
{
|
||||
"id": row.get("id"),
|
||||
"source": row.get("source"),
|
||||
"source_path": row.get("source_path"),
|
||||
"source_sha256": row.get("source_sha256"),
|
||||
"owner": row.get("owner"),
|
||||
"review_owner": "license-audit-reviewer",
|
||||
"license_status": row.get("license_status"),
|
||||
"license_spdx": row.get("license_spdx"),
|
||||
"declaration_audit": declarations.get(row.get("source_path")),
|
||||
"remediation_status": "blocked_pending_owner_decision",
|
||||
}
|
||||
for row in sorted(implementation, key=lambda value: str(value.get("id")))
|
||||
]
|
||||
report: dict[str, Any] = {
|
||||
"schema": "implementation-license-remediation.v1",
|
||||
"context_id": lock.get("context_id"),
|
||||
"task_id": "K8-REL-13",
|
||||
"status": "complete" if checks_ok else "blocked",
|
||||
"remediation_status": "blocked_pending_owner_decision",
|
||||
"release_decision": "blocked_not_release_candidate",
|
||||
"release_action": "STOP_RELEASE",
|
||||
"candidate_eligible": False,
|
||||
"counts": {"implementation_entries": len(rows), "blocked_unassigned": sum(row["license_status"] == "blocked_unassigned" for row in rows), "approved_spdx": sum(bool(row["license_spdx"]) for row in rows), "approved_by_this_artifact": 0},
|
||||
"checks": checks,
|
||||
"inputs": [artifact(root, CONTEXT, "context_lock"), artifact(root, LICENSE_REPORT, "license_closure"), artifact(root, LICENSE_INDEX, "license_source_index"), artifact(root, DECISION, "license_decision"), artifact(root, GATE, "release_candidate_gate"), artifact(root, INTAKE, "remediation_intake")],
|
||||
"entries": rows,
|
||||
"decision_policy": {
|
||||
"required_to_unblock": ["repository owner decision", "approved SPDX expression", "copyright/NOTICE policy", "distribution obligations review", "new context if metadata changes"],
|
||||
"current_decision": "do not inherit LinuxCNC license declarations for wasm-core/web",
|
||||
"forbidden": ["infer_upstream_license", "add_unapproved_spdx", "promote_blocked_unassigned", "delete_entries", "overwrite_context_lock_in_place"],
|
||||
},
|
||||
"errors": errors,
|
||||
"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=OUTPUT)
|
||||
args = parser.parse_args()
|
||||
report = generate(args.repo_root, args.output)
|
||||
print("implementation_license_remediation=" + report["status"] + " release=" + report["release_action"] + " blocked=" + str(report["counts"]["blocked_unassigned"]))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user