154 lines
8.0 KiB
Python
154 lines
8.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Validate the external six-decision owner input without mutating release state."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
LOCK = Path("context-lock.json")
|
|
INPUT = Path("coverage/owner-decisions.json")
|
|
OUTPUT = Path("coverage/owner-decision-intake.report.json")
|
|
TEMPLATE = Path("coverage/owner-decision-intake.template.json")
|
|
BLOCKERS = ("manifest_seed_unassigned", "accepted_entries_missing", "official_oracle_missing", "configuration_resources_missing", "browser_matrix_unverified", "license_closure_incomplete")
|
|
DECISIONS = {"approve", "reject", "defer"}
|
|
|
|
|
|
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 sha256(path: Path) -> str:
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
|
|
def evidence_error(root: Path, evidence_row: Any) -> str | None:
|
|
"""Validate an evidence reference without allowing paths outside the repo."""
|
|
if not isinstance(evidence_row, dict):
|
|
return "evidence_not_object"
|
|
path_value = evidence_row.get("path")
|
|
hash_value = evidence_row.get("sha256")
|
|
if not isinstance(path_value, str) or not path_value.strip():
|
|
return "evidence_path_missing"
|
|
if "\\" in path_value or Path(path_value).is_absolute():
|
|
return "evidence_path_not_relative"
|
|
path = Path(path_value)
|
|
if ".." in path.parts:
|
|
return "evidence_path_escapes_repo"
|
|
if not isinstance(hash_value, str) or not re.fullmatch(r"[0-9a-f]{64}", hash_value):
|
|
return "evidence_hash_invalid"
|
|
resolved = (root / path).resolve()
|
|
try:
|
|
resolved.relative_to(root)
|
|
except ValueError:
|
|
return "evidence_path_escapes_repo"
|
|
if not resolved.is_file():
|
|
return "evidence_file_missing"
|
|
if sha256(resolved) != hash_value:
|
|
return "evidence_hash_mismatch"
|
|
return None
|
|
|
|
|
|
def load_json(path: Path) -> tuple[dict[str, Any], str | None]:
|
|
try:
|
|
value = json.loads(path.read_text(encoding="utf-8"))
|
|
if not isinstance(value, dict): raise ValueError("root must be object")
|
|
return value, None
|
|
except (OSError, UnicodeError, json.JSONDecodeError, ValueError) as exc:
|
|
return {}, str(exc)
|
|
|
|
|
|
def validate(root: Path, input_path: Path = INPUT, output_path: Path = OUTPUT) -> dict[str, Any]:
|
|
root = root.resolve()
|
|
lock, lock_error = load_json(root / LOCK)
|
|
value, input_error = load_json(root / input_path)
|
|
input_present = (root / input_path).is_file()
|
|
errors = [error for error in (lock_error, input_error) if error]
|
|
decisions = value.get("decisions") if isinstance(value.get("decisions"), list) else []
|
|
if input_present:
|
|
for field in ("schema", "context_id", "decision_set_id", "decisions"):
|
|
field_value = value.get(field)
|
|
invalid_field = (
|
|
not isinstance(field_value, list)
|
|
if field == "decisions"
|
|
else not isinstance(field_value, str) or not field_value.strip()
|
|
)
|
|
if invalid_field:
|
|
errors.append(f"{field}_missing")
|
|
observed = [row.get("blocker") for row in decisions if isinstance(row, dict)]
|
|
duplicate_ids = sorted({blocker for blocker in observed if observed.count(blocker) > 1 and blocker is not None})
|
|
missing_ids = sorted(set(BLOCKERS) - set(observed))
|
|
unknown_ids = sorted(set(observed) - set(BLOCKERS))
|
|
row_errors: list[dict[str, Any]] = []
|
|
for index, row in enumerate(decisions):
|
|
if not isinstance(row, dict):
|
|
row_errors.append({"index": index, "error": "decision_not_object"}); continue
|
|
if row.get("blocker") not in BLOCKERS: row_errors.append({"index": index, "error": "unknown_blocker"})
|
|
if row.get("decision") not in DECISIONS: row_errors.append({"index": index, "error": "decision_invalid"})
|
|
for field in ("owner", "reviewer", "rationale"):
|
|
if not isinstance(row.get(field), str) or not row[field].strip(): row_errors.append({"index": index, "error": f"{field}_missing"})
|
|
if not isinstance(row.get("new_context_required"), bool): row_errors.append({"index": index, "error": "new_context_required_invalid"})
|
|
evidence = row.get("evidence")
|
|
if not isinstance(evidence, list) or not evidence: row_errors.append({"index": index, "error": "evidence_missing"})
|
|
else:
|
|
for evidence_row in evidence:
|
|
evidence_issue = evidence_error(root, evidence_row)
|
|
if evidence_issue:
|
|
row_errors.append({"index": index, "error": evidence_issue})
|
|
context_ok = bool(lock.get("context_id")) and value.get("context_id") == lock.get("context_id") if input_present else False
|
|
shape_ok = input_present and not errors and value.get("schema") == "owner-decision-intake.v1" and len(decisions) == 6 and not duplicate_ids and not missing_ids and not unknown_ids and not row_errors and context_ok
|
|
all_approved = shape_ok and all(row.get("decision") == "approve" for row in decisions)
|
|
checks = {
|
|
"input_file_present": {"status": "verified" if input_present else "blocked", "path": input_path.as_posix()},
|
|
"schema_shape": {"status": "verified" if shape_ok else "blocked", "schema": value.get("schema")},
|
|
"six_blocker_identity": {"status": "verified" if not missing_ids and not unknown_ids and not duplicate_ids and len(decisions) == 6 else "blocked", "missing": missing_ids, "unknown": unknown_ids, "duplicates": duplicate_ids},
|
|
"context_consistency": {"status": "verified" if context_ok else "blocked", "locked": lock.get("context_id"), "input": value.get("context_id")},
|
|
"decision_approval": {"status": "verified" if all_approved else "blocked", "approved": sum(row.get("decision") == "approve" for row in decisions if isinstance(row, dict)), "required": 6},
|
|
"gate_refresh_allowed": {"status": "blocked", "reason": "owner intake validation never refreshes the gate; reviewed new context is required after all six approvals"},
|
|
"context_lock_unchanged": {"status": "verified", "action": "no mutation"},
|
|
}
|
|
report: dict[str, Any] = {
|
|
"schema": "owner-decision-intake-report.v1",
|
|
"context_id": lock.get("context_id"),
|
|
"task_id": "K8-REL-18",
|
|
"status": "verified" if all_approved else "blocked",
|
|
"implementation_status": "verified",
|
|
"release_status": "blocked_external_owner",
|
|
"candidate_eligible": False,
|
|
"release_action": "STOP_RELEASE",
|
|
"input": {"path": input_path.as_posix(), "status": "present" if input_present else "missing", "sha256": sha256(root / input_path) if input_present else None},
|
|
"template": {"path": TEMPLATE.as_posix(), "sha256": sha256(root / TEMPLATE) if (root / TEMPLATE).is_file() else None},
|
|
"checks": checks,
|
|
"errors": errors + row_errors,
|
|
"next_transition": "wait for owner-decisions.json or new locked context input; do not rerun browser or implementation matrices",
|
|
"report_sha256": None,
|
|
}
|
|
report["report_sha256"] = digest({key: value for key, value in report.items() if key != "report_sha256"})
|
|
destination = output_path if output_path.is_absolute() else root / output_path
|
|
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()
|
|
parser.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1])
|
|
parser.add_argument("--input", type=Path, default=INPUT)
|
|
parser.add_argument("--output", type=Path, default=OUTPUT)
|
|
args = parser.parse_args()
|
|
report = validate(args.repo_root.resolve(), args.input, args.output)
|
|
print(f"owner_decision_intake={report['status']} present={report['input']['status']} refresh={report['checks']['gate_refresh_allowed']['status']}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|