ledger.py
python
sha256:8461d44b77376fbf06fa7c3e085d309e3010fd8d5886d63c63e69ce118811ad4
docs: record AFF-b feature-tip SHAs after AFF-b-ISR commit.
Human
3 days 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 | version = entry.get("v") |
| 51 | if type(version) is not int or version != 1: |
| 52 | return 22 |
| 53 | ts = entry.get("ts") |
| 54 | if not isinstance(ts, str) or not ts.strip(): |
| 55 | return 22 |
| 56 | stored_hash = entry.get("entry_hash") |
| 57 | if not isinstance(stored_hash, str): |
| 58 | return 22 |
| 59 | if entry.get("prev_hash") != expected_prev: |
| 60 | return 22 |
| 61 | computed = compute_entry_hash(entry) |
| 62 | if computed != stored_hash.lower(): |
| 63 | return 22 |
| 64 | expected_prev = stored_hash.lower() |
| 65 | |
| 66 | for entry in entries: |
| 67 | kind = entry.get("kind") |
| 68 | if kind == "genesis": |
| 69 | continue |
| 70 | if "provenance" in entry: |
| 71 | try: |
| 72 | validate_provenance(entry["provenance"]) |
| 73 | except EntryValidationError as exc: |
| 74 | return exc.exit_code |
| 75 | if signature_required_for_kind(require_agent_signature=require_agent_signature, kind=str(kind)): |
| 76 | if not provenance_has_signature(entry): |
| 77 | return 26 |
| 78 | sig_code = verify_entry_provenance(entry, regime=regime, registry=registry) |
| 79 | if sig_code != 0: |
| 80 | return sig_code |
| 81 | return 0 |
| 82 | |
| 83 | |
| 84 | def _resolve_ledger_path(config: OverseerConfig, repo_root: Path) -> Path: |
| 85 | ledger = config.honesty.ledger |
| 86 | if ledger is None or not ledger.strip(): |
| 87 | raise ValueError("honesty.ledger missing") |
| 88 | return confine_path(repo_root, ledger) |
| 89 | |
| 90 | |
| 91 | def _finalize_entry(body: dict[str, Any], prev_hash: str) -> dict[str, Any]: |
| 92 | """Fill envelope hashes; server-fill ``ts`` only when the client omitted it.""" |
| 93 | entry = dict(body) |
| 94 | if "ts" not in entry: |
| 95 | entry["ts"] = utc_now_z() |
| 96 | entry["prev_hash"] = prev_hash |
| 97 | entry["entry_hash"] = compute_entry_hash(entry) |
| 98 | return entry |
| 99 | |
| 100 | |
| 101 | def append_entry( |
| 102 | *, |
| 103 | config: OverseerConfig, |
| 104 | repo_root: Path, |
| 105 | options: LedgerAppendOptions, |
| 106 | ) -> LedgerResult: |
| 107 | """Append one or more ledger lines per §K9.9.""" |
| 108 | if honesty_module_disabled(config): |
| 109 | return LedgerResult(exit_code=4, stderr_extra="refused: honesty.enabled is false") |
| 110 | |
| 111 | roles_exit, roles_warn = check_roles_file(config.honesty, repo_root) |
| 112 | if roles_exit is not None: |
| 113 | return LedgerResult(exit_code=roles_exit) |
| 114 | |
| 115 | try: |
| 116 | ledger_path = _resolve_ledger_path(config, repo_root) |
| 117 | except Exception: |
| 118 | return LedgerResult(exit_code=4) |
| 119 | |
| 120 | try: |
| 121 | body = validate_append_body(kind=options.kind, body=dict(options.body)) |
| 122 | except EntryValidationError as exc: |
| 123 | return LedgerResult(exit_code=exc.exit_code, stderr_extra=str(exc)) |
| 124 | |
| 125 | existing = read_ledger_entries(ledger_path) |
| 126 | |
| 127 | if options.kind == "genesis" and existing: |
| 128 | return LedgerResult(exit_code=2, stderr_extra="genesis on non-empty ledger") |
| 129 | |
| 130 | if options.kind == "approval_recorded": |
| 131 | artifact_sha = body["artifact_sha256"] |
| 132 | bound_hash = body["bound_verdict_hash"] |
| 133 | if not find_passing_verdict(existing, artifact_sha256=artifact_sha, bound_verdict_hash=bound_hash): |
| 134 | return LedgerResult(exit_code=21, stderr_extra="approval without bound passing verdict") |
| 135 | |
| 136 | if signature_required_for_kind( |
| 137 | require_agent_signature=config.honesty.require_agent_signature, |
| 138 | kind=options.kind, |
| 139 | ) and not provenance_has_signature(body): |
| 140 | return LedgerResult(exit_code=26, stderr_extra="signature required but absent") |
| 141 | |
| 142 | lines_to_write: list[str] = [] |
| 143 | prev_hash = existing[-1]["entry_hash"].lower() if existing else GENESIS_PREV |
| 144 | |
| 145 | if not existing and options.kind != "genesis": |
| 146 | genesis = build_genesis_entry() |
| 147 | lines_to_write.append(serialize_entry(genesis)) |
| 148 | prev_hash = genesis["entry_hash"].lower() |
| 149 | |
| 150 | entry = _finalize_entry(body, prev_hash) |
| 151 | if provenance_has_signature(entry): |
| 152 | sig_code = verify_entry_provenance(entry, regime=config.vcs.regime) |
| 153 | if sig_code != 0: |
| 154 | return LedgerResult(exit_code=sig_code, stderr_extra="provenance signature verification failed") |
| 155 | lines_to_write.append(serialize_entry(entry)) |
| 156 | |
| 157 | try: |
| 158 | atomic_append_lines(ledger_path, lines_to_write) |
| 159 | except LedgerIOError as exc: |
| 160 | return LedgerResult(exit_code=5, stderr_extra=str(exc)) |
| 161 | |
| 162 | stderr = roles_warn or "" |
| 163 | return LedgerResult(exit_code=0, stderr_extra=stderr) |
| 164 | |
| 165 | |
| 166 | def verify_ledger_file( |
| 167 | *, |
| 168 | config: OverseerConfig, |
| 169 | repo_root: Path, |
| 170 | ) -> LedgerResult: |
| 171 | """Verify the configured ledger chain.""" |
| 172 | if honesty_module_disabled(config): |
| 173 | return LedgerResult(exit_code=4, stderr_extra="refused: honesty.enabled is false") |
| 174 | |
| 175 | roles_exit, roles_warn = check_roles_file(config.honesty, repo_root) |
| 176 | if roles_exit is not None: |
| 177 | return LedgerResult(exit_code=roles_exit) |
| 178 | |
| 179 | try: |
| 180 | ledger_path = _resolve_ledger_path(config, repo_root) |
| 181 | except Exception: |
| 182 | return LedgerResult(exit_code=4) |
| 183 | |
| 184 | if not ledger_path.is_file() or ledger_path.stat().st_size == 0: |
| 185 | return LedgerResult(exit_code=0, stderr_extra=roles_warn or "") |
| 186 | |
| 187 | try: |
| 188 | entries = read_ledger_entries(ledger_path) |
| 189 | except ValueError as exc: |
| 190 | return LedgerResult(exit_code=22, stderr_extra=str(exc)) |
| 191 | |
| 192 | code = verify_chain( |
| 193 | entries, |
| 194 | regime=config.vcs.regime, |
| 195 | require_agent_signature=config.honesty.require_agent_signature, |
| 196 | ) |
| 197 | return LedgerResult(exit_code=code, stderr_extra=roles_warn or "") |
| 198 | |
| 199 | |
| 200 | def show_entries( |
| 201 | *, |
| 202 | config: OverseerConfig, |
| 203 | repo_root: Path, |
| 204 | last_n: int, |
| 205 | ) -> LedgerResult: |
| 206 | """Print the last ``N`` ledger records as JSONL.""" |
| 207 | if last_n < 1: |
| 208 | return LedgerResult(exit_code=1, stderr_extra="usage: --last must be >= 1") |
| 209 | |
| 210 | if honesty_module_disabled(config): |
| 211 | return LedgerResult(exit_code=4, stderr_extra="refused: honesty.enabled is false") |
| 212 | |
| 213 | roles_exit, roles_warn = check_roles_file(config.honesty, repo_root) |
| 214 | if roles_exit is not None: |
| 215 | return LedgerResult(exit_code=roles_exit) |
| 216 | |
| 217 | try: |
| 218 | ledger_path = _resolve_ledger_path(config, repo_root) |
| 219 | except Exception: |
| 220 | return LedgerResult(exit_code=4) |
| 221 | |
| 222 | if not ledger_path.is_file() or ledger_path.stat().st_size == 0: |
| 223 | return LedgerResult(exit_code=0, stderr_extra=roles_warn or "") |
| 224 | |
| 225 | try: |
| 226 | entries = read_ledger_entries(ledger_path) |
| 227 | except ValueError as exc: |
| 228 | return LedgerResult(exit_code=22, stderr_extra=str(exc)) |
| 229 | |
| 230 | window = entries[-last_n:] |
| 231 | lines = [json.dumps(entry, ensure_ascii=False, separators=(",", ":"), sort_keys=True) for entry in window] |
| 232 | return LedgerResult(exit_code=0, stdout_lines=lines, stderr_extra=roles_warn or "") |
| 233 | |
| 234 | |
| 235 | def parse_append_body( |
| 236 | *, |
| 237 | repo_root: Path, |
| 238 | file_path: str | None, |
| 239 | stdin_text: str | None, |
| 240 | ) -> tuple[dict[str, Any] | None, int, str]: |
| 241 | """Load append JSON body from file or stdin; return (body, exit, error).""" |
| 242 | if file_path is not None: |
| 243 | try: |
| 244 | resolved = confine_path(repo_root, file_path) |
| 245 | except Exception: |
| 246 | return None, 4, "path escape" |
| 247 | if not resolved.is_file(): |
| 248 | return None, 4, "file missing" |
| 249 | try: |
| 250 | text = resolved.read_text(encoding="utf-8") |
| 251 | body = json.loads(text) |
| 252 | except json.JSONDecodeError: |
| 253 | return None, 2, "malformed JSON" |
| 254 | except OSError: |
| 255 | return None, 4, "unreadable" |
| 256 | if not isinstance(body, dict): |
| 257 | return None, 2, "body must be object" |
| 258 | return body, 0, "" |
| 259 | |
| 260 | if stdin_text is not None: |
| 261 | try: |
| 262 | body = json.loads(stdin_text) |
| 263 | except json.JSONDecodeError: |
| 264 | return None, 2, "malformed JSON" |
| 265 | if not isinstance(body, dict): |
| 266 | return None, 2, "body must be object" |
| 267 | return body, 0, "" |
| 268 | |
| 269 | return {}, 0, "" |
| 270 | |
| 271 | |
| 272 | def ledger_rel_path(config: OverseerConfig, repo_root: Path) -> str: |
| 273 | """Return repo-relative ledger path for JSON payloads.""" |
| 274 | ledger_path = _resolve_ledger_path(config, repo_root) |
| 275 | return repo_relative(repo_root, ledger_path) |
File History
1 commit
sha256:8461d44b77376fbf06fa7c3e085d309e3010fd8d5886d63c63e69ce118811ad4
docs: record AFF-b feature-tip SHAs after AFF-b-ISR commit.
Human
3 days ago