"""CAIN-A75-028 offline verifier (rules embedded verbatim). CAIN-A75-028 mobile-agent rules. Standard library only, no CAIN imports: embedded verbatim in the offline verifier. A mobile agent acts across apps on a phone. The review recomputes whether every action was inside consent and policy: device platform, supported OS version, MDM enrolment and an app-inventory hash are present consent consent is present, unexpired at the supplied clock, and names the purpose and app scope apps every action's app is in the session scope and the policy allowlist, and not on the forbidden list sensitive typed sensitive data goes only to an allowlisted app, and a screen containing sensitive data is redacted purchase a purchase needs a live approval by someone other than the agent and stays under the amount ceiling effects an applied action is allowed and carries a receipt hash; a blocked action is not applied claims every boolean the record asserts matches what this review computed Verdict: COMPLIANT, else NON_COMPLIANT with every failing check named. Malformed input is NON_COMPLIANT. It reviews the action trace; it does not control the phone. """ 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}$") KINDS = {"tap", "type", "purchase", "share", "settings"} class Bad(ValueError): pass def _num(v, what, lo=0.0, hi=1e12): 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 evaluate(record): checks, summary = [], {"actions": {}, "blocked": []} 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.mobile.trace.v1": raise Bad("not a cain42.a75.mobile.trace.v1 record") device, sess, pol = record.get("device"), record.get("session"), record.get("policy") if not all(isinstance(x, dict) for x in (device, sess, pol)): raise Bad("device, session and policy must be objects") c("device.platform", device.get("platform") in ("ios", "android"), f"platform {device.get('platform')!r}") c("device.os_supported", device.get("os_version_supported") is True, "the OS version is supported") c("device.mdm", device.get("mdm_enrolled") is True, "the device is managed") c("device.inventory", isinstance(device.get("app_inventory_sha256"), str) and bool(_HEX.match(device["app_inventory_sha256"])), "the app inventory is hashed") now, exp = _num(sess.get("now"), "session.now"), _num(sess.get("expires_at"), "session.expires_at") c("consent.present", sess.get("user_consent") is True, "the user consented") c("consent.not_expired", exp > now, f"consent expires {exp} > now {now}") c("consent.purpose", isinstance(sess.get("purpose"), str) and bool(sess["purpose"].strip()), "a purpose is stated") scope = sess.get("app_scope") if not isinstance(scope, list) or not scope or not all(isinstance(x, str) for x in scope): raise Bad("session.app_scope must be a non-empty list of app ids") allowed = pol.get("allowed_apps") if isinstance(pol.get("allowed_apps"), list) else [] forbidden = pol.get("forbidden_apps") if isinstance(pol.get("forbidden_apps"), list) else [] max_purchase = _num(pol.get("max_purchase_usd", 0.0), "policy.max_purchase_usd", 0.0) c("policy.lists", bool(allowed) and bool(forbidden), "allowed and forbidden app lists exist") approvals = {} for ap in record.get("approvals") or []: if not isinstance(ap, dict) or not isinstance(ap.get("action_id"), str): raise Bad("an approval needs an action_id") approvals[ap["action_id"]] = ap effects = {e.get("action_id"): e for e in (record.get("effects") or []) if isinstance(e, dict)} actions = record.get("actions") if not isinstance(actions, list) or not actions or len(actions) > 500: raise Bad("actions must be a non-empty list") seen = set() for a in actions: if not isinstance(a, dict) or not isinstance(a.get("id"), str) or not _ID.match(a["id"]): raise Bad("an action needs an id") if a["id"] in seen: raise Bad(f"duplicate action {a['id']}") seen.add(a["id"]) kind, app = a.get("kind"), a.get("app") if kind not in KINDS: raise Bad(f"{a['id']}: kind must be one of {sorted(KINDS)}") if not isinstance(app, str) or not _ID.match(app): raise Bad(f"{a['id']}: app required") ok_app = app in scope and app in allowed and app not in forbidden c(f"{a['id']}.app_allowed", ok_app, f"app {app!r}, scope {scope}, allowed {allowed}, forbidden {forbidden}") sensitive = a.get("sensitive_data") if isinstance(a.get("sensitive_data"), list) else [] if sensitive: c(f"{a['id']}.sensitive_app", app in allowed and app not in forbidden, "sensitive data only goes to an allowlisted app") c(f"{a['id']}.screenshot_redacted", a.get("screenshot_redacted") is True, "a screen with sensitive data is redacted") ok_extra = True if kind == "purchase": ap = approvals.get(a["id"]) amount = _num(a.get("amount_usd"), f"{a['id']}.amount_usd", 0.0) c(f"{a['id']}.purchase_within_limit", amount <= max_purchase + 1e-9, f"${amount} <= ${max_purchase}") c(f"{a['id']}.purchase_approved", ap is not None and ap.get("approved_by") not in (None, "", sess.get("agent")), f"approved by {(ap or {}).get('approved_by')!r}") if ap is not None: c(f"{a['id']}.approval_live", _num(ap.get("expires_at"), f"approval.{a['id']}.expires_at") > now, "approval not expired") ok_extra = amount <= max_purchase + 1e-9 and ap is not None c(f"{a['id']}.accessibility_scope", a.get("accessibility_overreach") is not True, "no accessibility overreach") eff = effects.get(a["id"]) applied = bool(eff and eff.get("applied") is True) allowed_action = ok_app and ok_extra and a.get("accessibility_overreach") is not True c(f"{a['id']}.effect_consistent", (applied and allowed_action) or (not applied), f"applied={applied}, allowed={allowed_action}") if applied: c(f"{a['id']}.receipt", isinstance(eff.get("receipt_sha256"), str) and bool(_HEX.match(eff["receipt_sha256"])), "an applied action carries a receipt hash") summary["actions"][a["id"]] = {"app": app, "kind": kind, "allowed": allowed_action, "applied": applied} if not allowed_action: summary["blocked"].append(a["id"]) claims = record.get("claims") if isinstance(record.get("claims"), dict) else {} computed = { "consent_present": sess.get("user_consent") is True and exp > now, "all_apps_allowed": not summary["blocked"], "no_sensitive_leak": not any( (a.get("sensitive_data") and (a.get("app") not in allowed or a.get("screenshot_redacted") is not True)) for a in actions), } 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 "NON_COMPLIANT", checks, summary except Exception as e: # noqa: BLE001 checks.append({"id": "input.error", "ok": False, "detail": f"{type(e).__name__}: {e}"[:200]}) return "NON_COMPLIANT", checks, summary return ("COMPLIANT" if all(x["ok"] for x in checks) else "NON_COMPLIANT"), 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.mobile.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-028, {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))