"""Verify the narrow product delta from DualCoach v1.0.0 to v1.1.0.""" from __future__ import annotations import argparse import base64 import csv import hashlib import io import json import zipfile from dataclasses import dataclass from pathlib import Path from typing import Final _TTL_MEMBER: Final = "gateway/platforms/telegram_customer_bootstrap.py" _RECORD_MEMBER: Final = "hermes_agent-0.17.0.dist-info/RECORD" _EXPECTED_CHANGED_MEMBERS: Final = (_TTL_MEMBER, _RECORD_MEMBER) _OLD_TTL: Final = b"_INVITE_TTL: Final = timedelta(minutes=30)" _NEW_TTL: Final = b"_INVITE_TTL: Final = timedelta(hours=24)" _DIGEST_LENGTH: Final = 64 @dataclass(frozen=True, slots=True) class WheelDelta: """Verified member-level difference between two Hermes wheels.""" member_count: int changed_members: tuple[str, ...] def _canonical(value: dict[str, object]) -> bytes: return json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False, ).encode() def _read_object(path: Path) -> dict[str, object]: value = json.loads(path.read_text(encoding="utf-8")) if not isinstance(value, dict) or any(not isinstance(key, str) for key in value): raise ValueError(f"{path.name} is not an object") return value def _sha256(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def _record_digest(encoded: str) -> str: algorithm, separator, value = encoded.partition("=") if algorithm != "sha256" or separator != "=": raise ValueError("wheel RECORD digest is invalid") padded = value + "=" * (-len(value) % 4) digest = base64.urlsafe_b64decode(padded) if len(digest) * 2 != _DIGEST_LENGTH: raise ValueError("wheel RECORD digest length is invalid") return digest.hex() def _verify_record(archive: zipfile.ZipFile, names: tuple[str, ...]) -> None: if len(names) != len(set(names)) or _RECORD_MEMBER not in names: raise ValueError("wheel member inventory is invalid") rows = csv.reader(io.StringIO(archive.read(_RECORD_MEMBER).decode("utf-8"))) seen: set[str] = set() for row in rows: if len(row) != 3: raise ValueError("wheel RECORD row is invalid") relative, encoded_hash, raw_size = row if relative in seen or relative not in names: raise ValueError("wheel RECORD path is invalid") seen.add(relative) if relative == _RECORD_MEMBER: if encoded_hash or raw_size: raise ValueError("wheel RECORD self-row is invalid") continue if not raw_size.isdecimal(): raise ValueError("wheel RECORD size is invalid") payload = archive.read(relative) if int(raw_size) != len(payload) or _record_digest(encoded_hash) != _sha256( payload ): raise ValueError("wheel RECORD content binding is invalid") if seen != set(names): raise ValueError("wheel RECORD inventory is incomplete") def verify_wheel_delta(base_wheel: Path, successor_wheel: Path) -> WheelDelta: """Require the successor wheel to change only TTL code and RECORD.""" with ( zipfile.ZipFile(base_wheel) as base, zipfile.ZipFile(successor_wheel) as successor, ): base_names = tuple(base.namelist()) successor_names = tuple(successor.namelist()) if base_names != successor_names: raise ValueError("wheel member order or inventory changed") _verify_record(base, base_names) _verify_record(successor, successor_names) changed = tuple( name for name in base_names if base.read(name) != successor.read(name) ) if changed != _EXPECTED_CHANGED_MEMBERS: raise ValueError("wheel delta exceeds the TTL seam") before = base.read(_TTL_MEMBER) after = successor.read(_TTL_MEMBER) if ( before.count(_OLD_TTL) != 1 or before.count(_NEW_TTL) != 0 or after.count(_OLD_TTL) != 0 or after.count(_NEW_TTL) != 1 or before.replace(_OLD_TTL, _NEW_TTL) != after ): raise ValueError("TTL source replacement is not exact") return WheelDelta(len(base_names), changed) def _verify_binding(document: dict[str, object]) -> dict[str, object]: if set(document) != { "schema", "derivation_inputs", "candidate_digest", "binding_sha256", }: raise ValueError("product binding fields are invalid") raw_inputs = document["derivation_inputs"] if not isinstance(raw_inputs, dict) or any( not isinstance(key, str) for key in raw_inputs ): raise ValueError("product derivation inputs are invalid") inputs = dict(raw_inputs) candidate = _sha256(_canonical(inputs)) unsigned = {key: value for key, value in document.items() if key != "binding_sha256"} if ( document["schema"] != "task26-preexecution-product-binding-v1" or document["candidate_digest"] != candidate or document["binding_sha256"] != _sha256(_canonical(unsigned)) ): raise ValueError("product binding digest is invalid") return inputs def verify_product_binding_delta( base_binding: Path, successor_binding: Path, ) -> str: """Require canonical bindings to differ only by the Hermes wheel hash.""" base = _read_object(base_binding) successor = _read_object(successor_binding) base_inputs = _verify_binding(base) successor_inputs = _verify_binding(successor) changed = { key for key in base_inputs.keys() | successor_inputs.keys() if base_inputs.get(key) != successor_inputs.get(key) } if changed != {"hermes_wheel_sha256"}: raise ValueError("product derivation delta exceeds the Hermes wheel") candidate = successor["candidate_digest"] if not isinstance(candidate, str) or len(candidate) != _DIGEST_LENGTH: raise ValueError("successor candidate digest is invalid") return candidate def verify_release_inputs( base_root: Path, successor_root: Path, ) -> dict[str, object]: """Verify product identity and both fresh Golden Path qualifications.""" base_binding_path = base_root / "preexecution-product-binding.json" successor_binding_path = successor_root / "preexecution-product-binding.json" candidate = verify_product_binding_delta( base_binding_path, successor_binding_path, ) binding = _read_object(successor_binding_path) inputs = _verify_binding(binding) base_wheel = ( base_root / "artifacts" / "hermes_agent-0.17.0-py3-none-any.whl" ) successor_wheel = ( successor_root / "artifacts" / "hermes_agent-0.17.0-py3-none-any.whl" ) profile_wheel = ( successor_root / "artifacts" / "physique_checkin_cli-0.1.0-py3-none-any.whl" ) if ( inputs.get("hermes_wheel_sha256") != _sha256(successor_wheel.read_bytes()) or inputs.get("profile_wheel_sha256") != _sha256(profile_wheel.read_bytes()) ): raise ValueError("successor wheel hash differs from product binding") delta = verify_wheel_delta(base_wheel, successor_wheel) statuses: dict[str, str] = {} for mode, expected in ( ("source", "ACTUAL_SOURCE_GOLDEN_PATH_PASS"), ("installed", "ACTUAL_INSTALLED_GOLDEN_PATH_PASS"), ): result = _read_object(successor_root / "receipts" / f"{mode}-result.json") verification = _read_object( successor_root / "receipts" / f"{mode}-verify.json" ) if ( result.get("candidate_digest") != candidate or result.get("candidate_product_binding_sha256") != binding.get("binding_sha256") or result.get("hermes_wheel_sha256") != inputs.get("hermes_wheel_sha256") or result.get("profile_wheel_sha256") != inputs.get("profile_wheel_sha256") or verification.get("candidate_digest") != candidate or verification.get("status") != expected ): raise ValueError(f"{mode} Golden Path binding is invalid") statuses[mode] = expected return { "candidate_digest": candidate, "installed_golden_path": statuses["installed"], "source_golden_path": statuses["source"], "wheel_member_count": delta.member_count, } def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("base", type=Path) parser.add_argument("successor", type=Path) args = parser.parse_args() result = { "status": "DUALCOACH_V11_CANDIDATE_INPUTS_PASS", **verify_release_inputs(args.base, args.successor), } print(json.dumps(result, sort_keys=True, separators=(",", ":"))) return 0 if __name__ == "__main__": raise SystemExit(main())