validate.py
python
sha256:4671b7f787ddbe63ced31c895b688c77ab495653b65a730b423329f26b3c1439
feat: K1-P1 complete — agent provenance, build-verification…
Sonnet 4.6
patch
57 days ago
| 1 | """Ledger entry validation (§K9.7 / §K9.8).""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | from typing import Any |
| 6 | |
| 7 | from tools.honesty.provenance import validate_provenance |
| 8 | from tools.honesty.types import ACTOR_ROLES, ENTRY_KINDS, EntryValidationError |
| 9 | |
| 10 | |
| 11 | def _require_mapping(value: Any, field: str) -> dict[str, Any]: |
| 12 | if not isinstance(value, dict): |
| 13 | raise EntryValidationError(2, f"{field} must be an object") |
| 14 | return value |
| 15 | |
| 16 | |
| 17 | def _require_non_empty_str(value: Any, field: str) -> str: |
| 18 | if not isinstance(value, str) or not value.strip(): |
| 19 | raise EntryValidationError(2, f"{field} must be a non-empty string") |
| 20 | return value |
| 21 | |
| 22 | |
| 23 | def validate_append_body(*, kind: str, body: dict[str, Any]) -> dict[str, Any]: |
| 24 | """Validate and normalize an append body before hashing.""" |
| 25 | if kind not in ENTRY_KINDS: |
| 26 | raise EntryValidationError(2, f"unknown entry kind: {kind}") |
| 27 | |
| 28 | if "entry_hash" in body or "prev_hash" in body: |
| 29 | raise EntryValidationError(2, "client must not supply entry_hash or prev_hash") |
| 30 | |
| 31 | body_kind = body.get("kind") |
| 32 | if body_kind is not None and body_kind != kind: |
| 33 | raise EntryValidationError(2, "body kind must match --kind when present") |
| 34 | |
| 35 | merged = dict(body) |
| 36 | merged["kind"] = kind |
| 37 | |
| 38 | version = merged.get("v", 1) |
| 39 | if version != 1: |
| 40 | raise EntryValidationError(2, "v must be integer 1") |
| 41 | |
| 42 | merged["v"] = 1 |
| 43 | |
| 44 | if kind == "genesis": |
| 45 | if "actor_role" in merged or "actor_session_id" in merged: |
| 46 | raise EntryValidationError(2, "genesis must not carry actor fields") |
| 47 | for key in ("assignment", "artifact_sha256", "passed", "evidence", "subject", "ruling", "bound_verdict_hash", "hook", "ok", "reason", "provenance"): |
| 48 | if key in merged: |
| 49 | raise EntryValidationError(2, f"genesis must not carry {key}") |
| 50 | return merged |
| 51 | |
| 52 | actor_role = merged.get("actor_role") |
| 53 | if actor_role not in ACTOR_ROLES: |
| 54 | raise EntryValidationError(23 if kind == "verdict" else 2, "invalid or missing actor_role") |
| 55 | |
| 56 | actor_session = merged.get("actor_session_id") |
| 57 | _require_non_empty_str(actor_session, "actor_session_id") |
| 58 | |
| 59 | if kind == "task_assigned": |
| 60 | if actor_role != "overseer": |
| 61 | raise EntryValidationError(23, "task_assigned requires actor_role=overseer") |
| 62 | _require_mapping(merged.get("assignment"), "assignment") |
| 63 | elif kind == "verdict": |
| 64 | if actor_role != "verifier": |
| 65 | raise EntryValidationError(23, "verdict requires actor_role=verifier") |
| 66 | _require_non_empty_str(merged.get("artifact_sha256"), "artifact_sha256") |
| 67 | passed = merged.get("passed") |
| 68 | if not isinstance(passed, bool): |
| 69 | raise EntryValidationError(2, "passed must be a boolean") |
| 70 | evidence = _require_mapping(merged.get("evidence"), "evidence") |
| 71 | reexecuted = evidence.get("reexecuted") |
| 72 | if not isinstance(reexecuted, list) or not reexecuted: |
| 73 | raise EntryValidationError(24, "evidence.reexecuted must be a non-empty list") |
| 74 | if not all(isinstance(item, str) for item in reexecuted): |
| 75 | raise EntryValidationError(2, "evidence.reexecuted entries must be strings") |
| 76 | elif kind == "dispute_opened": |
| 77 | _require_non_empty_str(merged.get("subject"), "subject") |
| 78 | elif kind == "overseer_ruling": |
| 79 | if actor_role != "overseer": |
| 80 | raise EntryValidationError(23, "overseer_ruling requires actor_role=overseer") |
| 81 | _require_non_empty_str(merged.get("ruling"), "ruling") |
| 82 | elif kind == "approval_recorded": |
| 83 | if actor_role != "owner": |
| 84 | raise EntryValidationError(23, "approval_recorded requires actor_role=owner") |
| 85 | _require_non_empty_str(merged.get("artifact_sha256"), "artifact_sha256") |
| 86 | _require_non_empty_str(merged.get("bound_verdict_hash"), "bound_verdict_hash") |
| 87 | elif kind == "board_advance": |
| 88 | _require_non_empty_str(merged.get("artifact_sha256"), "artifact_sha256") |
| 89 | _require_non_empty_str(merged.get("bound_verdict_hash"), "bound_verdict_hash") |
| 90 | elif kind == "hook_check": |
| 91 | hook = merged.get("hook") |
| 92 | if hook not in {"board_done", "handoff", "register"}: |
| 93 | raise EntryValidationError(2, "hook_check.hook must be board_done|handoff|register") |
| 94 | ok = merged.get("ok") |
| 95 | if not isinstance(ok, bool): |
| 96 | raise EntryValidationError(2, "ok must be a boolean") |
| 97 | reason = merged.get("reason") |
| 98 | if reason is not None and not isinstance(reason, str): |
| 99 | raise EntryValidationError(2, "reason must be a string when present") |
| 100 | |
| 101 | if "provenance" in merged: |
| 102 | merged["provenance"] = validate_provenance(merged["provenance"]) |
| 103 | |
| 104 | return merged |
| 105 | |
| 106 | |
| 107 | def find_passing_verdict( |
| 108 | entries: list[dict[str, Any]], |
| 109 | *, |
| 110 | artifact_sha256: str, |
| 111 | bound_verdict_hash: str, |
| 112 | ) -> bool: |
| 113 | """Return True when a passing verifier verdict matches the bound hash.""" |
| 114 | for entry in entries: |
| 115 | if entry.get("kind") != "verdict": |
| 116 | continue |
| 117 | if entry.get("actor_role") != "verifier": |
| 118 | continue |
| 119 | if entry.get("passed") is not True: |
| 120 | continue |
| 121 | if entry.get("artifact_sha256") != artifact_sha256: |
| 122 | continue |
| 123 | if entry.get("entry_hash") == bound_verdict_hash: |
| 124 | return True |
| 125 | return False |
File History
1 commit
sha256:4671b7f787ddbe63ced31c895b688c77ab495653b65a730b423329f26b3c1439
feat: K1-P1 complete — agent provenance, build-verification…
Sonnet 4.6
patch
57 days ago