{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# 0307 - Aeneas semantics: Rust (safe subset) -> LLBC -> pure Lean, the Result monad, and what a proof of `not P` certifies\n", "Companion to `notes/0307 - Aeneas Semantics.pdf`. Pure Python, exhaustive enumeration over `u8`; no Lean needed.\n", "The \"Rust abstract machine\" below is a small stateful interpreter (store, borrows, panics, fuel). The \"translation\" is a pure function\n", "into `Result` with explicit threading of borrowed values (backward functions). Every check compares them on the **whole** finite domain.\n", "This is a finite test of the refinement statement, not a proof of Aeneas' soundness theorem (that is a meta-theorem about the tool, see the note)." ] }, { "cell_type": "code", "execution_count": 1, "metadata": {}, "source": [ "import os\n", "os.environ[\"OMP_NUM_THREADS\"] = \"1\"\n", "import itertools, math\n", "from dataclasses import dataclass\n", "CHECKS = []\n", "def check(name, ok, detail=\"\"):\n", " CHECKS.append((name, bool(ok), detail))\n", " assert ok, (name, detail)\n", " print(f\"[ok] {name} {detail}\")" ], "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 1. The Result monad: ok / fail(e) / div\n", "`Result a = ok a | fail Error | div`. `bind` propagates `fail` and `div`. Errors we use: `integerOverflow`, `arrayOutOfBounds`, `panic`.\n", "Monad laws are checked on a finite carrier (all values of `u8`, three error tags, `div`)." ] }, { "cell_type": "code", "execution_count": 2, "metadata": {}, "source": [ "@dataclass(frozen=True)\n", "class R:\n", " tag: str # 'ok' | 'fail' | 'div'\n", " v: object = None # value for ok, error name for fail\n", "def ok(v): return R('ok', v)\n", "def fail(e): return R('fail', e)\n", "DIV = R('div')\n", "def bind(m, f):\n", " return f(m.v) if m.tag == 'ok' else m\n", "U8 = range(256)\n", "carrier = [ok(v) for v in U8] + [fail('integerOverflow'), fail('arrayOutOfBounds'), fail('panic'), DIV]\n", "f = lambda x: ok(x + 1) if x < 255 else fail('integerOverflow') # a fallible step\n", "g = lambda x: ok(x // 2) if x % 3 else fail('panic')\n", "left = all(bind(ok(v), f) == f(v) for v in U8)\n", "right = all(bind(m, ok) == m for m in carrier)\n", "assoc = all(bind(bind(m, f), g) == bind(m, lambda x: bind(f(x), g)) for m in carrier)\n", "check(\"1. monad laws on 260 values: left identity, right identity, associativity\", left and right and assoc)" ], "outputs": [ { "output_type": "stream", "name": "stdout", "text": [ "[ok] 1. monad laws on 260 values: left identity, right identity, associativity \n" ] } ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 2. Naive translation fails: integers without overflow\n", "Rust (debug / `overflow-checks`): `fn avg(a: u8, b: u8) -> u8 { (a + b) / 2 }`. The abstract machine panics iff `a + b > 255`.\n", "NAIVE translation: `a, b` as mathematical integers, no failure case. CORRECT translation: `let s <- a + b; s / 2` with checked `+` in `Result`." ] }, { "cell_type": "code", "execution_count": 3, "metadata": {}, "source": [ "def rust_avg(a, b): # abstract machine, checked arithmetic\n", " s = a + b\n", " if s > 255: return ('panic', 'integerOverflow')\n", " return ('ret', s // 2)\n", "def tr_naive_avg(a, b): return ok((a + b) // 2) # no Result\n", "def u8_add(a, b): return ok(a + b) if a + b <= 255 else fail('integerOverflow')\n", "def tr_avg(a, b): return bind(u8_add(a, b), lambda s: ok(s // 2))\n", "def obs(r): return ('ret', r.v) if r.tag == 'ok' else (('panic', r.v) if r.tag == 'fail' else ('div', None))\n", "DOM = [(a, b) for a in U8 for b in U8]\n", "bad_naive = sum(1 for a, b in DOM if obs(tr_naive_avg(a, b)) != rust_avg(a, b))\n", "bad_good = sum(1 for a, b in DOM if obs(tr_avg(a, b)) != rust_avg(a, b))\n", "panics = sum(1 for a, b in DOM if rust_avg(a, b)[0] == 'panic')\n", "check(\"2. naive translation disagrees with the machine on exactly the panicking inputs\", bad_naive == panics == 255 * 256 // 2, f\"disagree={bad_naive}, panics={panics}, 255*256/2={255*256//2}\")\n", "check(\"2. Result translation agrees with the machine on all 65536 inputs\", bad_good == 0, f\"disagree={bad_good}\")" ], "outputs": [ { "output_type": "stream", "name": "stdout", "text": [ "[ok] 2. naive translation disagrees with the machine on exactly the panicking inputs disagree=32640, panics=32640, 255*256/2=32640\n", "[ok] 2. Result translation agrees with the machine on all 65536 inputs disagree=0\n" ] } ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 3. Borrow elimination: `&mut` becomes \"return the new value\" (forward) and, when a borrow is returned, a backward function\n", "Machine: a store maps addresses to values; `&mut x` is an address. `fn bump(x: &mut u8) { *x += 1 }`.\n", "Translation: `bump (x : U8) : Result U8` returns the new value and the caller rebinds." ] }, { "cell_type": "code", "execution_count": 4, "metadata": {}, "source": [ "def machine_bump(store, addr): # returns (outcome, store')\n", " v = store[addr] + 1\n", " if v > 255: return ('panic', 'integerOverflow'), store # store unchanged when the panic happens before the write\n", " s2 = dict(store); s2[addr] = v\n", " return ('ret', None), s2\n", "def tr_bump(x): return u8_add(x, 1)\n", "good = 0\n", "for x in U8:\n", " out, st = machine_bump({0: x}, 0)\n", " r = tr_bump(x)\n", " if out[0] == 'ret': good += (r.tag == 'ok' and r.v == st[0])\n", " else: good += (r.tag == 'fail' and r.v == out[1])\n", "check(\"3a. bump: machine (store, outcome) = translation (returned value, Result) for all u8\", good == 256)\n", "# Returned mutable borrow: fn get_mut(v: &mut Vec, i: usize) -> &mut u8 ; caller: *get_mut(&mut v, i) += k\n", "def machine_incr_at(vec, i, k):\n", " heap = {j: x for j, x in enumerate(vec)} # elements are addressable places\n", " if i >= len(vec): return ('panic', 'arrayOutOfBounds'), vec\n", " v = heap[i] + k\n", " if v > 255: return ('panic', 'integerOverflow'), vec\n", " heap[i] = v\n", " return ('ret', None), [heap[j] for j in range(len(vec))]\n", "def tr_get_mut(v, i): # forward value and BACKWARD function back : u8 -> Vec\n", " if i >= len(v): return fail('arrayOutOfBounds')\n", " back = lambda new, v=v, i=i: tuple(new if j == i else x for j, x in enumerate(v))\n", " return ok((v[i], back))\n", "def tr_incr_at(v, i, k):\n", " return bind(tr_get_mut(v, i), lambda p: bind(u8_add(p[0], k), lambda new: ok(p[1](new))))\n", "cases = ties = 0\n", "for n in range(0, 4):\n", " for vec in itertools.product([0, 1, 254, 255], repeat=n):\n", " for i in range(0, 5):\n", " for k in (0, 1, 2):\n", " out, st = machine_incr_at(list(vec), i, k)\n", " r = tr_incr_at(vec, i, k)\n", " cases += 1\n", " if out[0] == 'ret': ties += (r.tag == 'ok' and list(r.v) == st)\n", " else: ties += (r.tag == 'fail' and r.v == out[1])\n", "check(\"3b. returned &mut: machine vs forward value + backward function, vec len<=3, i<=4, k<=2\", ties == cases, f\"{ties}/{cases} cases\")" ], "outputs": [ { "output_type": "stream", "name": "stdout", "text": [ "[ok] 3a. bump: machine (store, outcome) = translation (returned value, Result) for all u8 \n", "[ok] 3b. returned &mut: machine vs forward value + backward function, vec len<=3, i<=4, k<=2 1275/1275 cases\n" ] } ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 4. Loops become recursion; non-termination is `div`\n", "`fn spin(x: u8) { while x != 0 { } }`. The machine is run with a fuel bound F; `div` is what survives every F.\n", "`fn sum_to(n: u8) -> u8 { let mut s = 0; let mut i = 0; while i < n { i += 1; s += i } s }` : the accumulator overflows once n(n+1)/2 > 255, i.e. n >= 23." ] }, { "cell_type": "code", "execution_count": 5, "metadata": {}, "source": [ "def machine_spin(x, fuel):\n", " for _ in range(fuel):\n", " if x == 0: return ('ret', None)\n", " return ('fuel', None)\n", "def tr_spin(x): return ok(None) if x == 0 else DIV\n", "agree = all(((machine_spin(x, F)[0] == 'ret') == (tr_spin(x).tag == 'ok')) and ((machine_spin(x, F)[0] == 'fuel') == (tr_spin(x).tag == 'div')) for x in U8 for F in (1, 10, 1000))\n", "check(\"4a. spin: 'div' corresponds to fuel exhaustion for every tested fuel\", agree)\n", "def machine_sum_to(n):\n", " s = i = 0\n", " while i < n:\n", " i += 1; s += i\n", " if s > 255: return ('panic', 'integerOverflow')\n", " return ('ret', s)\n", "def tr_sum_loop(n, i=0, s=0): # the recursive function Aeneas would emit\n", " if i < n:\n", " i2 = i + 1\n", " r = u8_add(s, i2)\n", " return bind(r, lambda s2: tr_sum_loop(n, i2, s2))\n", " return ok(s)\n", "m = [machine_sum_to(n) for n in U8]\n", "check(\"4b. sum_to: loop = recursive translation for all n in u8\", all(obs(tr_sum_loop(n)) == m[n] for n in U8))\n", "first_bad = min(n for n in U8 if m[n][0] == 'panic')\n", "check(\"4b. first panicking n is 23 (n(n+1)/2 = 276 > 255, 22*23/2 = 253 <= 255)\", first_bad == 23 and 22 * 23 // 2 == 253 and 23 * 24 // 2 == 276, f\"first_bad={first_bad}\")" ], "outputs": [ { "output_type": "stream", "name": "stdout", "text": [ "[ok] 4a. spin: 'div' corresponds to fuel exhaustion for every tested fuel \n", "[ok] 4b. sum_to: loop = recursive translation for all n in u8 \n", "[ok] 4b. first panicking n is 23 (n(n+1)/2 = 276 > 255, 22*23/2 = 253 <= 255) first_bad=23\n" ] } ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 5. A proof of `not P`: closed proposition versus negation under binders\n", "Property of `avg`: `P(a,b,r)`: `min(a,b) <= r <= max(a,b)`, evaluated on the translated function, with `Q(a,b) := exists r, avg a b = ok r /\\ P(a,b,r)`.\n", "Closed statement S := forall a b, H a b -> Q a b (H trivially true here). Its negation not S <=> exists a b, H a b /\\ not Q a b.\n", "Different, much stronger statement S' := forall a b, H a b -> not Q a b. Under binders, `not Q a b` for a free (a,b) is a claim about *every* instance." ] }, { "cell_type": "code", "execution_count": 6, "metadata": {}, "source": [ "def Q(a, b):\n", " r = tr_avg(a, b)\n", " return r.tag == 'ok' and min(a, b) <= r.v <= max(a, b)\n", "H = lambda a, b: True\n", "S = all((not H(a, b)) or Q(a, b) for a, b in DOM)\n", "notS = any(H(a, b) and not Q(a, b) for a, b in DOM)\n", "Sprime = all((not H(a, b)) or (not Q(a, b)) for a, b in DOM)\n", "witnesses = [(a, b) for a, b in DOM if H(a, b) and not Q(a, b)]\n", "check(\"5a. S is false and (not S) holds, certified by one witness\", (not S) and notS and witnesses[0] == (1, 255), f\"|W|={len(witnesses)}, first witness {witnesses[0]}\")\n", "check(\"5a. the 'under binders' strengthening S' (forall x, not Q x) is false: Q holds at (0,0)\", (not Sprime) and Q(0, 0), f\"|{{Q}}|={len(DOM)-len(witnesses)}\")\n", "check(\"5a. |W| = number of panicking inputs = 32640\", len(witnesses) == 32640)\n", "# With a precondition H a b := a + b <= 255 the closed statement is TRUE and its negation is not provable\n", "H2 = lambda a, b: a + b <= 255\n", "check(\"5b. with H := a+b<=255, S_H holds (so no certificate for not S_H exists)\", all((not H2(a, b)) or Q(a, b) for a, b in DOM))\n", "# Vacuity: if H is unsatisfiable, the under-binders form is provable and yet there is no witness at all\n", "H3 = lambda a, b: a + b > 255 and a + b < 256\n", "check(\"5c. unsatisfiable H: S'_H = forall x, H x -> not Q x holds vacuously, but the witness set is empty (no bug evidence)\",\n", " all((not H3(a, b)) or (not Q(a, b)) for a, b in DOM) and not any(H3(a, b) for a, b in DOM))\n", "# And with H satisfiable and S'_H true (every H-input fails): S'_H implies the closed negation\n", "H4 = lambda a, b: a + b >= 256\n", "check(\"5c. satisfiable H := a+b>=256: S'_H holds (every such input fails) and so implies not S_H\", all((not H4(a, b)) or (not Q(a, b)) for a, b in DOM) and any(H4(a, b) for a, b in DOM))" ], "outputs": [ { "output_type": "stream", "name": "stdout", "text": [ "[ok] 5a. S is false and (not S) holds, certified by one witness |W|=32640, first witness (1, 255)\n", "[ok] 5a. the 'under binders' strengthening S' (forall x, not Q x) is false: Q holds at (0,0) |{Q}|=32896\n", "[ok] 5a. |W| = number of panicking inputs = 32640 \n", "[ok] 5b. with H := a+b<=255, S_H holds (so no certificate for not S_H exists) \n", "[ok] 5c. unsatisfiable H: S'_H = forall x, H x -> not Q x holds vacuously, but the witness set is empty (no bug evidence) \n", "[ok] 5c. satisfiable H := a+b>=256: S'_H holds (every such input fails) and so implies not S_H \n" ] } ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 6. Reachability: the certificate is about `avg`, not about the program that calls it\n", "Public caller: `fn mean_pct(a: u8, b: u8) -> u8 { avg(min(a, c), min(b, c)) }` for a clamp `c` (a property of the *caller*).\n", "Reach_c = {(a', b'): a', b' <= c}. Witnesses W from section 5 are real bugs of the program only if W intersect Reach_c is non-empty." ] }, { "cell_type": "code", "execution_count": 7, "metadata": {}, "source": [ "def count_formula(c): # pairs in [0,c]^2 with a+b >= 256\n", " return sum(max(0, min(c, 255) - max(0, 256 - a) + 1) for a in range(0, min(c, 255) + 1))\n", "rows = []\n", "for c in (50, 100, 127, 128, 150, 200, 255):\n", " brute = sum(1 for a, b in witnesses if a <= c and b <= c)\n", " rows.append((c, brute, count_formula(c)))\n", " assert brute == count_formula(c)\n", " print(f\" clamp c={c:3d}: |W \u2229 Reach_c| = {brute:6d} (closed form {count_formula(c)})\")\n", "check(\"6. |W \u2229 Reach_c| equals the closed form for all tested c\", all(b == f for _, b, f in rows))\n", "check(\"6. c <= 127: certificate holds for avg yet the program mean_pct never panics (false alarm)\", dict((c, b) for c, b, _ in rows)[127] == 0 and dict((c, b) for c, b, _ in rows)[128] == 1, \"c=127 -> 0, c=128 -> 1 (128+128=256)\")\n", "# end-to-end: run the caller on its whole input domain\n", "def mean_pct(a, b, c): return tr_avg(min(a, c), min(b, c))\n", "for c in (100, 127, 128, 200):\n", " panics_c = sum(1 for a, b in DOM if mean_pct(a, b, c).tag == 'fail')\n", " print(f\" mean_pct with clamp {c}: panicking inputs over all 65536 raw (a,b): {panics_c}\")\n", "check(\"6. the caller with c=127 never fails on any of 65536 raw inputs; with c=128 it fails (a,b >= 128)\", all(mean_pct(a, b, 127).tag == 'ok' for a, b in DOM) and sum(1 for a, b in DOM if mean_pct(a, b, 128).tag == 'fail') == 128 * 128)" ], "outputs": [ { "output_type": "stream", "name": "stdout", "text": [ " clamp c= 50: |W \u2229 Reach_c| = 0 (closed form 0)\n", " clamp c=100: |W \u2229 Reach_c| = 0 (closed form 0)\n", " clamp c=127: |W \u2229 Reach_c| = 0 (closed form 0)\n", " clamp c=128: |W \u2229 Reach_c| = 1 (closed form 1)\n", " clamp c=150: |W \u2229 Reach_c| = 1035 (closed form 1035)\n", " clamp c=200: |W \u2229 Reach_c| = 10585 (closed form 10585)\n", " clamp c=255: |W \u2229 Reach_c| = 32640 (closed form 32640)\n", "[ok] 6. |W \u2229 Reach_c| equals the closed form for all tested c \n", "[ok] 6. c <= 127: certificate holds for avg yet the program mean_pct never panics (false alarm) c=127 -> 0, c=128 -> 1 (128+128=256)\n", " mean_pct with clamp 100: panicking inputs over all 65536 raw (a,b): 0\n", " mean_pct with clamp 127: panicking inputs over all 65536 raw (a,b): 0\n", " mean_pct with clamp 128: panicking inputs over all 65536 raw (a,b): 16384\n", " mean_pct with clamp 200: panicking inputs over all 65536 raw (a,b): 29560\n", "[ok] 6. the caller with c=127 never fails on any of 65536 raw inputs; with c=128 it fails (a,b >= 128) \n" ] } ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 7. Translation soundness: wrong translations give wrong certificates in both directions\n", "* U1 (too lax): unbounded integers. Then `S` is *provable*, the real program violates it: a missed bug.\n", "* U2 (wrapping, release-mode semantics): `Q` fails on the same inputs but as a wrong value, and panic-freedom becomes provable. \"The bug\" depends on which Rust semantics is modelled.\n", "* U3 (too strict): checks overflow on the *widened* sum `(a as u16 + b as u16)/2`, which is fine in Rust, but translates the u16 add as a u8 add: spurious `fail`, a false certificate." ] }, { "cell_type": "code", "execution_count": 8, "metadata": {}, "source": [ "def Q_with(trf): return lambda a, b: (lambda r: r.tag == 'ok' and min(a, b) <= r.v <= max(a, b))(trf(a, b))\n", "trU1 = tr_naive_avg\n", "trU2 = lambda a, b: ok(((a + b) % 256) // 2)\n", "def rust_wide(a, b): return ('ret', (a + b) // 2) # `((a as u16 + b as u16) / 2) as u8` never panics\n", "trU3 = lambda a, b: bind(u8_add(a, b), lambda s: ok(s // 2)) # wrong: widened add checked at u8\n", "Wset = lambda q: sum(1 for a, b in DOM if not q(a, b))\n", "print(\" failing sets |{not Q}|: truth(avg, checked) =\", Wset(Q), \", U1 (unbounded) =\", Wset(Q_with(trU1)), \", U2 (wrapping) =\", Wset(Q_with(trU2)))\n", "check(\"7a. U1: S is provable under the unbounded translation (no witness), but the program panics on 32640 inputs\", Wset(Q_with(trU1)) == 0 and Wset(Q) == 32640)\n", "pf_checked = all(tr_avg(a, b).tag == 'ok' for a, b in DOM); pf_wrap = all(trU2(a, b).tag == 'ok' for a, b in DOM)\n", "same_set = {(a, b) for a, b in DOM if not Q_with(trU2)(a, b)} == set(witnesses)\n", "check(\"7b. U2: wrapping gives the SAME 32640 witnesses for P but for a different reason (wrong value, not panic); panic-freedom is provable under U2 and false under checked semantics\",\n", " same_set and pf_wrap and not pf_checked, f\"|W_wrap|={Wset(Q_with(trU2))}, panic-free: wrap={pf_wrap}, checked={pf_checked}\")\n", "# wide version of the program: truth never panics, so S_wide holds; the strict translation certifies a bug\n", "Qwide_true = lambda a, b: min(a, b) <= rust_wide(a, b)[1] <= max(a, b)\n", "check(\"7c. U3: program `avg_wide` is correct on all inputs; the over-strict translation yields 32640 spurious witnesses\", all(Qwide_true(a, b) for a, b in DOM) and Wset(Q_with(trU3)) == 32640)" ], "outputs": [ { "output_type": "stream", "name": "stdout", "text": [ " failing sets |{not Q}|: truth(avg, checked) = 32640 , U1 (unbounded) = 0 , U2 (wrapping) = 32640\n", "[ok] 7a. U1: S is provable under the unbounded translation (no witness), but the program panics on 32640 inputs \n", "[ok] 7b. U2: wrapping gives the SAME 32640 witnesses for P but for a different reason (wrong value, not panic); panic-freedom is provable under U2 and false under checked semantics |W_wrap|=32640, panic-free: wrap=True, checked=False\n", "[ok] 7c. U3: program `avg_wide` is correct on all inputs; the over-strict translation yields 32640 spurious witnesses \n" ] } ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Summary" ] }, { "cell_type": "code", "execution_count": 9, "metadata": {}, "source": [ "for n, ok_, d in CHECKS: print((\"PASS \" if ok_ else \"FAIL \") + n)\n", "assert all(c[1] for c in CHECKS)\n", "print(f\"\\n{len(CHECKS)} checks, all passed\")" ], "outputs": [ { "output_type": "stream", "name": "stdout", "text": [ "PASS 1. monad laws on 260 values: left identity, right identity, associativity\n", "PASS 2. naive translation disagrees with the machine on exactly the panicking inputs\n", "PASS 2. Result translation agrees with the machine on all 65536 inputs\n", "PASS 3a. bump: machine (store, outcome) = translation (returned value, Result) for all u8\n", "PASS 3b. returned &mut: machine vs forward value + backward function, vec len<=3, i<=4, k<=2\n", "PASS 4a. spin: 'div' corresponds to fuel exhaustion for every tested fuel\n", "PASS 4b. sum_to: loop = recursive translation for all n in u8\n", "PASS 4b. first panicking n is 23 (n(n+1)/2 = 276 > 255, 22*23/2 = 253 <= 255)\n", "PASS 5a. S is false and (not S) holds, certified by one witness\n", "PASS 5a. the 'under binders' strengthening S' (forall x, not Q x) is false: Q holds at (0,0)\n", "PASS 5a. |W| = number of panicking inputs = 32640\n", "PASS 5b. with H := a+b<=255, S_H holds (so no certificate for not S_H exists)\n", "PASS 5c. unsatisfiable H: S'_H = forall x, H x -> not Q x holds vacuously, but the witness set is empty (no bug evidence)\n", "PASS 5c. satisfiable H := a+b>=256: S'_H holds (every such input fails) and so implies not S_H\n", "PASS 6. |W \u2229 Reach_c| equals the closed form for all tested c\n", "PASS 6. c <= 127: certificate holds for avg yet the program mean_pct never panics (false alarm)\n", "PASS 6. the caller with c=127 never fails on any of 65536 raw inputs; with c=128 it fails (a,b >= 128)\n", "PASS 7a. U1: S is provable under the unbounded translation (no witness), but the program panics on 32640 inputs\n", "PASS 7b. U2: wrapping gives the SAME 32640 witnesses for P but for a different reason (wrong value, not panic); panic-freedom is provable under U2 and false under checked semantics\n", "PASS 7c. U3: program `avg_wide` is correct on all inputs; the over-strict translation yields 32640 spurious witnesses\n", "\n", "20 checks, all passed\n" ] } ] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "name": "python", "version": "3.12.11" } }, "nbformat": 4, "nbformat_minor": 4 }