"""Formal model of NIP-PL acceptance + lifecycle (PLANS/NIP_PL_PUSH_LEASES_DRAFT.md).
Exhaustive finite-state exploration of ONE lease address (author, 30350, d) under
an adversary who can present a bounded universe of candidate events -- including
forged (generation, created_at) combinations and replays -- in every order.
We check the safety/lifecycle invariants the spec ASSERTS as atomic/monotone:
I1 no-resurrection : once a tombstone (active:false) is the effective state,
no later ACCEPTED event may make the address effective-active
unless it strictly beats the tombstone on BOTH orderings.
(spec Acceptance check 8 + Lifecycle "replayed older event
can never resurrect a revoked lease")
I2 watermark-monotone : the persisted generation watermark never decreases, and
a REJECTED event never changes stored/effective/watermark.
(check 8: "leave stored event, effective push state, and
watermark all unchanged")
I3 no-watermark-poison : a high-generation / old-created_at event that LOSES the
NIP-01 ordering is rejected and MUST NOT raise the watermark.
(check 8 trap the spec calls out by name)
I4 dual-order-agree : the accepted (stored) event and the effective push state are
always the same event -- REQ view never disagrees with effect.
I5 replay-window : after natural expiry / tombstone retention, any replay of a
formerly-valid event fails (expiration lower bound), so the
watermark can be released without reopening resurrection.
Modeled acceptance sequence (spec "Acceptance and Origin Binding", ordered):
a candidate is ACCEPTED iff it passes structural checks (we assume the adversary
only ever submits structurally valid, correctly-signed, origin-bound events -- we
are testing ORDERING, not parsing) AND wins check 8:
(a) NIP-01 addressable ordering vs current stored winner:
greater created_at, tie -> lexically-lowest id ;
(b) generation strictly greater than the internal watermark.
BOTH required. Failing either -> reject, no state change.
On accept: commit (stored, effective, watermark) atomically; watermark := gen.
"""
from itertools import permutations
class Ev:
__slots__ = ("id", "gen", "created", "active")
def __init__(self, eid, gen, created, active):
self.id, self.gen, self.created, self.active = eid, gen, created, active
def __repr__(self):
s = "A" if self.active else "T" # active / tombstone
return f"{self.id}[g{self.gen},c{self.created},{s}]"
def nip01_beats(cand, cur):
"""NIP-01 addressable ordering: higher created_at; tie -> lexically LOWEST id."""
if cur is None:
return True
if cand.created != cur.created:
return cand.created > cur.created
return cand.id < cur.id # lower id wins the tie
class Address:
"""One (author,30350,d). Faithful encoding of acceptance check 8."""
def __init__(self):
self.stored = None # currently-stored winning event (what REQ serves)
self.effective_active = False # effective push state: matching on?
self.watermark = -1 # internal generation watermark
self.wm_history = [-1] # to check monotonicity
self.log = [] # (event, accepted?)
def submit(self, ev):
# check 8: must win BOTH orderings
wins_nip01 = nip01_beats(ev, self.stored)
wins_gen = ev.gen > self.watermark
if wins_nip01 and wins_gen:
# atomic commit
self.stored = ev
self.effective_active = ev.active
self.watermark = ev.gen
self.wm_history.append(self.watermark)
self.log.append((ev, True))
return True
else:
# MUST leave stored, effective, watermark unchanged
self.log.append((ev, False))
return False
def explore():
# Adversarial candidate universe for ONE address.
# ids chosen so we can force NIP-01 ties (same created, different id).
# Includes: an active lease, a higher-gen tombstone (legit revoke),
# a replayed OLD active event with a FORGED high generation (poison attempt),
# a same-created_at tie pair, and a stale low-gen active (resurrection attempt).
universe = [
Ev("e1", gen=1, created=100, active=True), # initial active lease
Ev("e2", gen=2, created=200, active=False), # legit revocation (tombstone)
Ev("e3", gen=9, created=150, active=True), # POISON: high gen, but created_at
# < tombstone e2 -> loses NIP-01
Ev("e4", gen=3, created=250, active=True), # legit reactivation (beats both)
Ev("e5", gen=1, created=100, active=True), # exact replay of e1 (stale both)
Ev("a1", gen=5, created=200, active=True), # NIP-01 tie with e2 (created=200);
# id "a1" < "e2" -> a1 wins NIP-01
Ev("z1", gen=0, created=300, active=True), # clause-(b) witness: highest
# created_at, STALE gen -> only
# the watermark rejects it
]
viol = {k: [] for k in ("I1", "I2", "I3", "I4", "I5")}
n = 0
# exhaust every ordering of every non-empty subset up to full universe.
# full permutation of all 6 = 720; we also test all shorter prefixes via
# permutations of the whole set (prefix coverage) -- and specifically every
# ordering that ends after a tombstone to probe resurrection.
from itertools import permutations as P
for perm in P(universe):
n += 1
addr = Address()
tomb_seen_effective = False
for ev in perm:
wm_before = addr.watermark
stored_before = addr.stored
eff_before = addr.effective_active
accepted = addr.submit(ev)
# I2: rejected event changes nothing
if not accepted:
if (addr.watermark != wm_before or addr.stored is not stored_before
or addr.effective_active != eff_before):
viol["I2"].append((perm, ev, "rejected event mutated state"))
# I2 (mono): watermark never decreases
if addr.watermark < wm_before:
viol["I2"].append((perm, ev, "watermark decreased"))
# I3: an event that LOSES nip01 but has high gen must NOT raise watermark
if not nip01_beats(ev, stored_before) and ev.gen > wm_before:
if addr.watermark != wm_before:
viol["I3"].append((perm, ev, "watermark poisoned by nip01-loser"))
# I4: stored event == effective source (never disagree)
if addr.stored is not None:
if addr.effective_active != addr.stored.active:
viol["I4"].append((perm, ev, "stored/effective disagree"))
if addr.effective_active is False and addr.stored is not None \
and not addr.stored.active:
tomb_seen_effective = True
# I1: once effective state is a tombstone, resurrection requires beating
# BOTH orderings. Detect: we were tombstoned, then became active.
if tomb_seen_effective and addr.effective_active:
# legitimate only if the reactivating event beat the tombstone on both.
# e4 (gen3,created250) is the only legit reactivator here.
if not (accepted and ev is not None and ev.active):
viol["I1"].append((perm, ev, "spurious resurrection"))
# deeper: the event that flipped us active must out-order the last
# tombstone on NIP-01 AND gen. addr.stored is that event.
# (structurally guaranteed by submit(); assert it held)
tomb_seen_effective = addr.stored.active is False # reset guard
# I5: replay-window release. Model: after retention, watermark may be dropped to
# a floor F. Any replayed event with created <= expiry_floor is rejected by the
# expiration lower bound (now - skew < expiration). We check that dropping the
# watermark to F does NOT let e5 (the stale replay) resurrect, BECAUSE e5 also
# fails NIP-01 vs the last stored tombstone. Encode as: even watermark=-1 (fully
# released) + expiration gate blocks e5.
for reset_wm in (-1, 0, 1):
addr = Address()
addr.submit(Ev("e1", 1, 100, True))
addr.submit(Ev("e2", 2, 200, False)) # tombstone stored, created=200
addr.watermark = reset_wm # simulate retention release
# expiration gate: replay is only accepted if its created_at still beats
# the stored tombstone on NIP-01 (created 100 < 200 -> loses regardless of wm)
before = (addr.stored, addr.effective_active)
addr.submit(Ev("e5", 1, 100, True)) # the replay
if addr.effective_active and not before[1]:
viol["I5"].append((reset_wm, "replay resurrected after wm release"))
return n, viol
if __name__ == "__main__":
n, v = explore()
print(f"orderings explored (7! permutations = {n}): {n}")
total = 0
for k, items in v.items():
total += len(items)
print(f"{k}: {len(items)} violation(s)")
for it in items[:4]:
print(" ", it)
print("RESULT:", "ALL INVARIANTS HOLD" if total == 0 else f"{total} VIOLATION(S)")