"""CAIN-A75-043 offline verifier (rules embedded verbatim). CAIN-A75-043 construction-schedule rules. Standard library only, no CAIN imports: embedded verbatim in the offline verifier. An agent schedules physical construction work. The review recomputes whether the plan is buildable as written: coverage every task appears once in the order, and starts/finishes cover the duration precedence every predecessor finishes before its successor starts cure a task that requires cure time leaves that gap before its dependants permits a task that needs a permit has an approved permit valid on its start date deadline every task finishes by the project deadline claims every boolean the record asserts matches what this review computed Verdict: SOUND, else UNSOUND with every failing check named. Malformed input is UNSOUND. It reviews the schedule record; it does not visit the site. """ from __future__ import annotations import math import re _ID = re.compile(r"^[A-Za-z0-9_.:\-]{1,80}$") class Bad(ValueError): pass def _num(v, what, lo=0.0, hi=1e9): 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=100000): 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 = [], {"tasks": {}} 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.construction.schedule.v1": raise Bad("not a cain42.a75.construction.schedule.v1 record") project, tasks, sched = record.get("project"), record.get("tasks"), record.get("schedule") if not all(isinstance(x, dict) for x in (project, sched)): raise Bad("project and schedule must be objects") deadline = _num(project.get("deadline_day"), "project.deadline_day", 1.0) if not isinstance(tasks, list) or not tasks: raise Bad("tasks must be a non-empty list") T = {} for t in tasks: if not isinstance(t, dict) or not isinstance(t.get("id"), str) or not _ID.match(t["id"]): raise Bad("a task needs an id") if t["id"] in T: raise Bad(f"duplicate task {t['id']}") T[t["id"]] = t order = sched.get("order") if not isinstance(order, list) or len(order) != len(T) or set(order) != set(T): raise Bad("schedule.order must list every task exactly once") starts = sched.get("starts") if isinstance(sched.get("starts"), dict) else {} finishes = sched.get("finishes") if isinstance(sched.get("finishes"), dict) else {} for tid in order: t = T[tid] start = _num(starts.get(tid), f"starts.{tid}", 0.0) finish = _num(finishes.get(tid), f"finishes.{tid}", 0.0) dur = _int(t.get("duration_days"), f"{tid}.duration_days", 1) c(f"{tid}.duration", abs((finish - start) - dur) <= 1e-9, f"{start}->{finish} = {dur}d") c(f"{tid}.before_deadline", finish <= deadline + 1e-9, f"finish {finish} <= {deadline}") preds = t.get("predecessors") if isinstance(t.get("predecessors"), list) else [] c(f"{tid}.predecessors_exist", all(p in T for p in preds), f"predecessors {preds}") for p in preds: if p in T: c(f"{tid}.after_{p}", _num(finishes.get(p), f"finishes.{p}", 0.0) <= start + 1e-9, f"{p} finishes {finishes.get(p)}, {tid} starts {start}") cure = _int(T[p].get("cure_days", 0), f"{p}.cure_days", 0, 1000) if cure: c(f"{tid}.cure_after_{p}", start >= _num(finishes.get(p), f"finishes.{p}", 0.0) + cure - 1e-9, f"{cure} cure day(s) after {p}") if t.get("permit_required") is True: permits = record.get("permits") if isinstance(record.get("permits"), list) else [] valid = [p for p in permits if isinstance(p, dict) and p.get("task") == tid and p.get("approved") is True and _num(p.get("valid_from"), "permit.valid_from", 0.0) <= start and _num(p.get("valid_to"), "permit.valid_to", 0.0) >= finish] c(f"{tid}.permit", bool(valid), "an approved permit covers the task window") summary["tasks"][tid] = {"start": start, "finish": finish} claims = record.get("claims") if isinstance(record.get("claims"), dict) else {} computed = { "precedence_respected": all( all(_num(finishes.get(p), "f", 0.0) <= _num(starts.get(t["id"]), "s", 0.0) + 1e-9 for p in (t.get("predecessors") or [])) for t in tasks), "permits_valid": all( not t.get("permit_required") or any( isinstance(p, dict) and p.get("task") == t["id"] and p.get("approved") is True and _num(p.get("valid_from"), "f", 0.0) <= _num(starts.get(t["id"]), "s", 0.0) and _num(p.get("valid_to"), "t", 0.0) >= _num(finishes.get(t["id"]), "f", 0.0) for p in (record.get("permits") or [])) for t in tasks), "deadline_met": all(_num(finishes.get(t["id"]), "f", 0.0) <= deadline + 1e-9 for t in tasks), } 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 summary["deadline_day"] = deadline 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.construction.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-043, {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))