f312e2cafa
claim_verify.py made the key, order and range checks itself and left the equation to `cryptography`; http_verify.py had its own RFC 8032 implementation; and pack_verify.py called `cryptography` alone, which accepts a key of small order, a second spelling of a key, and an R of small order. All three, and the JWP and signer verifiers through claim_verify, now call claim-conformance/_ed25519.py, written from RFC 8032 with the standard library (as http_verify's was): canonical A of large order, S < L, R decoded and not of small order, and the cofactorless equation with R compared byte for byte. It reports a bad key apart from a bad signature, keeps http_verify's --mutate small-order-key and small-order-r, and checks itself against RFC 8032's first two vectors and the three forgeries. Every committed corpus verifies unchanged: no vector's expectation moves. The OER reference keeps its own rule (oer-receipts §5). Co-Authored-By: Claude Opus 5.5 <[email protected]>
765 lines
30 KiB
Python
765 lines
30 KiB
Python
#!/usr/bin/env python3
|
|
"""Second verifier for signer-conformance/ — key discovery, signing authority and module
|
|
authorisation (extensions/jwp-receipts.md §7).
|
|
|
|
Written from the extension text. It shares no code with wai-rs (`crate::key_log`,
|
|
`crate::signer_id`), which generates the corpus. It takes from the claim corpus's own
|
|
second verifier (`../claim-conformance/claim_verify.py`) what §7 reuses from §3.1: the
|
|
JSON reading of step 1, the canonical form (JCS), BLAKE3 written from its specification,
|
|
the strict Ed25519 rule of step 3, the RFC 3339 syntax, and claim verification itself.
|
|
Everything §7 adds — the signer_id grammar and base58btc, the key log's rules, the head
|
|
walk, anchoring, confirmation, the time comparison and §7.9's module checks — is written
|
|
below.
|
|
|
|
python3 signer_verify.py [vectors.json] verify every vector and the canonical fixture
|
|
python3 signer_verify.py fixture print the fixture's canonical bytes,
|
|
signature and revision hash
|
|
python3 signer_verify.py verdicts <file> print `name<TAB>verdict` for every vector,
|
|
ignoring any declared verdict
|
|
|
|
Exit status 0 only if every vector reaches its declared verdict and the fixture this
|
|
script computes equals the literals pinned in wai-rs.
|
|
"""
|
|
import datetime, hashlib, json, os, re, sys
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance"))
|
|
import claim_verify as CV # noqa: E402
|
|
|
|
from cryptography.hazmat.primitives import serialization # noqa: E402
|
|
from cryptography.hazmat.primitives.asymmetric import ed25519 # noqa: E402
|
|
|
|
D_SET, D_LINK, D_NEXT = b"wai:key-set\x01", b"wai:key-set-link\x01", b"wai:key-set-next\x01"
|
|
D_GRANT, D_STATUS = b"wai:module-grant\x01", b"wai:key-status\x01"
|
|
GRANT_CEILING, STATUS_CEILING = 86400, 3600
|
|
HARDWARE = {"CalibratedInstrument", "OnChipCounter", "HwShunt"}
|
|
PINNED_ONLY = {"energy/CalibratedInstrument", "energy/OnChipCounter", "energy/HwShunt",
|
|
"lawful-basis/declaration", "lawful-basis/decision"}
|
|
REVISION = ["kind", "origin", "seq", "prev", "issued_at", "root", "next_root", "keys", "revoked_modules",
|
|
"grant_max_s", "status_max_age_s", "sig"]
|
|
ENTRY = ["id", "key", "may_sign", "status", "until", "compromised_since"]
|
|
GRANT = ["kind", "origin", "module", "form", "capabilities", "not_before", "not_after", "signer_pubkey",
|
|
"signer_id", "sig"]
|
|
STATUS = ["kind", "origin", "head", "seq", "issued_at", "sig"]
|
|
RANK = {"active": 0, "retired": 1, "revoked": 2}
|
|
|
|
|
|
class Stop(Exception):
|
|
"""A reason: a log not anchored, a revision not held, an item not confirmed."""
|
|
|
|
def __init__(self, reason, level=None, step=None):
|
|
super().__init__(reason)
|
|
self.reason, self.level, self.step = reason, level, step
|
|
|
|
|
|
# ------------------------------------------------------------------ §7.1 signer_id
|
|
|
|
B58 = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz"
|
|
|
|
|
|
def b58encode(b: bytes) -> str:
|
|
n = int.from_bytes(b, "big")
|
|
out = ""
|
|
while n:
|
|
n, r = divmod(n, 58)
|
|
out = B58[r] + out
|
|
return "1" * (len(b) - len(b.lstrip(b"\0"))) + out
|
|
|
|
|
|
def b58decode(s: str):
|
|
if any(c not in B58 for c in s):
|
|
return None
|
|
n = 0
|
|
for c in s:
|
|
n = n * 58 + B58.index(c)
|
|
body = n.to_bytes((n.bit_length() + 7) // 8, "big") if n else b""
|
|
return b"\0" * (len(s) - len(s.lstrip("1"))) + body
|
|
|
|
|
|
LABEL = re.compile(r"[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?\Z")
|
|
|
|
|
|
def host_ok(h):
|
|
labels = h.split(".")
|
|
return (len(h) <= 253 and len(labels) >= 2 and all(LABEL.match(x) for x in labels)
|
|
and not labels[-1].isdigit())
|
|
|
|
|
|
def port_ok(p):
|
|
return re.fullmatch(r"[1-9][0-9]{0,4}", p) is not None and 1 <= int(p) <= 65535 and int(p) != 443
|
|
|
|
|
|
def key_id_ok(k):
|
|
return re.fullmatch(r"[A-Za-z0-9._-]{1,64}", k) is not None
|
|
|
|
|
|
def did_key(pk: bytes) -> str:
|
|
return "did:key:z" + b58encode(b"\xed\x01" + pk)
|
|
|
|
|
|
def classify(sid):
|
|
"""('origin', origin, key_id) | ('didkey', key) | ('label',)."""
|
|
if sid.startswith("did:web:"):
|
|
rest = sid[len("did:web:"):]
|
|
if "#" in rest:
|
|
auth, kid = rest.split("#", 1)
|
|
host, port = (auth.split("%3A", 1) + [None])[:2]
|
|
if key_id_ok(kid) and host_ok(host) and (port is None or port_ok(port)):
|
|
return ("origin", "https://" + host + (":" + port if port else ""), kid)
|
|
return ("label",)
|
|
# The base58btc text of 0xed 0x01 and a 32-byte key is 47 characters; a longer one is
|
|
# not its spelling and is not decoded.
|
|
if sid.startswith("did:key:z") and len(sid) - len("did:key:z") <= 47:
|
|
b = b58decode(sid[len("did:key:z"):])
|
|
if b is not None and len(b) == 34 and b[:2] == b"\xed\x01" and did_key(b[2:]) == sid:
|
|
return ("didkey", b[2:])
|
|
return ("label",)
|
|
|
|
|
|
def origin_ok(o):
|
|
m = re.fullmatch(r"https://([^:/]+)(?::([0-9]+))?", o)
|
|
return m is not None and host_ok(m.group(1)) and (m.group(2) is None or port_ok(m.group(2)))
|
|
|
|
|
|
# ------------------------------------------------------------------ times
|
|
|
|
LOG_TIME = re.compile(r"[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-5][0-9]Z\Z")
|
|
|
|
|
|
def is_log_time(t):
|
|
return isinstance(t, str) and LOG_TIME.match(t) is not None and CV.is_rfc3339(t)
|
|
|
|
|
|
def secs(t):
|
|
d = datetime.datetime(int(t[0:4]), int(t[5:7]), int(t[8:10]), int(t[11:13]), int(t[14:16]), int(t[17:19]),
|
|
tzinfo=datetime.timezone.utc)
|
|
return int(d.timestamp())
|
|
|
|
|
|
def instant(t):
|
|
"""(minute in UTC, second as written, fraction digits without trailing zeros)."""
|
|
m = re.fullmatch(r"([0-9]{4})-([0-9]{2})-([0-9]{2})[Tt]([0-9]{2}):([0-9]{2}):([0-9]{2})(?:\.([0-9]+))?"
|
|
r"(?:[Zz]|([+-])([0-9]{2}):([0-9]{2}))", t)
|
|
day = datetime.date(int(m.group(1)), int(m.group(2)), int(m.group(3))).toordinal()
|
|
minute = day * 1440 + int(m.group(4)) * 60 + int(m.group(5))
|
|
if m.group(8):
|
|
off = int(m.group(9)) * 60 + int(m.group(10))
|
|
minute -= off if m.group(8) == "+" else -off
|
|
return minute, int(m.group(6)), (m.group(7) or "").rstrip("0")
|
|
|
|
|
|
def later_than(t, until):
|
|
if not (isinstance(t, str) and CV.is_rfc3339(t)):
|
|
return True
|
|
a, b = instant(t), instant(until)
|
|
if a[:2] != b[:2]:
|
|
return a[:2] > b[:2]
|
|
w = max(len(a[2]), len(b[2]))
|
|
return a[2].ljust(w, "0") > b[2].ljust(w, "0")
|
|
|
|
|
|
# ------------------------------------------------------------------ signatures and hashes
|
|
|
|
def hexb(v, n):
|
|
if isinstance(v, str) and len(v) == 2 * n and re.fullmatch(r"[0-9a-f]*", v):
|
|
return bytes.fromhex(v)
|
|
return None
|
|
|
|
|
|
def verifies(pk: bytes, sig: bytes, msg: bytes) -> bool:
|
|
try:
|
|
CV.check_signature(pk, sig, msg)
|
|
return True
|
|
except CV.Reject:
|
|
return False
|
|
|
|
|
|
def key_accepted(pk: bytes) -> bool:
|
|
return CV.ED.strict_key(pk)
|
|
|
|
|
|
def commitment(k: bytes) -> bytes:
|
|
return CV.blake3(D_NEXT + k)
|
|
|
|
|
|
def without_sig(d):
|
|
return {k: v for k, v in d.items() if k != "sig"}
|
|
|
|
|
|
def rev_hash(r) -> str:
|
|
return CV.blake3(D_LINK + CV.jcs(r).encode()).hex()
|
|
|
|
|
|
def module_digest(b: bytes) -> str:
|
|
return hashlib.sha256(b).hexdigest()
|
|
|
|
|
|
# ------------------------------------------------------------------ §7.2 the key log
|
|
|
|
def authority_ok(a):
|
|
return isinstance(a, str) and 1 <= len(a) <= 128 and re.fullmatch(r"[A-Za-z0-9._-]+(/[A-Za-z0-9._-]+)*", a)
|
|
|
|
|
|
def exact(d, members):
|
|
return isinstance(d, dict) and set(d) == set(members)
|
|
|
|
|
|
def check_entry(e):
|
|
ok = (exact(e, ENTRY) and isinstance(e["id"], str) and key_id_ok(e["id"]) and hexb(e["key"], 32) is not None
|
|
and key_accepted(bytes.fromhex(e["key"])) and isinstance(e["may_sign"], list) and e["may_sign"]
|
|
and all(authority_ok(a) for a in e["may_sign"]) and len(set(e["may_sign"])) == len(e["may_sign"])
|
|
and e["status"] in RANK and all(e[t] is None or is_log_time(e[t]) for t in ("until", "compromised_since")))
|
|
if not ok:
|
|
return False
|
|
st, u, c = e["status"], e["until"], e["compromised_since"]
|
|
return {"active": u is None and c is None, "retired": u is not None and c is None,
|
|
"revoked": c is not None}[st]
|
|
|
|
|
|
def check_revision(r):
|
|
"""Rule 1 on a parsed value."""
|
|
bad = Stop("key_log_malformed")
|
|
if not exact(r, REVISION) or r["kind"] != "wai-key-set" or not isinstance(r["origin"], str):
|
|
raise bad
|
|
if not CV.is_int(r["seq"]) or not (r["prev"] is None or hexb(r["prev"], 32) is not None):
|
|
raise bad
|
|
if (r["seq"] == 0) != (r["prev"] is None) or not is_log_time(r["issued_at"]):
|
|
raise bad
|
|
root, nxt = hexb(r["root"], 32), hexb(r["next_root"], 32)
|
|
if root is None or nxt is None or nxt == commitment(root) or hexb(r["sig"], 64) is None:
|
|
raise bad
|
|
if not (CV.is_int(r["grant_max_s"]) and 1 <= r["grant_max_s"] <= GRANT_CEILING):
|
|
raise bad
|
|
if not (CV.is_int(r["status_max_age_s"]) and 1 <= r["status_max_age_s"] <= STATUS_CEILING):
|
|
raise bad
|
|
if not isinstance(r["keys"], list) or not all(check_entry(e) for e in r["keys"]):
|
|
raise bad
|
|
ids, keys = [e["id"] for e in r["keys"]], [e["key"] for e in r["keys"]]
|
|
if len(set(ids)) != len(ids) or len(set(keys)) != len(keys):
|
|
raise bad
|
|
if any(bytes.fromhex(k) == root or commitment(bytes.fromhex(k)) == nxt for k in keys):
|
|
raise bad
|
|
rm = r["revoked_modules"]
|
|
if not isinstance(rm, list) or not all(hexb(d, 32) is not None for d in rm) or len(set(rm)) != len(rm):
|
|
raise bad
|
|
return r
|
|
|
|
|
|
def parse_revision(text):
|
|
try:
|
|
return check_revision(CV.parse_strict(text))
|
|
except ValueError:
|
|
raise Stop("key_log_malformed")
|
|
|
|
|
|
def check_held(r, origin):
|
|
"""Rules 2-3."""
|
|
if r["origin"] != origin:
|
|
raise Stop("origin_mismatch")
|
|
if not verifies(bytes.fromhex(r["root"]), bytes.fromhex(r["sig"]), D_SET + CV.jcs(without_sig(r)).encode()):
|
|
raise Stop("key_log_signature_invalid")
|
|
|
|
|
|
def is_held(r, origin):
|
|
try:
|
|
check_held(r, origin)
|
|
return True
|
|
except Stop:
|
|
return False
|
|
|
|
|
|
def rotation_of(c, p):
|
|
return c["root"] != p["root"] and commitment(bytes.fromhex(c["root"])).hex() == p["next_root"]
|
|
|
|
|
|
def authorised(c, p):
|
|
return c["root"] == p["root"] or rotation_of(c, p)
|
|
|
|
|
|
def check_link(p, c):
|
|
"""Rules 4-6."""
|
|
if c["prev"] != rev_hash(p) or c["seq"] != p["seq"] + 1 or c["issued_at"] < p["issued_at"]:
|
|
raise Stop("key_log_chain_break")
|
|
if not ((c["root"] == p["root"] and c["next_root"] == p["next_root"]) or rotation_of(c, p)):
|
|
raise Stop("key_log_root_not_committed")
|
|
pk, ck = p["keys"], c["keys"]
|
|
if len(ck) < len(pk) or c["revoked_modules"][:len(p["revoked_modules"])] != p["revoked_modules"]:
|
|
raise Stop("key_log_rewrites_history")
|
|
for a, b in zip(pk, ck):
|
|
same = a["id"] == b["id"] and a["key"] == b["key"] and a["may_sign"] == b["may_sign"]
|
|
if not same or RANK[b["status"]] < RANK[a["status"]]:
|
|
raise Stop("key_log_rewrites_history")
|
|
for t in ("until", "compromised_since"):
|
|
if a[t] is not None and a[t] != b[t]:
|
|
raise Stop("key_log_rewrites_history")
|
|
|
|
|
|
def read_log(origin, text):
|
|
try:
|
|
doc = CV.parse_strict(text)
|
|
except ValueError:
|
|
raise Stop("key_log_malformed")
|
|
if not exact(doc, ["kind", "revisions"]) or doc["kind"] != "wai-key-log":
|
|
raise Stop("key_log_malformed")
|
|
revs = doc["revisions"]
|
|
if not isinstance(revs, list) or not revs:
|
|
raise Stop("key_log_malformed")
|
|
for r in revs:
|
|
check_revision(r)
|
|
check_held(r, origin)
|
|
if revs[0]["seq"] != 0:
|
|
raise Stop("key_log_chain_break")
|
|
for p, c in zip(revs, revs[1:]):
|
|
check_link(p, c)
|
|
return revs
|
|
|
|
|
|
# ------------------------------------------------------------------ §7.4 step 3, the head
|
|
|
|
def head(held, anchor):
|
|
"""Returns (head_hash, head_revision, moved_to or None)."""
|
|
by = {rev_hash(r): r for r in held}
|
|
if anchor not in by:
|
|
raise Stop("origin_not_anchored")
|
|
|
|
def kids(h):
|
|
p = by[h]
|
|
return [(k, c) for k, c in sorted(by.items()) if c["prev"] == h and authorised(c, p)]
|
|
|
|
line = [anchor]
|
|
while by[line[-1]]["prev"] in by and by[line[-1]]["prev"] not in line:
|
|
line.append(by[line[-1]]["prev"])
|
|
line.reverse()
|
|
start, moved = anchor, None
|
|
for t, n in zip(line, line[1:]):
|
|
if rotation_of(by[n], by[t]):
|
|
continue
|
|
rots = [(k, c) for k, c in kids(t) if k != n and rotation_of(c, by[t])]
|
|
if len(rots) == 1:
|
|
check_link(by[t], rots[0][1])
|
|
start = moved = rots[0][0]
|
|
break
|
|
if len(rots) > 1:
|
|
raise Stop("key_log_fork")
|
|
cur = start
|
|
while True:
|
|
ks = kids(cur)
|
|
if not ks:
|
|
return cur, by[cur], moved
|
|
if len(ks) > 1:
|
|
ks = [kc for kc in ks if rotation_of(kc[1], by[cur])]
|
|
if len(ks) != 1:
|
|
raise Stop("key_log_fork")
|
|
check_link(by[cur], ks[0][1])
|
|
cur = ks[0][0]
|
|
|
|
|
|
# ------------------------------------------------------------------ §7.3 anchoring
|
|
|
|
_PUBLISHED = None
|
|
|
|
|
|
def is_published(k: bytes) -> bool:
|
|
"""Whether k is the public key of a one-byte seed, or the commitment to one."""
|
|
global _PUBLISHED
|
|
if _PUBLISHED is None:
|
|
keys = [ed25519.Ed25519PrivateKey.from_private_bytes(bytes([b]) * 32).public_key().public_bytes(
|
|
serialization.Encoding.Raw, serialization.PublicFormat.Raw) for b in range(256)]
|
|
_PUBLISHED = set(keys) | {commitment(pk) for pk in keys}
|
|
return k in _PUBLISHED
|
|
|
|
|
|
def lists_published(r):
|
|
return any(is_published(bytes.fromhex(x)) for x in [r["root"], r["next_root"]] + [e["key"] for e in r["keys"]])
|
|
|
|
|
|
def anchor_log(origin, text, path, current):
|
|
"""('anchored', hash, path) | ('not_anchored', reason) | ('replaced',)."""
|
|
try:
|
|
revs = read_log(origin, text)
|
|
except Stop as s:
|
|
return ("not_anchored", s.reason)
|
|
hs = [rev_hash(r) for r in revs]
|
|
if current is None or path == "pinned":
|
|
return ("anchored", hs[-1], path, hs[-1] if path == "pinned" else None)
|
|
cur, held = current
|
|
if cur["revision"] in hs:
|
|
# Forward: a pinned record stays pinned and keeps the revision that was pinned.
|
|
if cur["path"] == "pinned" and cur.get("pinned_revision") is not None:
|
|
return ("anchored", hs[-1], "pinned", cur["pinned_revision"])
|
|
return ("anchored", hs[-1], path, None)
|
|
hb = {rev_hash(r): r for r in held if is_held(r, origin)}
|
|
trusted = [cur["revision"]] # newest first
|
|
while trusted[-1] in hb and hb[trusted[-1]]["prev"] in hb and hb[trusted[-1]]["prev"] not in trusted:
|
|
trusted.append(hb[trusted[-1]]["prev"])
|
|
if hs[-1] in trusted:
|
|
return ("not_anchored", "key_log_rolled_back")
|
|
shared = [i for i, x in enumerate(hs) if x in trusted]
|
|
if shared:
|
|
i = shared[-1]
|
|
if i + 1 < len(revs) and rotation_of(revs[i + 1], revs[i]):
|
|
pos = trusted.index(hs[i])
|
|
child = hb.get(trusted[pos - 1]) if pos > 0 else None
|
|
if child is None or not rotation_of(child, revs[i]):
|
|
return ("anchored", hs[-1], path, None)
|
|
return ("replaced",)
|
|
|
|
|
|
# ------------------------------------------------------------------ §7.4-§7.5 confirmation
|
|
|
|
def claim_authorities(c):
|
|
body = c["body"]
|
|
prov = body.get("energy_provenance") if isinstance(body, dict) else None
|
|
if c["class"] == "energy":
|
|
own = "energy/" + prov
|
|
elif c["class"] == "lawful-basis":
|
|
own = "lawful-basis/" + body["record"]
|
|
else:
|
|
own = c["class"]
|
|
out = [own]
|
|
if isinstance(prov, str) and "energy/" + prov not in out:
|
|
out.append("energy/" + prov)
|
|
return out
|
|
|
|
|
|
def figure_authorities(c):
|
|
"""§7.5: the energy figure of a `reconstruct-session` claim is an item of its own,
|
|
needing `reconstruct-session` and the energy authority of the class its `energy`
|
|
writes; None for a claim of another class, or a session that was not metered."""
|
|
e = c["body"].get("energy") if c["class"] == "reconstruct-session" else None
|
|
return None if e is None else ["reconstruct-session", "energy/" + e["energy_provenance"]]
|
|
|
|
|
|
def item_authorities(kind, classes):
|
|
if kind == "group":
|
|
# delivery, and energy/<class> for each class the group's labelled leaves and
|
|
# its labelled total carry (section 7.5).
|
|
out = ["delivery"]
|
|
for c in classes:
|
|
if "energy/" + c not in out:
|
|
out.append("energy/" + c)
|
|
return out
|
|
# receipt, and energy/<class> for each class the receipt's figures carry.
|
|
out = ["receipt"]
|
|
for c in classes:
|
|
if "energy/" + c not in out:
|
|
out.append("energy/" + c)
|
|
return out
|
|
|
|
|
|
def confirm(pubkey, sid, authorities, time, revisions, anchors, expected):
|
|
"""`expected` is None or a list of {origin, source}; the origin must be in a list given."""
|
|
form = classify(sid)
|
|
if form[0] == "didkey":
|
|
if form[1] != pubkey:
|
|
return "refused step=1 reason=signer_id_names_another_key"
|
|
return "unconfirmed step=1 reason=signer_id_not_origin_bound"
|
|
if form[0] == "label":
|
|
return "unconfirmed step=1 reason=signer_id_not_origin_bound"
|
|
_, origin, kid = form
|
|
held = [r for r in revisions if is_held(r, origin)]
|
|
anc = next((a for a in anchors if a["origin"] == origin), None)
|
|
if anc is None or anc["revision"] not in [rev_hash(r) for r in held]:
|
|
return "unconfirmed step=2 reason=origin_not_anchored"
|
|
try:
|
|
hh, hr, moved = head(held, anc["revision"])
|
|
except Stop as s:
|
|
return "unconfirmed step=3 reason=" + s.reason
|
|
e = next((x for x in hr["keys"] if x["id"] == kid), None)
|
|
if e is None:
|
|
return "refused step=4 reason=key_not_listed"
|
|
if bytes.fromhex(e["key"]) != pubkey:
|
|
return "refused step=4 reason=key_mismatch"
|
|
if e["status"] == "revoked":
|
|
return "refused step=5 reason=key_revoked"
|
|
if e["status"] == "retired" and time is not None and later_than(time, e["until"]):
|
|
return "refused step=5 reason=key_retired"
|
|
if not all(a in e["may_sign"] for a in authorities):
|
|
return "refused step=6 reason=authority_not_granted"
|
|
pinned = anc["path"] == "pinned" and anc.get("pinned_revision") is not None and moved is None
|
|
if any(a in PINNED_ONLY for a in authorities) and not pinned:
|
|
return "unconfirmed step=6a reason=anchor_not_pinned"
|
|
match = None
|
|
if expected is not None:
|
|
match = next((x for x in expected if x["origin"] == origin), None)
|
|
if match is None:
|
|
return "refused step=7 reason=origin_not_expected"
|
|
v = f"confirmed origin={origin} key_id={kid} head={hh} anchor={anc['path']}"
|
|
if moved:
|
|
v += " moved_to=" + moved
|
|
if e["status"] == "retired":
|
|
v += " retired_until=" + e["until"]
|
|
if time is None:
|
|
v += " untimed=true"
|
|
if match is not None:
|
|
v += " expected=" + match["source"]
|
|
if any(a.startswith("energy/") and a[len("energy/"):] in HARDWARE for a in authorities):
|
|
v += " also_delivers=" + ("true" if "delivery" in e["may_sign"] else "false")
|
|
return v
|
|
|
|
|
|
# ------------------------------------------------------------------ §7.9 modules
|
|
|
|
def parse_grant(text):
|
|
try:
|
|
g = CV.parse_strict(text)
|
|
except ValueError:
|
|
return None
|
|
if not exact(g, GRANT) or g["kind"] != "wai-module-grant":
|
|
return None
|
|
caps = g["capabilities"]
|
|
ok = (isinstance(g["origin"], str) and origin_ok(g["origin"]) and hexb(g["module"], 32) is not None
|
|
and g["form"] in ("decoder", "reconstruct") and isinstance(caps, list) and caps
|
|
and all(isinstance(c, str) and CV.is_token(c) for c in caps) and len(set(caps)) == len(caps)
|
|
and is_log_time(g["not_before"]) and is_log_time(g["not_after"]) and g["not_before"] <= g["not_after"]
|
|
and hexb(g["signer_pubkey"], 32) is not None and isinstance(g["signer_id"], str)
|
|
and classify(g["signer_id"])[:2] == ("origin", g["origin"])
|
|
and hexb(g["sig"], 64) is not None)
|
|
if not ok or not verifies(bytes.fromhex(g["signer_pubkey"]), bytes.fromhex(g["sig"]),
|
|
D_GRANT + CV.jcs(without_sig(g)).encode()):
|
|
return None
|
|
return g
|
|
|
|
|
|
def parse_status(text):
|
|
try:
|
|
s = CV.parse_strict(text)
|
|
except ValueError:
|
|
return None
|
|
if (not exact(s, STATUS) or s["kind"] != "wai-key-status" or not isinstance(s["origin"], str)
|
|
or hexb(s["head"], 32) is None or not CV.is_int(s["seq"]) or not is_log_time(s["issued_at"])
|
|
or hexb(s["sig"], 64) is None):
|
|
return None
|
|
return s
|
|
|
|
|
|
def revoked_by(digest, revisions, anchors):
|
|
"""The origin of the first anchor whose head (or, where none can be reached, whose
|
|
anchor revision) lists digest in revoked_modules, or None (7.9 check 5)."""
|
|
for a in anchors:
|
|
ah = [r for r in revisions if is_held(r, a["origin"])]
|
|
try:
|
|
lst = head(ah, a["revision"])[1]["revoked_modules"]
|
|
except Stop:
|
|
ar = next((r for r in ah if rev_hash(r) == a["revision"]), None)
|
|
lst = ar["revoked_modules"] if ar is not None else []
|
|
if digest in lst:
|
|
return a["origin"]
|
|
return None
|
|
|
|
|
|
def module(c):
|
|
v = module_checks(c)
|
|
if c["object"] != "parameter-set" or v.startswith("run") or "reason=module_revoked" in v:
|
|
return v
|
|
# SPEC 3.1 check 5: a pinned set whose digest a held anchor's head revokes is refused
|
|
# as revoked, whichever check a grant failed first, and with no grant at all.
|
|
o = revoked_by(module_digest(bytes.fromhex(c["module"])), held_revisions(c["revisions"]), c["anchors"])
|
|
if o is not None:
|
|
fb = "fallback" if c["fallback"] else "inert"
|
|
return f"{fb} step=5 reason=module_revoked revoked_by={o} refusal=prior_revoked"
|
|
return v
|
|
|
|
|
|
def module_checks(c):
|
|
fb = "fallback" if c["fallback"] else "inert"
|
|
if c["grant"] is None and c["object"] == "parameter-set":
|
|
return f"{fb} no_grant refusal=prior_not_authorised"
|
|
|
|
def no(step, reason, revoked_by=None):
|
|
v = f"{fb} step={step} reason={reason}"
|
|
if revoked_by is not None:
|
|
v += " revoked_by=" + revoked_by
|
|
if c["object"] == "parameter-set":
|
|
# SPEC 3.1 check 5: a revoked digest, or any other failure of 7.9.
|
|
v += " refusal=" + ("prior_revoked" if reason == "module_revoked" else "prior_not_authorised")
|
|
return v
|
|
|
|
g = parse_grant(c["grant"])
|
|
if g is None:
|
|
return no(1, "module_grant_invalid")
|
|
if module_digest(bytes.fromhex(c["module"])) != g["module"]:
|
|
return no(2, "module_digest_mismatch")
|
|
revisions = held_revisions(c["revisions"])
|
|
v = confirm(bytes.fromhex(g["signer_pubkey"]), g["signer_id"], ["module/" + g["form"]], g["not_before"],
|
|
revisions, c["anchors"], c["module_origins"])
|
|
if not v.startswith("confirmed"):
|
|
return no(3, v.split("reason=")[1])
|
|
fields = dict(x.split("=", 1) for x in v.split()[1:])
|
|
origin, hh = fields["origin"], fields["head"]
|
|
held = [r for r in revisions if is_held(r, origin)]
|
|
hr = next(r for r in held if rev_hash(r) == hh)
|
|
now = c["now"]
|
|
if not is_log_time(now):
|
|
return no(4, "module_status_stale")
|
|
# The head and its held ancestors: a status naming anything else is not held.
|
|
by = {rev_hash(r): r for r in held}
|
|
line = [hh]
|
|
while by[line[-1]]["prev"] in by and by[line[-1]]["prev"] not in line:
|
|
line.append(by[line[-1]]["prev"])
|
|
valid = []
|
|
for t in c["statuses"]:
|
|
s = parse_status(t)
|
|
if s is None or s["origin"] != origin or s["head"] not in line:
|
|
continue
|
|
for r in held:
|
|
if (rev_hash(r) == s["head"] and r["seq"] == s["seq"]
|
|
and verifies(bytes.fromhex(r["root"]), bytes.fromhex(s["sig"]),
|
|
D_STATUS + CV.jcs(without_sig(s)).encode())):
|
|
valid.append(s)
|
|
break
|
|
current = [s["issued_at"] for s in valid if s["head"] == hh]
|
|
if not current:
|
|
return no(4, "module_status_not_descended" if valid else "module_status_stale")
|
|
age = secs(now) - secs(max(current))
|
|
if age < 0 or age > hr["status_max_age_s"]:
|
|
return no(4, "module_status_stale")
|
|
if g["module"] in hr["revoked_modules"]:
|
|
return no(5, "module_revoked")
|
|
o = revoked_by(g["module"], revisions, c["anchors"])
|
|
if o is not None:
|
|
return no(5, "module_revoked", o)
|
|
if now < g["not_before"]:
|
|
return no(6, "module_grant_not_yet_valid")
|
|
if now > g["not_after"]:
|
|
return no(6, "module_grant_expired")
|
|
if secs(g["not_after"]) - secs(g["not_before"]) > hr["grant_max_s"]:
|
|
return no(7, "module_grant_too_long")
|
|
if c["capability"] not in g["capabilities"]:
|
|
return no(8, "module_capability_not_granted")
|
|
return (f"run origin={origin} key_id={fields['key_id']} head={hh} anchor={fields['anchor']} "
|
|
f"expected={fields['expected']} kill_latency_s={hr['grant_max_s'] + hr['status_max_age_s']}")
|
|
|
|
|
|
# ------------------------------------------------------------------ vectors
|
|
|
|
def held_revisions(texts):
|
|
out = []
|
|
for t in texts:
|
|
try:
|
|
out.append(parse_revision(t))
|
|
except Stop:
|
|
pass # a revision that fails rule 1 is not held
|
|
return out
|
|
|
|
|
|
def verdict(v):
|
|
c = v["context"]
|
|
if v["procedure"] == "confirm":
|
|
if "receipt" in v:
|
|
try:
|
|
cl = CV.verify_claim(v["receipt"])
|
|
except CV.Reject as r:
|
|
return f"reject step={r.step} reason={r.reason}"
|
|
pub, sid, auth, time = bytes.fromhex(cl["signer_pubkey"]), cl["signer_id"], claim_authorities(cl), cl["issued_at"]
|
|
figure = figure_authorities(cl)
|
|
else:
|
|
it = c["item"]
|
|
pub, sid, time = bytes.fromhex(it["signer_pubkey"]), it["signer_id"], it["time"]
|
|
auth, figure = item_authorities(it["kind"], it["classes"]), None
|
|
exp = c.get("expected_origin")
|
|
held, expected = held_revisions(c["revisions"]), None if exp is None else [exp]
|
|
out = confirm(pub, sid, auth, time, held, c["anchors"], expected)
|
|
if figure is not None:
|
|
out += "; figure " + confirm(pub, sid, figure, time, held, c["anchors"], expected)
|
|
return out
|
|
if v["procedure"] == "anchor":
|
|
cur = c.get("current")
|
|
current = (cur["anchor"], held_revisions(cur["held"])) if cur else None
|
|
out = anchor_log(c["origin"], c["log"], c["path"], current)
|
|
if out[0] == "anchored":
|
|
latest = read_log(c["origin"], c["log"])[-1]
|
|
note = " note=published_test_key" if lists_published(latest) else ""
|
|
pinned = f" pinned_revision={out[3]}" if out[3] is not None and out[3] != out[1] else ""
|
|
return f"anchored revision={out[1]} anchor={out[2]}{pinned}{note}"
|
|
if out[0] == "not_anchored":
|
|
return "not_anchored reason=" + out[1]
|
|
return "replaced reason=key_log_replaced"
|
|
if v["procedure"] == "module":
|
|
return module(c)
|
|
raise ValueError("unknown procedure")
|
|
|
|
|
|
# ------------------------------------------------------------------ canonical fixture
|
|
|
|
def seed_key(b):
|
|
return ed25519.Ed25519PrivateKey.from_private_bytes(bytes([b]) * 32)
|
|
|
|
|
|
def pub(sk):
|
|
return sk.public_key().public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw)
|
|
|
|
|
|
def fixture():
|
|
"""The revision wai-rs pins: computed here from its members and the seeds 0x61 (root),
|
|
0x62 (committed) and 0x63 (its one key)."""
|
|
sk = seed_key(0x61)
|
|
r = {
|
|
"status_max_age_s": 600, "keys": [{"until": None, "status": "active", "key": pub(seed_key(0x63)).hex(),
|
|
"may_sign": ["delivery", "energy/OnChipCounter"], "id": "d1",
|
|
"compromised_since": None}],
|
|
"root": pub(sk).hex(), "seq": 0, "prev": None, "origin": "https://publisher.example",
|
|
"next_root": commitment(pub(seed_key(0x62))).hex(), "kind": "wai-key-set",
|
|
"issued_at": "2026-09-01T00:00:00Z", "revoked_modules": [], "grant_max_s": 3600,
|
|
}
|
|
r["sig"] = sk.sign(D_SET + CV.jcs(r).encode()).hex()
|
|
return r, CV.jcs(r), rev_hash(r)
|
|
|
|
|
|
def fixture_matches_wai_rs():
|
|
src_path = os.path.join(HERE, "..", "wai-rs", "src", "key_log", "corpus.rs")
|
|
try:
|
|
src = open(src_path, encoding="utf-8").read()
|
|
except OSError as e:
|
|
return [f"cannot read {src_path}: {e}"]
|
|
r, canon, h = fixture()
|
|
fails = []
|
|
m = re.search(r'const CANONICAL: &str = "((?:[^"\\]|\\.)*)";', src)
|
|
pinned = re.sub(r'\\(["\\])', r"\1", m.group(1)) if m else None
|
|
if pinned != canon:
|
|
fails.append("canonical bytes differ from wai-rs")
|
|
m = re.search(r'const HASH: &str = "([0-9a-f]{64})";', src)
|
|
if not m or m.group(1) != h:
|
|
fails.append("revision hash differs from wai-rs")
|
|
if r["sig"] not in src:
|
|
fails.append("signature differs from wai-rs")
|
|
return fails
|
|
|
|
|
|
def main(argv):
|
|
if argv[1:2] == ["verdicts"] and len(argv) == 3:
|
|
doc = json.load(open(argv[2], encoding="utf-8"))
|
|
for v in doc["vectors"]:
|
|
try:
|
|
got = verdict(v)
|
|
except (ValueError, KeyError, TypeError):
|
|
got = "error"
|
|
print(f"{v['name']}\t{got}")
|
|
return 0
|
|
if argv[1:2] == ["fixture"]:
|
|
r, canon, h = fixture()
|
|
print("sig ", r["sig"])
|
|
print("canonical ", canon)
|
|
print("revision_hash", h)
|
|
return 0
|
|
fail = 0
|
|
pk11 = pub(seed_key(0x11))
|
|
sid = did_key(pk11) == "did:key:z6MktULudTtAsAhRegYPiZ6631RV3viv12qd4GQF8z1xB22S"
|
|
print(("PASS" if sid else "FAIL") + " did:key of seed 0x11 reproduces its literal")
|
|
fail += not sid
|
|
ff = fixture_matches_wai_rs()
|
|
print(("PASS" if not ff else "FAIL") + " canonical fixture equals the literals pinned in wai-rs " + ",".join(ff))
|
|
fail += bool(ff)
|
|
path = argv[1] if len(argv) > 1 else os.path.join(HERE, "vectors.json")
|
|
doc = json.load(open(path, encoding="utf-8"))
|
|
for v in doc["vectors"]:
|
|
got = verdict(v)
|
|
ok = got == v["expected"]
|
|
fail += not ok
|
|
print(f"{'PASS' if ok else 'FAIL'} {v['name']} expected={v['expected']} got={got}")
|
|
print(f"{len(doc['vectors'])} vectors, {fail} failure(s)")
|
|
return 1 if fail else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main(sys.argv))
|