proof-aware-crypto-tooling-agent/notebooks/10_verified_custody_wallet.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Lecture 10: The Verified-Custody Wallet (warden)\n",
    "\n",
    "Everything so far *decided* which cryptographic code to trust.\n",
    "This lecture *acts* on the decision: we build a custody boundary\n",
    "out of the four proven curve25519-dalek forks and use it to\n",
    "guard signatures - inbound and outbound.\n",
    "\n",
    "The one idea: **inbound acceptance requires a unanimous quorum of\n",
    "provably-equivalent verifiers, and every outbound signature must\n",
    "pass the same quorum before it is released.**\n",
    "\n",
    "We keep the course's ratchet rule: every load-bearing idea runs\n",
    "twice - napkin scale by hand, then real scale against the live\n",
    "system - and both are executable here.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Learning Objectives\n",
    "\n",
    "- Explain why a *unanimous* quorum of provably-equivalent\n",
    "  verifiers turns disagreement into evidence of a fault, and why\n",
    "  majority voting would hide exactly that fault.\n",
    "- Classify a quorum divergence as a documented semantic edge\n",
    "  (note) versus unexplained (tamper -> latch).\n",
    "- Describe the outbound signing firewall as verify-after-sign\n",
    "  with a proven verifier, and state warden's honest asymmetry\n",
    "  (verify custody-grade, sign trusted base).\n",
    "- Recompute a custody card's inclusion proof as a counterparty -\n",
    "  trust by recomputation, not by assertion.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Why a quorum, when one proof would do?\n",
    "\n",
    "Each member is *proven* to decide the same predicate,\n",
    "`accept(A,m,R,s) ⇔ decompress(R) = [k](−A) + [s]B`. So on the\n",
    "proven domain they cannot disagree about *meaning*. Classic\n",
    "N-version programming hopes independent code won't share a bug;\n",
    "we do not hope - we know the semantics coincide, so a runtime\n",
    "disagreement is not opinion, it is **evidence of a fault**: a\n",
    "corrupted build, a memory error, or tampering. The quorum turns\n",
    "\"the verifiers differed\" into an alarm with a theorem behind it.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Napkin scale: a 3-of-3 quorum with toy verifiers\n",
    "\n",
    "Forget real curves for a moment. Model three verifiers as\n",
    "functions and watch the boundary logic: unanimity accepts,\n",
    "any disagreement fails closed and is classified.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def toy_quorum(verdicts):\n",
    "    kinds = set(verdicts.values())\n",
    "    if kinds == {\"accept\"}:\n",
    "        return \"unanimous-accept\", True\n",
    "    if kinds == {\"reject\"}:\n",
    "        return \"unanimous-reject\", False\n",
    "    return \"divergence -> FAIL CLOSED + incident\", False\n",
    "\n",
    "print(toy_quorum({\"dalek\": \"accept\", \"anza\": \"accept\", \"risc0\": \"accept\"}))\n",
    "print(toy_quorum({\"dalek\": \"reject\", \"anza\": \"reject\", \"risc0\": \"reject\"}))\n",
    "print(toy_quorum({\"dalek\": \"accept\", \"anza\": \"reject\", \"risc0\": \"accept\"}))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The third line is the whole point: a lone dissenter does not get\n",
    "out-voted. Acceptance needs *everyone*; anything else is a\n",
    "refusal plus a recorded incident. Majority voting would hide\n",
    "exactly the fault we most want to see.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## The divergence taxonomy\n",
    "\n",
    "The forks are *allowed* to differ on documented degenerate\n",
    "inputs (anza rejects `A = 0` and a legacy excluded-small-order-R\n",
    "list). We still fail closed; the taxonomy only grades the alarm:\n",
    "\n",
    "- **semantic-edge** - they differ AND a documented edge flag\n",
    "  applies (small-order R, non-canonical s, zero key): severity\n",
    "  *note*.\n",
    "- **unexplained** - they differ with no documented reason, or a\n",
    "  member errored: severity *tamper* -> custody **latches**.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import sys, pathlib\n",
    "for parent in [pathlib.Path.cwd(), *pathlib.Path.cwd().parents]:\n",
    "    if (parent / \"src\" / \"pacta\").exists():\n",
    "        sys.path.insert(0, str(parent / \"src\")); ROOT = parent; break\n",
    "\n",
    "from pacta.quorum import semantic_edge_flags, SMALL_ORDER_ENCODINGS\n",
    "\n",
    "small_order_R = sorted(SMALL_ORDER_ENCODINGS)[0]\n",
    "print(\"edge flags for a small-order R:\",\n",
    "      semantic_edge_flags(b\"\\x02\" * 32, small_order_R + b\"\\x00\" * 32))\n",
    "print(\"edge flags for an ordinary sig:\",\n",
    "      semantic_edge_flags(b\"\\x02\" * 32, b\"\\x01\" * 64))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A divergence on the first input is a documented edge (note); a\n",
    "divergence on the second has no excuse (tamper). Same fail-closed\n",
    "verdict, very different alarm.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Real scale: the four proven forks, if built\n",
    "\n",
    "If you have run `pacta wallet build-quorum`, the next cell drives\n",
    "the **real** four-fork quorum: sign a payload with the dogfood\n",
    "(attested) signer, then watch all four proven verifiers agree on\n",
    "accept, and on reject for a flipped byte. If the binaries are not\n",
    "built, we say so and skip - honestly, the way the wallet itself\n",
    "fails closed.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pacta.quorum import load_quorum, binary_path\n",
    "\n",
    "built = [b for b in (\"dalek\", \"anza\", \"risc0\", \"betrusted\") if binary_path(b).exists()]\n",
    "if len(built) < 2:\n",
    "    print(\"quorum not built (need >=2). Run: pacta wallet build-quorum --sources-root <...>\")\n",
    "else:\n",
    "    import tempfile, os\n",
    "    from pacta.dogfood import locate_verifier, pem_public_key_to_raw, sign_payload_dogfood\n",
    "    from pacta.signing import generate_ed25519_keypair\n",
    "    v = locate_verifier()\n",
    "    if v is None:\n",
    "        print(\"dogfood signer not built; run pacta dogfood-build\")\n",
    "    else:\n",
    "        d = tempfile.mkdtemp()\n",
    "        key, pub = os.path.join(d, \"k.pem\"), os.path.join(d, \"k.pub\")\n",
    "        generate_ed25519_keypair(key, pub)\n",
    "        payload = b\"curriculum lecture 10 payload\"\n",
    "        sig = sign_payload_dogfood(payload, key, v)\n",
    "        pk = pem_public_key_to_raw(pub)\n",
    "        q = load_quorum(min_members=2)\n",
    "        print(\"members:\", sorted(q.members))\n",
    "        good = q.verify(payload, sig, pk)\n",
    "        print(\"valid signature   ->\", good.classification, \"accepted =\", good.accepted)\n",
    "        bad = q.verify(payload, bytes([sig[0] ^ 0xFF]) + sig[1:], pk)\n",
    "        print(\"one flipped byte  ->\", bad.classification, \"accepted =\", bad.accepted)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## The signing firewall: verify-after-sign, but proven\n",
    "\n",
    "Outbound is `intent -> sign -> firewall -> release`. The fresh\n",
    "signature faces the same quorum; only unanimity releases it. A\n",
    "rejected self-signature is *quarantined, never returned*, and\n",
    "custody latches. This is the textbook fault-injection\n",
    "countermeasure - verify a signer's output before trusting it -\n",
    "with the verifier upgraded to machine-checked code.\n",
    "\n",
    "Note the honest asymmetry: the *verify* paths are certificate-\n",
    "covered (custody-grade), but the *signing* step is trusted base -\n",
    "the attested artifact, not a third implementation. The firewall\n",
    "is exactly how we fence that weaker edge.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Two voices, one boundary (the domain split, again)\n",
    "\n",
    "Lecture 06 split provider and agent. warden inherits the split:\n",
    "\n",
    "- **The operator voice** seals the capsule: it runs the R4 gate,\n",
    "  pins the attested source commits, and stores the transparency-\n",
    "  log receipts that authorized each member.\n",
    "- **The counterparty (agent) voice** never trusts the operator's\n",
    "  adjectives. It reads the *custody card* and recomputes the\n",
    "  inclusion proofs itself - trust by recomputation.\n",
    "\n",
    "The next cell is the counterparty side: given a card, verify a\n",
    "member's inclusion proof with nothing but stdlib hashing.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Counterparty-side check of ONE member's inclusion proof.\n",
    "# (Works whenever you have a wallet + its fetched evidence; here\n",
    "#  we show the primitive the card relies on.)\n",
    "from pacta.transparency import verify_inclusion, leaf_bytes_for_attestation\n",
    "import json, glob\n",
    "\n",
    "ev = sorted(glob.glob(str(ROOT / \"examples\" / \"wallet-evidence\" / \"*.attestation.json\")))\n",
    "if not ev:\n",
    "    print(\"no bundled evidence; fetch with `pacta log-fetch` to try live\")\n",
    "else:\n",
    "    att = json.load(open(ev[0]))\n",
    "    rec = json.load(open(ev[0].replace(\".attestation.\", \".receipt.\")))\n",
    "    ok = verify_inclusion(\n",
    "        leaf_bytes_for_attestation(att),\n",
    "        rec[\"leaf_index\"], rec[\"tree_size\"],\n",
    "        [bytes.fromhex(h) for h in rec[\"inclusion_proof\"]],\n",
    "        bytes.fromhex(rec[\"sth\"][\"root_hash\"]),\n",
    "    )\n",
    "    print(f\"{att['subject']['component']}: inclusion recomputes ->\", ok)\n",
    "    print(\"The counterparty believed no adjective; it recomputed a Merkle root.\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Corrupt a member, watch the pin catch it (executable)\n",
    "\n",
    "The wallet seals each member's SHA-256 into its capsule. The\n",
    "next cell stages a COPY of a real member binary in a temp\n",
    "directory, \"seals\" its hash the way the capsule does, appends\n",
    "one byte (a supply-chain attack in miniature), and re-checks.\n",
    "Nothing on your machine is modified.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import hashlib, shutil, tempfile, pathlib\n",
    "from pacta.quorum import binary_path\n",
    "\n",
    "member = binary_path(\"dalek\")\n",
    "if not member.exists():\n",
    "    print(\"quorum not built; run pacta wallet build-quorum first\")\n",
    "else:\n",
    "    stage = pathlib.Path(tempfile.mkdtemp()) / member.name\n",
    "    shutil.copy2(member, stage)\n",
    "    sealed = hashlib.sha256(stage.read_bytes()).hexdigest()   # capsule pin\n",
    "    print(\"sealed :\", sealed[:24], \"...\")\n",
    "    with stage.open(\"ab\") as f:\n",
    "        f.write(b\"\\x00\")                                      # the attack\n",
    "    current = hashlib.sha256(stage.read_bytes()).hexdigest()\n",
    "    print(\"current:\", current[:24], \"...\")\n",
    "    if current != sealed:\n",
    "        print(\"PIN CAUGHT IT: wallet.quorum() would refuse to assemble ->\")\n",
    "        print(\"  'quorum member dalek binary hash changed since the capsule was sealed'\")\n",
    "    else:\n",
    "        print(\"impossible: SHA-256 collision\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "One appended byte and the wallet refuses to even *assemble* the\n",
    "quorum - before any verification runs. Note what this control\n",
    "is and is not: it stops binary substitution *between* wallet\n",
    "sessions; an attacker with live root outranks it (see\n",
    "docs/threat-model.md, attacker #7 - that is what the choir and\n",
    "the airgap profiles are for).\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Exercises\n",
    "\n",
    "- Change `toy_quorum` to majority voting and write two sentences\n",
    "  on exactly which attack that lets through.\n",
    "- Extend the corrupt-a-member cell: corrupt the capsule JSON\n",
    "  itself instead of the binary. What catches that, and when?\n",
    "  (Hint: nothing does until the ledger genesis is compared -\n",
    "  write down the exact trust statement the capsule hash in the\n",
    "  genesis entry provides.)\n",
    "- The signing path is trusted base. Write the strongest *true*\n",
    "  sentence you can about warden's outbound safety, and the\n",
    "  strongest *false* one a marketer would write - and name the\n",
    "  word that makes the second one false.\n",
    "- Design `warden-treasury`: which member re-verifies Solana\n",
    "  transactions, and what exactly the RPC provider is still\n",
    "  trusted for after you do.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## The human surface: see this wallet through the cockpit\n",
    "\n",
    "Everything this notebook built programmatically has a read-only human console:\n",
    "\n",
    "```\n",
    "pacta wallet cockpit --demo            # throwaway demo wallet, zero setup\n",
    "pacta wallet cockpit --wallet DIR      # the wallet you just sealed here\n",
    "```\n",
    "\n",
    "Open `/deck` for all six role stations live in parallel (the quorum bench you built is the\n",
    "indigo pane; the ledger you hash-chained is re-verified on every page load), and `/manual`\n",
    "for the lab-manual sessions that teach each role — Session 4's tamper drill breaks a *copy*\n",
    "of a ledger exactly like this notebook's and watches two independent surfaces catch it."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "name": "python",
   "pygments_lexer": "ipython3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}