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) \u21d4 decompress(R) = [k](\u2212A) + [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"
      ]
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    },
    "language_info": {
      "name": "python",
      "pygments_lexer": "ipython3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 5
}