aaronrene / overseer-kit public
ledger.py python
271 lines 9.3 KB
Raw
sha256:0e9549ec7b463911bc08b7d586dc320b1ac9b1f5c943ee7e3865dcc6cb0f6f83 chore(governance): sync handover+roadmap to 84db8c8 (drift:… Human 1 day ago
1 """Ledger verify, append, and show (§K9.7 / §K9.9)."""
2
3 from __future__ import annotations
4
5 import json
6 from pathlib import Path
7 from typing import Any
8
9 from adapters.config import OverseerConfig
10 from cli.paths import confine_path, repo_relative
11 from tools.honesty.canonical import compute_entry_hash
12 from tools.honesty.gate import check_roles_file, honesty_module_disabled
13 from tools.honesty.genesis import GENESIS_PREV, build_genesis_entry, utc_now_z
14 from tools.honesty.ledger_io import (
15 LedgerIOError,
16 atomic_append_lines,
17 read_ledger_entries,
18 serialize_entry,
19 )
20 from tools.honesty.muse_registry import MuseAgentKeyRegistry
21 from tools.honesty.provenance import (
22 provenance_has_signature,
23 signature_required_for_kind,
24 validate_provenance,
25 verify_entry_provenance,
26 )
27 from tools.honesty.types import EntryValidationError, LedgerAppendOptions, LedgerResult
28 from tools.honesty.validate import find_passing_verdict, validate_append_body
29
30
31 def verify_chain(
32 entries: list[dict[str, Any]],
33 *,
34 regime: str = "git-only",
35 require_agent_signature: bool = False,
36 registry: MuseAgentKeyRegistry | None = None,
37 ) -> int:
38 """Walk the hash chain and verify optional provenance; return ``0``, ``2``, ``22``, ``25``, or ``26``.
39
40 Exit ``2`` when a stored ``provenance`` envelope is structurally malformed
41 (§P0.6 names ``verify`` as a surface for the malformed-provenance code), ``22``
42 on chain breakage, ``25`` on signature failure, ``26`` when a required signature
43 is absent.
44 """
45 if not entries:
46 return 0
47
48 expected_prev = GENESIS_PREV
49 for entry in entries:
50 if entry.get("v") != 1:
51 return 22
52 stored_hash = entry.get("entry_hash")
53 if not isinstance(stored_hash, str):
54 return 22
55 if entry.get("prev_hash") != expected_prev:
56 return 22
57 computed = compute_entry_hash(entry)
58 if computed != stored_hash.lower():
59 return 22
60 expected_prev = stored_hash.lower()
61
62 for entry in entries:
63 kind = entry.get("kind")
64 if kind == "genesis":
65 continue
66 if "provenance" in entry:
67 try:
68 validate_provenance(entry["provenance"])
69 except EntryValidationError as exc:
70 return exc.exit_code
71 if signature_required_for_kind(require_agent_signature=require_agent_signature, kind=str(kind)):
72 if not provenance_has_signature(entry):
73 return 26
74 sig_code = verify_entry_provenance(entry, regime=regime, registry=registry)
75 if sig_code != 0:
76 return sig_code
77 return 0
78
79
80 def _resolve_ledger_path(config: OverseerConfig, repo_root: Path) -> Path:
81 ledger = config.honesty.ledger
82 if ledger is None or not ledger.strip():
83 raise ValueError("honesty.ledger missing")
84 return confine_path(repo_root, ledger)
85
86
87 def _finalize_entry(body: dict[str, Any], prev_hash: str) -> dict[str, Any]:
88 """Fill envelope hashes and default timestamp."""
89 entry = dict(body)
90 if not entry.get("ts"):
91 entry["ts"] = utc_now_z()
92 entry["prev_hash"] = prev_hash
93 entry["entry_hash"] = compute_entry_hash(entry)
94 return entry
95
96
97 def append_entry(
98 *,
99 config: OverseerConfig,
100 repo_root: Path,
101 options: LedgerAppendOptions,
102 ) -> LedgerResult:
103 """Append one or more ledger lines per §K9.9."""
104 if honesty_module_disabled(config):
105 return LedgerResult(exit_code=4, stderr_extra="refused: honesty.enabled is false")
106
107 roles_exit, roles_warn = check_roles_file(config.honesty, repo_root)
108 if roles_exit is not None:
109 return LedgerResult(exit_code=roles_exit)
110
111 try:
112 ledger_path = _resolve_ledger_path(config, repo_root)
113 except Exception:
114 return LedgerResult(exit_code=4)
115
116 try:
117 body = validate_append_body(kind=options.kind, body=dict(options.body))
118 except EntryValidationError as exc:
119 return LedgerResult(exit_code=exc.exit_code, stderr_extra=str(exc))
120
121 existing = read_ledger_entries(ledger_path)
122
123 if options.kind == "genesis" and existing:
124 return LedgerResult(exit_code=2, stderr_extra="genesis on non-empty ledger")
125
126 if options.kind == "approval_recorded":
127 artifact_sha = body["artifact_sha256"]
128 bound_hash = body["bound_verdict_hash"]
129 if not find_passing_verdict(existing, artifact_sha256=artifact_sha, bound_verdict_hash=bound_hash):
130 return LedgerResult(exit_code=21, stderr_extra="approval without bound passing verdict")
131
132 if signature_required_for_kind(
133 require_agent_signature=config.honesty.require_agent_signature,
134 kind=options.kind,
135 ) and not provenance_has_signature(body):
136 return LedgerResult(exit_code=26, stderr_extra="signature required but absent")
137
138 lines_to_write: list[str] = []
139 prev_hash = existing[-1]["entry_hash"].lower() if existing else GENESIS_PREV
140
141 if not existing and options.kind != "genesis":
142 genesis = build_genesis_entry()
143 lines_to_write.append(serialize_entry(genesis))
144 prev_hash = genesis["entry_hash"].lower()
145
146 entry = _finalize_entry(body, prev_hash)
147 if provenance_has_signature(entry):
148 sig_code = verify_entry_provenance(entry, regime=config.vcs.regime)
149 if sig_code != 0:
150 return LedgerResult(exit_code=sig_code, stderr_extra="provenance signature verification failed")
151 lines_to_write.append(serialize_entry(entry))
152
153 try:
154 atomic_append_lines(ledger_path, lines_to_write)
155 except LedgerIOError as exc:
156 return LedgerResult(exit_code=5, stderr_extra=str(exc))
157
158 stderr = roles_warn or ""
159 return LedgerResult(exit_code=0, stderr_extra=stderr)
160
161
162 def verify_ledger_file(
163 *,
164 config: OverseerConfig,
165 repo_root: Path,
166 ) -> LedgerResult:
167 """Verify the configured ledger chain."""
168 if honesty_module_disabled(config):
169 return LedgerResult(exit_code=4, stderr_extra="refused: honesty.enabled is false")
170
171 roles_exit, roles_warn = check_roles_file(config.honesty, repo_root)
172 if roles_exit is not None:
173 return LedgerResult(exit_code=roles_exit)
174
175 try:
176 ledger_path = _resolve_ledger_path(config, repo_root)
177 except Exception:
178 return LedgerResult(exit_code=4)
179
180 if not ledger_path.is_file() or ledger_path.stat().st_size == 0:
181 return LedgerResult(exit_code=0, stderr_extra=roles_warn or "")
182
183 try:
184 entries = read_ledger_entries(ledger_path)
185 except ValueError as exc:
186 return LedgerResult(exit_code=22, stderr_extra=str(exc))
187
188 code = verify_chain(
189 entries,
190 regime=config.vcs.regime,
191 require_agent_signature=config.honesty.require_agent_signature,
192 )
193 return LedgerResult(exit_code=code, stderr_extra=roles_warn or "")
194
195
196 def show_entries(
197 *,
198 config: OverseerConfig,
199 repo_root: Path,
200 last_n: int,
201 ) -> LedgerResult:
202 """Print the last ``N`` ledger records as JSONL."""
203 if last_n < 1:
204 return LedgerResult(exit_code=1, stderr_extra="usage: --last must be >= 1")
205
206 if honesty_module_disabled(config):
207 return LedgerResult(exit_code=4, stderr_extra="refused: honesty.enabled is false")
208
209 roles_exit, roles_warn = check_roles_file(config.honesty, repo_root)
210 if roles_exit is not None:
211 return LedgerResult(exit_code=roles_exit)
212
213 try:
214 ledger_path = _resolve_ledger_path(config, repo_root)
215 except Exception:
216 return LedgerResult(exit_code=4)
217
218 if not ledger_path.is_file() or ledger_path.stat().st_size == 0:
219 return LedgerResult(exit_code=0, stderr_extra=roles_warn or "")
220
221 try:
222 entries = read_ledger_entries(ledger_path)
223 except ValueError as exc:
224 return LedgerResult(exit_code=22, stderr_extra=str(exc))
225
226 window = entries[-last_n:]
227 lines = [json.dumps(entry, ensure_ascii=False, separators=(",", ":"), sort_keys=True) for entry in window]
228 return LedgerResult(exit_code=0, stdout_lines=lines, stderr_extra=roles_warn or "")
229
230
231 def parse_append_body(
232 *,
233 repo_root: Path,
234 file_path: str | None,
235 stdin_text: str | None,
236 ) -> tuple[dict[str, Any] | None, int, str]:
237 """Load append JSON body from file or stdin; return (body, exit, error)."""
238 if file_path is not None:
239 try:
240 resolved = confine_path(repo_root, file_path)
241 except Exception:
242 return None, 4, "path escape"
243 if not resolved.is_file():
244 return None, 4, "file missing"
245 try:
246 text = resolved.read_text(encoding="utf-8")
247 body = json.loads(text)
248 except json.JSONDecodeError:
249 return None, 2, "malformed JSON"
250 except OSError:
251 return None, 4, "unreadable"
252 if not isinstance(body, dict):
253 return None, 2, "body must be object"
254 return body, 0, ""
255
256 if stdin_text is not None:
257 try:
258 body = json.loads(stdin_text)
259 except json.JSONDecodeError:
260 return None, 2, "malformed JSON"
261 if not isinstance(body, dict):
262 return None, 2, "body must be object"
263 return body, 0, ""
264
265 return {}, 0, ""
266
267
268 def ledger_rel_path(config: OverseerConfig, repo_root: Path) -> str:
269 """Return repo-relative ledger path for JSON payloads."""
270 ledger_path = _resolve_ledger_path(config, repo_root)
271 return repo_relative(repo_root, ledger_path)
File History 1 commit
sha256:6abcf1fa82a7a621ccbc945f19acdba5bc0db54569599404a1452fb4a096a199 fix(ISR): default require_independent_second_reviewer to require Human minor ⚠ 1 day ago