# -*- coding: utf-8 -*- """ ots_verify.py — verify an OpenTimestamps proof by WALKING it, not by reading its header. WHY THIS EXISTS. commitment_gate.py used to establish "the attestation covers this file" by reading the 32 bytes after the proof's magic header. A red-teamer pointed out that those 32 bytes are a local file an attacker edits with one `sed`, and that `ots info` reads the same header — so the corroboration corroborated the forgery. He then did it: rewrote a registered prediction's threshold, patched the commitment file, patched 32 bytes of the .ots, and got BUILD PASSED with the gate printing a fabricated attestation line. That reduced every downstream "frozen" guarantee to a comment. WHAT A PROOF ACTUALLY IS. A timestamp is a merkle path: start at the file's sha256, apply a recorded sequence of prepends, appends and hashes, and arrive at a digest that IS the merkle root of a Bitcoin block. The header digest is a convenience label. The path is the evidence. Forging the label is one sed; forging the path means finding a preimage chain that lands on a real block's merkle root. WHAT THIS DOES, and does not: * Parses the proof from the OpenTimestamps serialization — no `ots` binary needed. * Walks every branch of the path from the file's own sha256. * For each BitcoinBlockHeaderAttestation, fetches that block's merkle root from INDEPENDENT public explorers and requires the walked digest to equal it. Two sources are queried and required to agree with each other, so one compromised explorer is not enough. * Reports PendingAttestations honestly as pending — a calendar promise, not a block. * It does NOT run a Bitcoin node, so the block header itself is taken from explorers rather than from proof-of-work this machine validated. That is a real, stated limit: it moves the trust from "a local file nobody checks" to "two independent public indexes of the chain agreeing", which is not the same as running a node. Run: python3 scripts/ots_verify.py timestamp/COMMITMENT.txt.ots """ import hashlib, json, os, sys, urllib.request MAGIC = b"\x00OpenTimestamps\x00\x00Proof\x00\xbf\x89\xe2\xe8\x84\xe8\x92\x94" ATTESTATION = 0x00 FORK = 0xff BITCOIN_TAG = b"\x05\x88\x96\x0d\x73\xd7\x19\x01" PENDING_TAG = b"\x83\xdf\xe3\x0d\x2e\xf9\x0c\x8e" EXPLORERS = [ ("blockstream.info", "https://blockstream.info/api/block-height/%d", "https://blockstream.info/api/block/%s"), ("mempool.space", "https://mempool.space/api/block-height/%d", "https://mempool.space/api/block/%s"), ] class Reader: def __init__(self, b): self.b, self.i = b, 0 def byte(self): v = self.b[self.i]; self.i += 1; return v def bytes(self, n): v = self.b[self.i:self.i + n] if len(v) != n: raise ValueError("truncated proof") self.i += n; return v def varuint(self): v, shift = 0, 0 while True: b = self.byte() v |= (b & 0x7f) << shift if not b & 0x80: return v shift += 7 def varbytes(self): return self.bytes(self.varuint()) def done(self): return self.i >= len(self.b) # Unary and binary operations, by opcode. The proof records them; we apply them. def _op(code, r, msg): if code == 0xf0: # append return msg + r.varbytes() if code == 0xf1: # prepend return r.varbytes() + msg if code == 0x02: # reverse return msg[::-1] if code == 0x03: # hexlify return msg.hex().encode() if code == 0x08: return hashlib.sha256(msg).digest() if code == 0x67: h = hashlib.new("ripemd160"); h.update(msg); return h.digest() if code == 0x11: return hashlib.sha1(msg).digest() raise ValueError("unknown opcode 0x%02x" % code) def walk(r, msg, out): """Depth-first through the timestamp tree, collecting (kind, detail, digest).""" while True: if r.done(): return tag = r.byte() if tag == ATTESTATION: t = r.bytes(8) payload = Reader(r.varbytes()) if t == BITCOIN_TAG: out.append(("bitcoin", payload.varuint(), msg)) elif t == PENDING_TAG: out.append(("pending", payload.varbytes().decode("utf8", "replace"), msg)) else: out.append(("unknown", t.hex(), msg)) return if tag == FORK: # A fork duplicates the current message down each branch. Recurse on the # first, continue the loop on the second -- the serialization writes one # branch fully, then the rest. walk(r, msg, out) continue msg = _op(tag, r, msg) def parse(path): b = open(path, "rb").read() if not b.startswith(MAGIC): raise SystemExit("%s: not an OpenTimestamps proof" % os.path.basename(path)) r = Reader(b[len(MAGIC):]) version = r.varuint() op = r.byte() if op != 0x08: raise SystemExit("%s: attests over digest op 0x%02x, not sha256" % (os.path.basename(path), op)) file_digest = r.bytes(32) out = [] walk(r, file_digest, out) return version, file_digest, out def block_merkle_root(height, timeout=15): """(merkle_root, block_unix_time), from two independent explorers that must agree. The block TIME is the cryptographic defence against re-stamping. A valid proof made today over rewritten content reaches a block mined today; requiring the anchored block to predate the freeze deadline means rewritten content cannot be anchored to a block that predates the rewrite, and forging a path into an old block is the preimage problem. A red-teamer showed the pin alone (ANCHORS.txt) is movable; the block time is not.""" got = {} for name, hurl, burl in EXPLORERS: try: h = urllib.request.urlopen(hurl % height, timeout=timeout).read().decode().strip() d = json.load(urllib.request.urlopen(burl % h, timeout=timeout)) got[name] = (h, d["merkle_root"], int(d["timestamp"])) except Exception as e: got[name] = ("error", "%s: %s" % (type(e).__name__, e), 0) ok = {k: v for k, v in got.items() if v[0] != "error"} if len(ok) < 2: return None, None, got roots = {v[1] for v in ok.values()} hashes = {v[0] for v in ok.values()} times = {v[2] for v in ok.values()} if len(roots) != 1 or len(hashes) != 1: raise SystemExit("BLOCK EXPLORERS DISAGREE about block %d: %r — refusing to " "certify anything on that basis" % (height, got)) return list(roots)[0], min(times), got def verify(path, offline=False): """Returns (ok, lines, state, block_time). ok is False only on a PROVEN failure. block_time is the unix time of the verified Bitcoin block, or None.""" try: version, digest, atts = parse(path) except ValueError as e: # A truncated or malformed proof (unknown opcode, short read) must fail the # verification cleanly, not crash the gate with a traceback. A compromised calendar # returning garbage is a verification failure like any other — fail closed. return False, ["the proof is malformed and could not be parsed: %s" % e], "malformed", None target = path[:-4] if path.endswith(".ots") else None lines, state, proven, block_time = [], "unknown", True, None if target and os.path.exists(target): actual = hashlib.sha256(open(target, "rb").read()).hexdigest() if actual != digest.hex(): return False, ["the proof attests to %s but the file hashes to %s" % (digest.hex(), actual)], "mismatch", None lines.append("attests over %s (sha256 matches the file)" % digest.hex()[:16]) for kind, detail, msg in atts: if kind == "bitcoin": if offline: lines.append("Bitcoin block %s — NOT CHECKED (offline)" % detail) if state != "confirmed": state = "unchecked" continue root, btime, got = block_merkle_root(detail) if root is None: lines.append("Bitcoin block %s — could not reach two explorers: %r" % (detail, got)) if state != "confirmed": state = "unchecked" continue # OTS carries the merkle root in internal byte order; explorers display it # reversed. Compare in one order and say which. if msg[::-1].hex() == root: lines.append("VERIFIED against Bitcoin block %s (mined %s UTC) — the walked " "merkle path equals that block's merkle root, confirmed by %s" % (detail, _iso(btime), " and ".join( k for k, v in got.items() if v[0] != "error"))) state = "confirmed" block_time = btime else: proven = False lines.append("FORGED OR CORRUPT: the path walks to %s, but block %s has " "merkle root %s" % (msg[::-1].hex(), detail, root)) state = "forged" elif kind == "pending": lines.append("pending at %s — a calendar's promise, not a block" % detail) if state != "confirmed": state = "pending" else: lines.append("unknown attestation type %s" % detail) if not atts: proven = False lines.append("the proof carries NO attestation at all") state = "empty" return proven, lines, state, block_time def _iso(t): import datetime return datetime.datetime.utcfromtimestamp(t).strftime("%Y-%m-%d %H:%M") if t else "?" if __name__ == "__main__": args = [a for a in sys.argv[1:] if not a.startswith("-")] off = "--offline" in sys.argv bad = 0 for a in args: ok, lines, state, _bt = verify(a, offline=off) print("%s [%s]" % (os.path.basename(a), state)) for l in lines: print(" " + l) if not ok: bad = 1 sys.exit(bad)