"""CAIN-A75-020 offline verifier (rules embedded verbatim). CAIN-A75-020 os-containment rules. Standard library only, no CAIN imports: embedded verbatim in the offline verifier. An agent runs inside an OS sandbox. The review recomputes whether the sandbox is actually confining: namespaces user, mount, pid and net namespaces are all enabled syscalls a seccomp profile exists, its default action denies, and dangerous syscalls (mount, ptrace, bpf, kexec_load, init_module) are not on the allow list privilege no_new_privs is set and the capability bounding set excludes CAP_SYS_ADMIN mounts a sensitive host path (/etc, /proc/sys, /root, the container socket) is never writable network egress is denied, or an allowlist is non-empty and contains no catch-all resources memory and pid limits are set secrets no inherited environment name looks like a credential and no secret is mounted claims every boolean the record asserts matches what this review computed Verdict: CONTAINED, else NOT_CONTAINED with every failing check named. Malformed input is NOT_CONTAINED. It reviews the sandbox configuration record; it does not start the sandbox. """ from __future__ import annotations import re _ID = re.compile(r"^[A-Za-z0-9_.:\-]{1,80}$") _SENSITIVE = ("/etc", "/proc/sys", "/root", "/var/run/docker.sock", "/run/docker.sock", "/dev/kmsg") _ALWAYS_FORBIDDEN = ("/var/run/docker.sock", "/run/docker.sock", "/dev/kmsg") _DANGEROUS = {"mount", "umount2", "ptrace", "bpf", "kexec_load", "init_module", "finit_module", "pivot_root", "unshare"} _SECRET = re.compile(r"(KEY|TOKEN|SECRET|PASSWORD|PASSWD|CREDENTIAL|PRIVATE)", re.I) class Bad(ValueError): pass def evaluate(record): checks, summary = [], {} 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.os.containment.v1": raise Bad("not a cain42.a75.os.containment.v1 record") sb = record.get("sandbox") if not isinstance(sb, dict): raise Bad("sandbox must be an object") if sb.get("engine") not in ("bubblewrap", "firejail", "nsjail", "gvisor", "docker", "runc"): raise Bad("sandbox.engine must be a known sandbox") ns = record.get("namespaces") if isinstance(record.get("namespaces"), dict) else {} for name in ("user", "mount", "pid", "net"): c(f"namespace.{name}", ns.get(name) is True, f"{name} namespace {ns.get(name)!r}") sc = record.get("seccomp") if isinstance(record.get("seccomp"), dict) else {} c("seccomp.present", bool(sc.get("profile")), f"profile {sc.get('profile')!r}") c("seccomp.default_deny", sc.get("default_action") in ("SCMP_ACT_ERRNO", "SCMP_ACT_KILL", "SCMP_ACT_KILL_PROCESS"), f"default {sc.get('default_action')!r}") allowed = sc.get("allowed_syscalls") if not isinstance(allowed, list) or not all(isinstance(x, str) for x in allowed): raise Bad("seccomp.allowed_syscalls must be a list of names") dangerous = sorted(set(allowed) & _DANGEROUS) c("seccomp.no_dangerous_syscalls", not dangerous, f"dangerous allowed: {dangerous}" if dangerous else "no dangerous syscall allowed") caps = record.get("capabilities") if isinstance(record.get("capabilities"), dict) else {} c("privilege.no_new_privs", caps.get("no_new_privs") is True, "no_new_privs is set") bounding = caps.get("bounding_set") if not isinstance(bounding, list) or not all(isinstance(x, str) for x in bounding): raise Bad("capabilities.bounding_set must be a list") c("privilege.no_sys_admin", "CAP_SYS_ADMIN" not in bounding, f"bounding set {bounding}") mounts = record.get("mounts") if isinstance(record.get("mounts"), list) else [] mount_targets = set() for m in mounts: if not isinstance(m, dict) or not isinstance(m.get("target"), str): raise Bad("a mount needs a target") if m["target"] in mount_targets: raise Bad(f"{m['target']}: mounted twice") mount_targets.add(m["target"]) if m.get("mode") not in ("ro", "rw"): raise Bad(f"{m['target']}: mode must be ro or rw") source = m.get("source", "") sensitive = isinstance(source, str) and any(source == s or source.startswith(s + "/") for s in _SENSITIVE) c(f"mount.{m['target']}.sensitive_not_writable", not (sensitive and m["mode"] == "rw"), f"{source!r} mounted {m['mode']}") c(f"mount.{m['target']}.not_forbidden", not (isinstance(source, str) and source in _ALWAYS_FORBIDDEN), f"{source!r} is never mountable into an agent sandbox, even read-only") net = record.get("network") if isinstance(record.get("network"), dict) else {} egress = net.get("egress") allow = net.get("allowlist") if isinstance(net.get("allowlist"), list) else [] c("network.egress", egress in ("deny", "allowlist"), f"egress {egress!r}") c("network.no_catch_all", not any(a in ("0.0.0.0/0", "::/0", "*") for a in allow), f"allowlist {allow}") c("network.allowlist_if_needed", egress != "allowlist" or bool(allow), "an allowlist is non-empty") res = record.get("resources") if isinstance(record.get("resources"), dict) else {} c("resources.memory", isinstance(res.get("memory_max_mb"), int) and not isinstance(res.get("memory_max_mb"), bool) and res["memory_max_mb"] > 0, f"memory_max_mb {res.get('memory_max_mb')!r}") c("resources.pids", isinstance(res.get("pids_max"), int) and not isinstance(res.get("pids_max"), bool) and res["pids_max"] > 0, f"pids_max {res.get('pids_max')!r}") sec = record.get("secrets") if isinstance(record.get("secrets"), dict) else {} env = sec.get("inherited_env") if isinstance(sec.get("inherited_env"), list) else [] leaked = sorted(x for x in env if isinstance(x, str) and _SECRET.search(x)) c("secrets.no_credential_env", not leaked, f"credential-like env: {leaked}" if leaked else "no credential-like env") mounted = sec.get("mounted_files") if isinstance(sec.get("mounted_files"), list) else [] c("secrets.no_secret_mount", not mounted, f"mounted secrets: {mounted}") claims = record.get("claims") if isinstance(record.get("claims"), dict) else {} computed = { "namespaces_complete": all(ns.get(n) is True for n in ("user", "mount", "pid", "net")), "seccomp_denies_by_default": sc.get("default_action") in ("SCMP_ACT_ERRNO", "SCMP_ACT_KILL", "SCMP_ACT_KILL_PROCESS") and not dangerous, "no_new_privs": caps.get("no_new_privs") is True, "no_secret_leak": not leaked and not mounted, } 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["sandbox"] = sb.get("engine") except Bad as e: checks.append({"id": "input.well_formed", "ok": False, "detail": str(e)}) return "NOT_CONTAINED", checks, summary except Exception as e: # noqa: BLE001 checks.append({"id": "input.error", "ok": False, "detail": f"{type(e).__name__}: {e}"[:200]}) return "NOT_CONTAINED", checks, summary return ("CONTAINED" if all(x["ok"] for x in checks) else "NOT_CONTAINED"), 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.oscontainment.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-020, {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))