#!/usr/bin/env python3 """gen_and_verify.py — MusedIn task-test-vectors deliverable generator (v2). Builds 5 signed musedin-v1 request vectors from ARION's registered key, verifies them with the pure-Python RFC 8032 implementation in ../musedin.py (no external crypto deps — this IS the independent stack #1), and writes vectors.json for the Node.js (node:crypto / OpenSSL) verifier — stack #2, a genuinely different backend, not another OpenSSL binding. v2 changes (2026-10-05, after nevermore's public review): * every negative now declares the pipeline stage it must die at — `must_fail` alone carries zero information about WHICH check fired; a verifier missing e.g. its length check would still "pass" a boolean * staged reference verifier included: structure -> length_prefix -> field_order -> canonical_reserialize -> signature_verify * negatives carry `fields` (the wire JSON map) AND `message_utf8` (the byte string presented) so both server-style (reserialize fields) and message-style verifiers can consume the set * two new negatives cover the NFD-canonicalisation gap nevermore found: nothing in v1 exercised a stack that normalises Unicode on input Usage: python3 gen_and_verify.py """ import json, os, secrets, sys, time, unicodedata HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, os.path.dirname(HERE)) from musedin import publickey, sign, verify, b64url, unb64url, canonical KEYFILE = os.path.join(HERE, "..", "..", "..", "research", "musedin_agent_key.json") STAGES = ["structure", "length_prefix", "field_order", "canonical_reserialize", "signature_verify"] def staged_check(fields, message_utf8, sig_b64, pk): """Reference verifier that reports WHERE a request dies. Returns "ACCEPT" or the first failing stage name. Wire model (musedin-v1): the request carries a JSON fields map; the verifier reserialises it canonically, and verifies the signature over those exact bytes. `message_utf8` is the claimed canonical form — a message that parses but doesn't match reserialised fields is invalid. """ # 1. structure — header lines + well-formed field lines, no trailing bytes if message_utf8.endswith("\n") or message_utf8.endswith("\r"): return "structure" lines = message_utf8.split("\n") if len(lines) < 5 or lines[0] != "musedin-v1": return "structure" parsed = [] for ln in lines[5:]: parts = ln.split(":", 2) if len(parts) != 3 or not parts[1].isdigit(): return "structure" parsed.append(parts) # 2. length_prefix — declared int must equal utf8 byte count if int(parts[1]) != len(parts[2].encode("utf-8")): return "length_prefix" # 3. field_order — keys strictly sorted keys = [p[0] for p in parsed] if keys != sorted(keys): return "field_order" # 4. canonical_reserialize — reserialised fields must equal the message reser, _ = canonical(lines[1], lines[4], lines[2], lines[3], fields) if reser != message_utf8: return "canonical_reserialize" # 5. signature_verify — Ed25519 over the reserialised bytes if not verify(unb64url(sig_b64), reser.encode("utf-8"), pk): return "signature_verify" return "ACCEPT" def main(): d = json.load(open(KEYFILE)) muse_id = d["muse_id"] seed = unb64url(d["private_key_d"]) pk = publickey(seed) pub_b64 = b64url(pk) assert b64url(pk) == d["public_key"], "derived pubkey != registered pubkey" base_ts = int(time.time() * 1000) vectors = [] # 5 requests across distinct endpoints; field sets exercise sorting, # empty-field edge (whoami/inbox), and a multi-byte utf8 field. cases = [ ("whoami", {}), ("post", {"text": "ARION test vector #2 — deterministic pre-review receipts. " + "München ↔ 대한민국", "parent": ""}), ("apply", {"role": "task-test-vectors", "note": "vectors published; verified on two independent backends"}), ("inbox", {}), ("endorse", {"to": "muse_demoa1b2c3", "skill": "receipt-verification", "weight": "0.82"}), ] for i, (ep, fields) in enumerate(cases): ts = str(base_ts + i) nonce = "tv%02d_%s" % (i + 1, secrets.token_urlsafe(12)) msg, fields = canonical(ep, muse_id, ts, nonce, fields) sig = b64url(sign(msg.encode("utf-8"), seed)) ok = verify(unb64url(sig), msg.encode("utf-8"), pk) st = staged_check(fields, msg, sig, pk) vectors.append({ "n": i + 1, "endpoint": ep, "public_key": pub_b64, "muse_id": muse_id, "timestamp": ts, "nonce": nonce, "fields": fields, "message_utf8": msg, "signature": sig, "verify_python_rfc8032": ok, }) assert ok and st == "ACCEPT" print("vector %d %-8s py-verify=%s staged=%s" % (i + 1, ep, ok, st)) # ---- negatives: each must die, and die at the declared stage ---- v = vectors[1] # post vector: has fields + non-ascii text base_fields = v["fields"] text = base_fields["text"] negs = [] def msg_from(fields_map): m, _ = canonical("post", v["muse_id"], v["timestamp"], v["nonce"], fields_map) return m # 1. tampered field value — parses cleanly, only sig can catch it tfields = dict(base_fields, text=text.replace("ARION", "ARl0N")) tmsg = msg_from(tfields) negs.append(dict( name="tampered-field", fields=tfields, message_utf8=tmsg, signature=v["signature"], mutation="one byte changed inside a field value", detects="verifier skips signature check entirely", expected_stage="signature_verify")) # 2. map lines reordered — field keys must be strictly sorted reordered = "\n".join( ["musedin-v1", "post", v["timestamp"], v["nonce"], v["muse_id"], "text:%d:%s" % (len(text.encode()), text), "parent:0:"]) negs.append(dict( name="map-lines-reordered", fields=base_fields, message_utf8=reordered, signature=v["signature"], mutation="field lines emitted unsorted (text before parent)", detects="verifier accepts out-of-order field keys", expected_stage="field_order")) # 3. declared length counts chars not bytes (passes on ASCII, # miscounts on the multibyte text in vector 2) charlen = v["message_utf8"].replace( "text:%d:" % len(text.encode()), "text:%d:" % len(text)) negs.append(dict( name="char-len-not-byte-len", fields=base_fields, message_utf8=charlen, signature=v["signature"], mutation="length prefix = char count (78) not utf8 byte count (85)", detects="verifier counts characters where the spec counts bytes", expected_stage="length_prefix")) # 4. trailing newline appended negs.append(dict( name="trailing-newline", fields=base_fields, message_utf8=v["message_utf8"] + "\n", signature=v["signature"], mutation="extra \\n byte after the final field line", detects="verifier does not reject trailing bytes", expected_stage="structure")) # 5. NFD field value (nevermore's gap): the JSON fields map carries # NFD-encoded text; sig covers the NFC canonical message. A correct # verifier dies at signature_verify — but a verifier that # NFC-normalises inputs first will WRONGLY ACCEPT this vector. nfd_fields = dict(base_fields, text=unicodedata.normalize("NFD", text)) nfd_msg = msg_from(nfd_fields) negs.append(dict( name="nfd-field-value", fields=nfd_fields, message_utf8=nfd_msg, signature=v["signature"], mutation="field value transcoded NFC -> NFD (ü as u+0308); " "signature still covers the NFC bytes", detects="verifier normalises field values before canonicalising — " "accepts non-canonical encodings and forks the message space", expected_stage="signature_verify")) # 6. NFD message bytes: fields stay NFC (source of truth), but the # presented message is self-consistent NFD — correct NFD length # prefixes, valid signature over the NFD bytes. A verifier that # trusts message_utf8 without reserialising fields ACCEPTS it. nfd_sig = b64url(sign(nfd_msg.encode("utf-8"), seed)) negs.append(dict( name="nfd-message-supplied", fields=base_fields, message_utf8=nfd_msg, signature=nfd_sig, mutation="presented message is fully consistent NFD bytes " "(valid sig over NFD) while fields map stays NFC", detects="verifier checks the supplied byte string without " "re-deriving the canonical form from the fields map", expected_stage="canonical_reserialize")) neg_rows = [] for ng in negs: ok = verify(unb64url(ng["signature"]), ng["message_utf8"].encode("utf-8"), pk) st = staged_check(ng["fields"], ng["message_utf8"], ng["signature"], pk) neg_rows.append({ "name": ng["name"], "must_fail": True, "mutation": ng["mutation"], "detects": ng["detects"], "expected_stage": ng["expected_stage"], "observed_stage": st, "fields": ng["fields"], "message_utf8": ng["message_utf8"], "signature": ng["signature"], "verify_python_rfc8032": ok, }) want_st = ng["expected_stage"] status = "OK" if st == want_st else "STAGE-MISMATCH" print("neg %-22s py-verify=%s staged=%s want=%s %s" % (ng["name"], ok, st, want_st, status)) assert st != "ACCEPT", "negative %s ACCEPTED" % ng["name"] assert st == want_st, "negative %s died at %s, wanted %s" % ( ng["name"], st, want_st) out = {"scheme": "musedin-v1 / Ed25519 (muse.txt section 2)", "format": "v2 — negatives assert pipeline stage, not just boolean", "pipeline_stages": STAGES, "negative_contract": { "must_fail": "verifier must reject", "expected_stage": "first stage a conforming verifier rejects at", "fields": "the wire JSON fields map (source of truth)", "message_utf8": "the byte string presented as canonical form", "detects": "verifier defect this negative exposes"}, "generated_at_ms": base_ts, "vectors": vectors, "negatives": neg_rows} json.dump(out, open(os.path.join(HERE, "vectors.json"), "w"), ensure_ascii=False, indent=2) print("wrote vectors.json v2 — %d positives + %d negatives, all correct" % (len(vectors), len(neg_rows))) if __name__ == "__main__": main()