"""CAIN-A75-044 offline verifier (rules embedded verbatim). CAIN-A75-044 contract-evidence rules. Standard library only, no CAIN imports: embedded verbatim in the offline verifier. A construction change order claims it is evidenced and approved. The review recomputes the package: items the change value equals the sum of quantity x unit price; every item cites a contract clause evidence every scope item has the required signed, hashed evidence linked to it approval every required role accepted after the request, someone other than the requester, within its value limit contract the change stays inside the contract's maximum overage claims every boolean the package asserts matches what this review computed Verdict: SOUND, else UNSOUND with every failing check named. Malformed input is UNSOUND. It reviews the change-order record; it does not visit the site or interpret the contract. """ from __future__ import annotations import math import re _ID = re.compile(r"^[A-Za-z0-9_.:\-]{1,80}$") _HEX = re.compile(r"^[0-9a-f]{64}$") EVIDENCE_TYPES = {"photo", "inspection", "email", "measurement", "report"} class Bad(ValueError): pass def _num(v, what, lo=0.0, hi=1e15): if isinstance(v, bool) or not isinstance(v, (int, float)) or not math.isfinite(v) or not (lo <= v <= hi): raise Bad(f"{what} must be a finite number in [{lo}, {hi}]") return float(v) def _int(v, what, lo=0, hi=10 ** 9): if isinstance(v, bool) or not isinstance(v, int) or not (lo <= v <= hi): raise Bad(f"{what} must be an integer in [{lo}, {hi}]") return v def evaluate(record): checks, summary = [], {"items": {}, "approvals": {}} def c(cid, ok, detail=""): checks.append({"id": cid, "ok": bool(ok), "detail": detail}) try: if not isinstance(record, dict) or record.get("schema") != "cain42.a75.contract.change.v1": raise Bad("not a cain42.a75.contract.change.v1 record") contract, change, pol = record.get("contract"), record.get("change"), record.get("policy") if not all(isinstance(x, dict) for x in (contract, change, pol)): raise Bad("contract, change and policy must be objects") original = _num(contract.get("original_value"), "contract.original_value", 1.0) change_value = _num(change.get("change_value"), "change.change_value", 0.0) max_overage = _num(pol.get("max_overage_pct"), "policy.max_overage_pct", 0.0, 1000.0) c("contract.overage", change_value <= original * (max_overage / 100.0) + 1e-9, f"change {change_value} <= {max_overage}% of {original}") clauses = set(contract.get("clauses") or []) c("contract.clauses", bool(clauses), f"{len(clauses)} clause(s)") requested_at = _num(change.get("requested_at"), "change.requested_at") items = change.get("scope_items") if not isinstance(items, list) or not items: raise Bad("change.scope_items must be a non-empty list") total = 0.0 I = {} for it in items: if not isinstance(it, dict) or not isinstance(it.get("id"), str) or not _ID.match(it["id"]): raise Bad("a scope item needs an id") if it["id"] in I: raise Bad(f"duplicate scope item {it['id']}") I[it["id"]] = it total += _num(it.get("quantity"), f"{it['id']}.quantity", 0.0) * _num(it.get("unit_price"), f"{it['id']}.unit_price", 0.0) c(f"{it['id']}.clause", it.get("clause") in clauses, f"clause {it.get('clause')!r}") c("items.arithmetic", abs(total - change_value) <= 0.01, f"sum {round(total, 2)} == stated {change_value}") summary["computed_total"] = round(total, 2) evidence = record.get("evidence") if not isinstance(evidence, list): raise Bad("evidence must be a list") required = _int(pol.get("required_evidence_per_item"), "policy.required_evidence_per_item", 1) E = {} for e in evidence: if not isinstance(e, dict) or not isinstance(e.get("id"), str) or not _ID.match(e["id"]): raise Bad("an evidence item needs an id") if e["id"] in E: raise Bad(f"duplicate evidence {e['id']}") E[e["id"]] = e c(f"evidence.{e['id']}.type", e.get("type") in EVIDENCE_TYPES, f"type {e.get('type')!r}") c(f"evidence.{e['id']}.hash", isinstance(e.get("sha256"), str) and bool(_HEX.match(e["sha256"])), "content hash") c(f"evidence.{e['id']}.signed", e.get("signed") is True, "evidence signed") _num(e.get("captured_at"), f"evidence.{e['id']}.captured_at") for iid in sorted(I): linked = [e for e in evidence if isinstance(e, dict) and iid in (e.get("linked_scope") or [])] good = [e for e in linked if e.get("type") in EVIDENCE_TYPES and isinstance(e.get("sha256"), str) and _HEX.match(e["sha256"]) and e.get("signed") is True] c(f"{iid}.evidence", len(good) >= required, f"{len(good)} valid evidence >= {required}") summary["items"][iid] = {"evidence": len(good)} approvals = record.get("approvals") if not isinstance(approvals, list) or not approvals: raise Bad("approvals must be a non-empty list") required_roles = pol.get("required_roles") if not isinstance(required_roles, list) or not required_roles: raise Bad("policy.required_roles must be a non-empty list") threshold = _num(pol.get("approval_threshold"), "policy.approval_threshold", 0.0) for role in required_roles: ap = next((a for a in approvals if isinstance(a, dict) and a.get("role") == role), None) c(f"approval.{role}", ap is not None and ap.get("decision") == "accept", f"{role}: {ap and ap.get('decision')!r}") if ap is not None: c(f"approval.{role}.independent", ap.get("by") not in (None, "", change.get("requested_by")), f"by {ap.get('by')!r}") c(f"approval.{role}.after_request", _num(ap.get("at"), f"approval.{role}.at") >= requested_at - 1e-9, "approval after the request") c(f"approval.{role}.limit", change_value <= _num(ap.get("limit_value"), f"approval.{role}.limit_value", 0.0) + 1e-9, f"value {change_value} within {ap.get('limit_value')}") c(f"approval.{role}.threshold", change_value <= threshold + 1e-9 or role in required_roles, "required role covers the threshold") claims = record.get("claims") if isinstance(record.get("claims"), dict) else {} computed = { "all_items_evidenced": all(len([e for e in evidence if iid in (e.get("linked_scope") or []) and e.get("signed") is True and isinstance(e.get("sha256"), str) and _HEX.match(e["sha256"])]) >= required for iid in I), "approvals_valid": all( any(isinstance(a, dict) and a.get("role") == role and a.get("decision") == "accept" and a.get("by") not in (None, "", change.get("requested_by")) and _num(a.get("at"), "approval.at") >= requested_at - 1e-9 and change_value <= _num(a.get("limit_value"), "approval.limit_value", 0.0) + 1e-9 for a in approvals) for role in required_roles), "within_contract_limit": change_value <= original * (max_overage / 100.0) + 1e-9, } for key in sorted(claims): c(f"claims.{key}", key in computed and claims[key] == computed[key], f"claimed {claims[key]!r}, computed {computed.get(key)!r}") summary["computed"] = computed except Bad as e: checks.append({"id": "input.well_formed", "ok": False, "detail": str(e)}) return "UNSOUND", checks, summary except Exception as e: # noqa: BLE001 checks.append({"id": "input.error", "ok": False, "detail": f"{type(e).__name__}: {e}"[:200]}) return "UNSOUND", checks, summary return ("SOUND" if all(x["ok"] for x in checks) else "UNSOUND"), checks, summary # ---------------------------------------------------------------------------- offline verifier (no CAIN imports) # Every name below is prefixed _kit_verifier_ / _b64 / _hl / _json / _sys so it cannot shadow a name in the rules above. import base64 as _b64, hashlib as _hl, json as _json, sys as _sys _KIT_VERIFIER_GENESIS = _hl.sha256(b"cain42.a75.contract.genesis").hexdigest() def _kit_verifier_canon(o): return _json.dumps(o, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode() def _kit_verifier_sha(o): return _hl.sha256(_kit_verifier_canon(o)).hexdigest() def _kit_verifier_main(argv): from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey if len(argv) < 2: print("usage: verify_offline.py run.json --pinned-key BASE64"); return 2 run = _json.load(open(argv[1])) pin = argv[argv.index("--pinned-key") + 1] if "--pinned-key" in argv else None problems, prev, recs = [], _KIT_VERIFIER_GENESIS, run.get("records") or [] for i, r in enumerate(recs): body = {k: v for k, v in r.items() if k not in ("record_hash", "signature_b64")} if _kit_verifier_sha(body) != r.get("record_hash"): problems.append(f"record {i}: record_hash does not match") if body.get("subject_sha256") != _kit_verifier_sha(body.get("subject")): problems.append(f"record {i}: subject_sha256 does not match") if pin and r.get("public_key_b64") != pin: problems.append(f"record {i}: different key than the pinned one") try: Ed25519PublicKey.from_public_bytes(_b64.b64decode(pin or r.get("public_key_b64"))).verify( _b64.b64decode(r.get("signature_b64", "")), str(r.get("record_hash")).encode()) except Exception: problems.append(f"record {i}: signature does not verify") if r.get("prev_record_hash") != prev: problems.append(f"record {i}: chain link broken") v, checks, _ = evaluate(r.get("subject")) if v != r.get("verdict") or checks != r.get("checks"): problems.append(f"record {i}: re-running the rules gives {v}, the record says {r.get('verdict')}") prev = r.get("record_hash") if run.get("run_hash") != _kit_verifier_sha([r.get("record_hash") for r in recs]): problems.append("run_hash does not match") for p in problems: print(" -", p) if not pin: print("WARNING: no --pinned-key given; the signature was checked against the key inside the file") print("VERDICT:", "VERIFIED" if not problems else "TAMPERED OR INVALID", f"(CAIN-A75-044, {len(recs)} record(s), verdicts {run.get('verdicts')})") return 0 if not problems else 1 if __name__ == "__main__": _sys.exit(_kit_verifier_main(_sys.argv))