#!/usr/bin/env python3 """ MIZAN starter kit: CODE GATE / ESCROW RELEASE --------------------------------------------- Hold a payment. Release it only on a signed ALLOW against a pinned rule. 1. read the rule you pin GET /api/requirements/{ref} (its exact hash) 2. hold + release in the sandbox POST /api/sandbox/release-demo {requirement, requirement_hash, payee} 3. verify the signed answer offline, Ed25519 over header.payload 4. release only if: authorization, ALLOW, same payee, same rule, same hold Live layers: published rules, release guard (hold -> release), signed authorization or signed denial, public log leaves. Sandbox: the hold is real, NO MONEY MOVES. For your own holds: POST /api/release/holds (with your key), then /release. Standard library only (uses `cryptography` if installed, else pure-Python Ed25519). run: python3 escrow_release.py python3 escrow_release.py some-payee.example.com MIZAN_RULE=req_xxx@1 python3 escrow_release.py (pin your own published rule) """ # --------------------------------------------------------------------------- # The MIZAN plumbing: HTTP + offline Ed25519 verification. Copy as-is. # --------------------------------------------------------------------------- import base64, hashlib, json, os, sys, time, urllib.error, urllib.request API = os.environ.get("MIZAN_API", "https://studio.mizan.market") KEYS_URL = API + "/.well-known/mizan-keys.json" def call(path, body=None, key=None): """GET (no body) or POST JSON to the live API. Returns the parsed JSON reply.""" headers = {"Content-Type": "application/json", "User-Agent": "mizan-starter-kit/1"} if key: headers["Authorization"] = "Bearer " + key data = None if body is None else json.dumps(body).encode() req = urllib.request.Request(API + path, data=data, headers=headers, method="POST" if data else "GET") try: with urllib.request.urlopen(req, timeout=30) as r: return json.load(r) except urllib.error.HTTPError as e: # MIZAN answers errors in JSON too return json.load(e) def b64u(s): return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4)) def _ed25519_pure(pub, msg, sig): """RFC 8032 Ed25519 verify in pure Python (used only if `cryptography` is missing).""" q = 2**255 - 19 L = 2**252 + 27742317777372353535851937790883648493 d = -121665 * pow(121666, q - 2, q) % q def add(P, Q): A = (P[1] - P[0]) * (Q[1] - Q[0]) % q B = (P[1] + P[0]) * (Q[1] + Q[0]) % q C = 2 * P[3] * Q[3] * d % q D = 2 * P[2] * Q[2] % q E, F, G, H = B - A, D - C, D + C, B + A return (E * F % q, G * H % q, F * G % q, E * H % q) def mul(s, P): Q = (0, 1, 1, 0) while s: if s & 1: Q = add(Q, P) P, s = add(P, P), s >> 1 return Q def point(b): y = int.from_bytes(b, "little") sign, y = y >> 255, y & ((1 << 255) - 1) if y >= q: return None x2 = (y * y - 1) * pow(d * y * y + 1, q - 2, q) % q x = pow(x2, (q + 3) // 8, q) if (x * x - x2) % q: x = x * pow(2, (q - 1) // 4, q) % q if (x * x - x2) % q or (x == 0 and sign): return None if (x & 1) != sign: x = q - x return (x, y, 1, x * y % q) if len(pub) != 32 or len(sig) != 64: return False A, R, s = point(pub), point(sig[:32]), int.from_bytes(sig[32:], "little") if A is None or R is None or s >= L: return False B = point((4 * pow(5, q - 2, q) % q).to_bytes(32, "little")) h = int.from_bytes(hashlib.sha512(sig[:32] + pub + msg).digest(), "little") % L lhs, rhs = mul(s, B), add(R, mul(h, A)) return (lhs[0] * rhs[2] - rhs[0] * lhs[2]) % q == 0 and (lhs[1] * rhs[2] - rhs[1] * lhs[2]) % q == 0 def ed25519_verify(pub, msg, sig): try: from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey except ImportError: return _ed25519_pure(pub, msg, sig) try: Ed25519PublicKey.from_public_bytes(pub).verify(sig, msg) return True except Exception: return False def load_keys(): """Fetch the key set once and pin it. After this, verifying needs no call to MIZAN.""" ks = call("/.well-known/mizan-keys.json") return {k["key_id"]: base64.b64decode(k["public_key_b64"]) for k in ks["keys"] if k.get("alg") == "Ed25519"} def verify_token(token, keys): """Offline. A verdict token is mzv1.., Ed25519 over 'mzv1.' + claims. A release token is
.., Ed25519 over 'header.payload'. Returns the claims, or raises ValueError.""" parts = token.split(".") if len(parts) != 3: raise ValueError("malformed") if parts[0] == "mzv1": claims, signed = json.loads(b64u(parts[1])), ("mzv1." + parts[1]).encode() kid = claims.get("kid") else: header, claims = json.loads(b64u(parts[0])), json.loads(b64u(parts[1])) signed, kid = (parts[0] + "." + parts[1]).encode(), header.get("kid") if kid not in keys: raise ValueError("unknown_kid") if not ed25519_verify(keys[kid], signed, b64u(parts[2])): raise ValueError("bad_signature") if claims.get("exp", 0) <= time.time(): raise ValueError("expired") return claims def get_key(name): """Your MIZAN key: $MIZAN_KEY if set, else one call claims one (no account, no form).""" if os.environ.get("MIZAN_KEY"): return os.environ["MIZAN_KEY"] body = {"name": name} if os.environ.get("MIZAN_TEST"): # marks a test identity, kept out of public counts body["test"] = True d = call("/api/agents/claim", body) if not d.get("api_key"): sys.exit("could not claim a key: %s" % d) print("claimed a key for %r (set MIZAN_KEY=... to reuse it)" % name) return d["api_key"] def check(subject, rule=None): from urllib.parse import quote q = "?requirement=" + quote(rule, safe="@") if rule else "" return call("/api/check/" + quote(subject, safe=":/") + q) # --------------------------------------------------------------------------- # Your app. Change these. # --------------------------------------------------------------------------- RULE = os.environ.get("MIZAN_RULE", "req_b68b0bb66d46@1") # a public rule; pin your own PAYEES = sys.argv[1:] or ["try-me", "unknown-payee.example.com"] AMOUNT_USD = 2500.00 def main(): r = call("/api/requirements/" + RULE) if not r.get("hash"): sys.exit("rule not found: %s" % RULE) print("pinned rule %s\n %s\n hash %s" % (RULE, r.get("shown_back"), r["hash"])) keys = load_keys() for payee in PAYEES: print("\nHOLD USD %s for %s (your escrow, simulated)" % (format(AMOUNT_USD, ",.2f"), payee)) d = call("/api/sandbox/release-demo", {"requirement": RULE, "requirement_hash": r["hash"], "payee": payee}) if not d.get("ok"): print(" error: %s" % (d.get("errors") or d.get("error"))) continue try: claims = verify_token(d["token"], keys) except (KeyError, ValueError) as e: print(" token did not verify (%s): keep holding" % e) continue req = claims.get("requirement") or {} release = (claims.get("typ") == "mizan-release-authorization/1" and claims.get("decision") == "ALLOW" and claims.get("payee") == payee and req.get("ref") == RULE and req.get("hash") == r["hash"] and claims.get("hold_id") == d["hold"]["hold_id"]) shown = {k: claims.get(k) for k in ("typ", "decision", "payee", "hold_id", "requirement", "evidence_rung", "single_use", "status", "reason", "jti", "exp") if claims.get(k) is not None} print(" hold %s (log leaf %s)" % (d["hold"]["hold_id"], d["hold"]["leaf"]["index"])) print(" signed %s, verified offline:" % d.get("token_kind")) print(" " + json.dumps(shown, indent=2).replace("\n", "\n ")) print(" token: %s..." % d["token"][:60]) if release: print(" -> RELEASE USD %s to %s (simulated: redeem the authorization once)" % (format(AMOUNT_USD, ",.2f"), payee)) else: print(" -> STILL HELD. Your escrow keeps the funds.") for g in (d.get("release") or {}).get("gaps") or []: print(" what would pass: %s" % g.get("step")) print("\nmoney moved: none (sandbox)") if __name__ == "__main__": main()