jlpoole/reticiulum-specification
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Bootstrap test-vectors/{announces,lxmf,links}.json + regenerators

Browse source 
Three deterministic vector files complete the test-vectors/ bootstrap.
Each regenerator pins every random source so output is byte-identical
across runs against a fixed upstream RNS / LXMF version.

- announces.json: two vectors (no-ratchet + with-ratchet) signed by
  Alice. Determinism via patched Identity.get_random_hash + module-
  local time.time shim inside RNS.Destination.

- lxmf.json: two opportunistic-LXMF vectors Alice -> Bob, captures
  full plaintext (S5.2 layout) plus Token-encrypted ciphertext (S3).
  Determinism via fixed LXMessage.timestamp, ephemeral X25519 priv,
  and Token CBC IV.

- links.json: full Link handshake — LINKREQUEST + LRPROOF wire bytes,
  derived link_id, ECDH shared secret, and HKDF-derived session key
  that both initiator and responder MUST agree on. Determinism via
  three queued ephemeral priv-key blobs (initiator X25519, initiator
  Ed25519, responder X25519) consumed in source-call order at
  RNS/Link.py:285, :286, :278.

Status table in test-vectors/README.md and tools/README.md updated to
reflect the completed bootstrap. todo.md cleaned up to reflect actual
state (the previous "Open ⚠️ items needing a runtime verifier" section
was stale — all three verifiers were completed earlier).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Rob 2026-05-04 21:56:44 -04:00
commit 038e39401f
9 changed files with 1028 additions and 55 deletions
Download patch file Download diff file Expand all files Collapse all files

10
test-vectors/README.md
View file

@ -4,12 +4,14 @@ Known-good byte sequences that any Reticulum-compatible implementation should be
## Status
Partially populated against RNS 1.2.0:
Populated against RNS 1.2.0 / LXMF 0.9.6:
- ✅ `identities.json` — Alice + Bob identity vectors (regenerator: `../tools/regen_identities.py`, verifier: `../tools/verify_destination_hash.py`).
- ⏳ `announces.json` — not yet populated.
- ⏳ `lxmf.json` — not yet populated.
- ⏳ `links.json` — not yet populated.
- ✅ `announces.json` — two announce vectors (no-ratchet + with-ratchet) signed by Alice (regenerator: `../tools/regen_announces.py`, verifier: `../tools/verify_announce_roundtrip.py`).
- ✅ `lxmf.json` — two opportunistic-LXMF vectors Alice → Bob (regenerator: `../tools/regen_lxmf.py`, verifier: `../tools/verify_lxmf_opportunistic.py`).
- ✅ `links.json` — full Link handshake vector (LINKREQUEST + LRPROOF + derived session key) Alice → Bob (regenerator: `../tools/regen_links.py`, verifier: `../tools/verify_link_handshake.py`).
All four files are byte-deterministic across runs: regenerators pin every random source (ephemeral keys, IVs, `random_hash` prefix + timestamp, LXMF timestamp) so the output is reproducible against a fixed upstream RNS / LXMF version.
See [`../agent.md`](../agent.md) §5 and [`../todo.md`](../todo.md) for the remaining bootstrap task list.

88
test-vectors/announces.json Normal file
View file

@ -0,0 +1,88 @@
{
"_about": "Announce test vectors. Each entry's `expected.wire_bytes_hex` is the full packed announce bytes (header + body). Drop them into a fresh RNS.Packet via the inbound path and call RNS.Identity.validate_announce; it MUST accept. The `expected.fields` block decomposes the wire bytes into named slices per SPEC.md S4.1 for human inspection. Regenerate with the script in `generator_script`.",
"vectors": [
{
"label": "alice_lxmf_no_ratchet",
"context_flag": 0,
"with_ratchet": false,
"inputs": {
"identity_label": "alice",
"destination_full_name": "vectors.alice_announce_no_ratchet",
"random_hash_prefix_hex": "a1a2a3a4a5",
"random_hash_timestamp": 1700000000,
"ratchet_priv_hex": null,
"app_data_msgpack_hex": "92c409416c6963655465737400",
"app_data_decoded": [
"AliceTest",
0
]
},
"expected": {
"destination_hash_hex": "d9587f0be518490591c181755404d851",
"wire_bytes_hex": "0100d9587f0be518490591c181755404d8510076fce269b2356a51b6a832a1a25099155acb20733b453f9538aaa8069e854d5a780708b44424373474ee1607c3f2b4a1cd5643de508e106e6b8cf4a10f00ec7c8b5739ff0fe7afaf7157a1a2a3a4a5006553f1009b0f121c51fda21cbce043b5b9d89b09817f29d320d2027c0f6c67144ace9d577722791e9ca1c5d24678ced4166862d77650756a98369c48a8455865c279e20092c409416c6963655465737400",
"fields": {
"header_flags_hex": "01",
"header_hops_hex": "00",
"header_dest_hash_hex": "d9587f0be518490591c181755404d851",
"header_context_hex": "00",
"body_public_key_hex": "76fce269b2356a51b6a832a1a25099155acb20733b453f9538aaa8069e854d5a780708b44424373474ee1607c3f2b4a1cd5643de508e106e6b8cf4a10f00ec7c",
"body_name_hash_hex": "8b5739ff0fe7afaf7157",
"body_random_hash_hex": "a1a2a3a4a5006553f100",
"body_signature_hex": "9b0f121c51fda21cbce043b5b9d89b09817f29d320d2027c0f6c67144ace9d577722791e9ca1c5d24678ced4166862d77650756a98369c48a8455865c279e200",
"body_app_data_hex": "92c409416c6963655465737400"
}
},
"rns_version_at_generation": "1.2.0",
"generator_script": "tools/regen_announces.py",
"verifies_spec_sections": [
"2.1",
"4.1",
"4.2",
"4.3",
"4.5"
]
},
{
"label": "alice_lxmf_with_ratchet",
"context_flag": 1,
"with_ratchet": true,
"inputs": {
"identity_label": "alice",
"destination_full_name": "vectors.alice_announce_with_ratchet",
"random_hash_prefix_hex": "a1a2a3a4a5",
"random_hash_timestamp": 1700000000,
"ratchet_priv_hex": "b1b2b3b4b5b6b7b8b9babbbcbdbebfc0c1c2c3c4c5c6c7c8c9cacbcccdcecfd0",
"app_data_msgpack_hex": "92c409416c6963655465737400",
"app_data_decoded": [
"AliceTest",
0
]
},
"expected": {
"destination_hash_hex": "141410d233872609cf7b9f075afb4ebb",
"wire_bytes_hex": "2100141410d233872609cf7b9f075afb4ebb0076fce269b2356a51b6a832a1a25099155acb20733b453f9538aaa8069e854d5a780708b44424373474ee1607c3f2b4a1cd5643de508e106e6b8cf4a10f00ec7c5130f0a9b2e01f693bd0a1a2a3a4a5006553f100cd700e88f9e99b19c1a8a8dcd58182fd101e5e032a69ce317fde23e8ee265c51e4985b2edb0694b51ddcb9e1aa73f60acd297bf8dd087056f90c2c9ee1e47587feef3b5f6f18de160bad45e49abe5f8c7d74ccb893e207061136f5222434620392c409416c6963655465737400",
"fields": {
"header_flags_hex": "21",
"header_hops_hex": "00",
"header_dest_hash_hex": "141410d233872609cf7b9f075afb4ebb",
"header_context_hex": "00",
"body_public_key_hex": "76fce269b2356a51b6a832a1a25099155acb20733b453f9538aaa8069e854d5a780708b44424373474ee1607c3f2b4a1cd5643de508e106e6b8cf4a10f00ec7c",
"body_name_hash_hex": "5130f0a9b2e01f693bd0",
"body_random_hash_hex": "a1a2a3a4a5006553f100",
"body_signature_hex": "e4985b2edb0694b51ddcb9e1aa73f60acd297bf8dd087056f90c2c9ee1e47587feef3b5f6f18de160bad45e49abe5f8c7d74ccb893e207061136f52224346203",
"body_app_data_hex": "92c409416c6963655465737400",
"body_ratchet_pub_hex": "cd700e88f9e99b19c1a8a8dcd58182fd101e5e032a69ce317fde23e8ee265c51"
}
},
"rns_version_at_generation": "1.2.0",
"generator_script": "tools/regen_announces.py",
"verifies_spec_sections": [
"2.1",
"4.1",
"4.2",
"4.3",
"4.5"
]
}
]
}

46
test-vectors/links.json Normal file
View file

@ -0,0 +1,46 @@
{
"_about": "Link handshake test vectors. Each vector records a full Reticulum Link handshake: LINKREQUEST (initiator -> responder) and LRPROOF (responder -> initiator). The ephemeral X25519/Ed25519 keys are pinned via the `inputs.*_priv_hex` blobs; both Ed25519 signatures are RFC 8032 deterministic so the resulting wire bytes are reproducible. A clean-room implementation can verify by: (a) packing a LINKREQUEST from the recorded initiator ephemerals and confirming bytes match `linkrequest_raw_hex`; (b) computing `link_id` per SPEC.md S6.3 (N=2 for HEADER_1) and matching `link_id_hex`; (c) packing an LRPROOF as the responder, with bob's identity Ed25519 sig over `link_id || responder_X25519_pub || responder_long_term_Ed25519_pub || signalling`, and matching `lrproof_raw_hex`; (d) running ECDH+HKDF on either side and matching `derived_key_hex`. Regenerate with `generator_script`.",
"vectors": [
{
"label": "alice_to_bob_aes256cbc",
"inputs": {
"initiator_identity_label": "alice",
"responder_identity_label": "bob",
"destination_full_name": "vectors.link",
"initiator_x25519_priv_hex": "1111111111111111111111111111111111111111111111111111111111111111",
"initiator_ed25519_priv_hex": "2222222222222222222222222222222222222222222222222222222222222222",
"responder_x25519_priv_hex": "3333333333333333333333333333333333333333333333333333333333333333",
"mode": "MODE_AES256_CBC (0x01)"
},
"expected": {
"linkrequest_raw_hex": "02008c670c64308e0325ea0fd7c72787449d007b4e909bbe7ffe44c465a220037d608ee35897d31ef972f07f74892cb0f73f13a09aa5f47a6759802ff955f8dc2d2a14a5c99d23be97f864127ff9383455a4f02001f4",
"linkrequest_body_hex": "7b4e909bbe7ffe44c465a220037d608ee35897d31ef972f07f74892cb0f73f13a09aa5f47a6759802ff955f8dc2d2a14a5c99d23be97f864127ff9383455a4f02001f4",
"linkrequest_fields": {
"initiator_x25519_pub_hex": "7b4e909bbe7ffe44c465a220037d608ee35897d31ef972f07f74892cb0f73f13",
"initiator_ed25519_pub_hex": "a09aa5f47a6759802ff955f8dc2d2a14a5c99d23be97f864127ff9383455a4f0",
"signalling_hex": "2001f4"
},
"link_id_hex": "7ee5fe3e4952c9ac4519b537f6278474",
"lrproof_raw_hex": "0f007ee5fe3e4952c9ac4519b537f6278474ff1de2168a36a816163aec0bb0749ff6792f78eb4f7b39156f8ee5c8693e83ebd67439ac28d9e4603334428713154edd04395b0b8acec2f703c05c3d38af133e0c7b0d47d93427f8311160781c7c733fd89f88970aef490d8aa0ee19a4cb8a1b142001f4",
"lrproof_body_hex": "1de2168a36a816163aec0bb0749ff6792f78eb4f7b39156f8ee5c8693e83ebd67439ac28d9e4603334428713154edd04395b0b8acec2f703c05c3d38af133e0c7b0d47d93427f8311160781c7c733fd89f88970aef490d8aa0ee19a4cb8a1b142001f4",
"lrproof_fields": {
"signature_hex": "1de2168a36a816163aec0bb0749ff6792f78eb4f7b39156f8ee5c8693e83ebd67439ac28d9e4603334428713154edd04395b0b8acec2f703c05c3d38af133e0c",
"responder_x25519_pub_hex": "7b0d47d93427f8311160781c7c733fd89f88970aef490d8aa0ee19a4cb8a1b14",
"signalling_hex": "2001f4"
},
"shared_secret_hex": "5bf22caf31c0316785b0b9bc60e56d48582ce59435ce5b3c028052be42631e0f",
"derived_key_hex": "d4c8238d23a1810c3dbe4caec15253d5a86d7fe6afa8dfa76f915579723fd88cbcd2ab3a0cd96f5b6ffd8abec8307f05cd791dc9c4fca900f706b0313a51ab65",
"mtu": 500,
"mode": 1
},
"rns_version_at_generation": "1.2.0",
"generator_script": "tools/regen_links.py",
"verifies_spec_sections": [
"6.1",
"6.2",
"6.3",
"6.6"
]
}
]
}

76
test-vectors/lxmf.json Normal file
View file

@ -0,0 +1,76 @@
{
"_about": "Opportunistic LXMF test vectors. `expected.lxmf_packed_hex` is the full plaintext body per SPEC.md S5.2: dest(16) || src(16) || sig(64) || msgpack. `expected.opportunistic_plaintext_hex` is the same with the leading dest_hash stripped (S5.1 wire form). `expected.token_ciphertext_hex` is the deterministic Token encryption of the opportunistic plaintext (S3) using the fixed ephemeral X25519 priv + IV. To verify against upstream: decrypt with Bob's identity, re-prepend Bob's dest_hash, then call LXMessage.unpack_from_bytes \u2014 it MUST succeed with signature_validated == True and title/content/fields matching the `inputs` block. Regenerate with `generator_script`.",
"vectors": [
{
"label": "alice_to_bob_simple",
"inputs": {
"src_identity_label": "alice",
"dst_identity_label": "bob",
"title_utf8": "hello",
"content_utf8": "hi bob",
"fields": {},
"lxmf_timestamp": 1700000000.0,
"ephemeral_x25519_priv_hex": "d1d2d3d4d5d6d7d8d9dadbdcdddedfe0e1e2e3e4e5e6e7e8e9eaebecedeeeff0",
"token_iv_hex": "11223344556677889900aabbccddeeff"
},
"expected": {
"lxmf_packed_hex": "9695d17f22fa6e45d2b0cd3439a7ca7ec33c40a5b030596d95617dc4ca163aaedc758d54b21a2ca01a8fcc3e21c45eb60918d2dc64508037ce640e000a295a81951e5a7d0f8fedb90ec4df0b0a05b437b43d6692c9dd7faa98c4b679935a940e94cb41d954fc40000000c40568656c6c6fc406686920626f6280",
"opportunistic_plaintext_hex": "c33c40a5b030596d95617dc4ca163aaedc758d54b21a2ca01a8fcc3e21c45eb60918d2dc64508037ce640e000a295a81951e5a7d0f8fedb90ec4df0b0a05b437b43d6692c9dd7faa98c4b679935a940e94cb41d954fc40000000c40568656c6c6fc406686920626f6280",
"token_ciphertext_hex": "21c3332b61be6a7b6ab8461e155651b17501b6e07532ecf9ab6661bd5a2ca57511223344556677889900aabbccddeeffefa9d24b76e1adf393cfd588214e236e219743697af96912b8eae84f3a1f28a3d68abd62f3e42c6944015c3d00e5e7aa8af732123d079ab10353597669c8cd3ba57cfae3a28ea1a99a44e0b492ba5deedd23232d2edab78fa037967757808c8578496aee7b21c70ce2476c54540d96d928e8ddf35c6bfb5d76261c07f1bb48af9d7bec8261cd30f3b03986614ba93173",
"fields_layout": {
"destination_hash_hex": "9695d17f22fa6e45d2b0cd3439a7ca7e",
"source_hash_hex": "c33c40a5b030596d95617dc4ca163aae",
"signature_hex": "dc758d54b21a2ca01a8fcc3e21c45eb60918d2dc64508037ce640e000a295a81951e5a7d0f8fedb90ec4df0b0a05b437b43d6692c9dd7faa98c4b679935a940e",
"msgpack_payload_hex": "94cb41d954fc40000000c40568656c6c6fc406686920626f6280"
}
},
"rns_version_at_generation": "1.2.0",
"lxmf_version_at_generation": "0.9.6",
"generator_script": "tools/regen_lxmf.py",
"verifies_spec_sections": [
"3",
"5.1",
"5.2",
"5.5",
"5.6"
]
},
{
"label": "alice_to_bob_with_fields",
"inputs": {
"src_identity_label": "alice",
"dst_identity_label": "bob",
"title_utf8": "meeting",
"content_utf8": "see attached",
"fields": {
"1": "k1",
"2": 42
},
"lxmf_timestamp": 1700000000.0,
"ephemeral_x25519_priv_hex": "d1d2d3d4d5d6d7d8d9dadbdcdddedfe0e1e2e3e4e5e6e7e8e9eaebecedeeeff0",
"token_iv_hex": "11223344556677889900aabbccddeeff"
},
"expected": {
"lxmf_packed_hex": "9695d17f22fa6e45d2b0cd3439a7ca7ec33c40a5b030596d95617dc4ca163aaef23eb2c325e59493c187b8fa1cc0efb6306f3eeed159a33aeec954576cbd354451caaa176aff84b57cc154b4e4113197b5eb92f1ec6a0e7635fa3d0508c67e0194cb41d954fc40000000c4076d656574696e67c40c7365652061747461636865648201a26b31022a",
"opportunistic_plaintext_hex": "c33c40a5b030596d95617dc4ca163aaef23eb2c325e59493c187b8fa1cc0efb6306f3eeed159a33aeec954576cbd354451caaa176aff84b57cc154b4e4113197b5eb92f1ec6a0e7635fa3d0508c67e0194cb41d954fc40000000c4076d656574696e67c40c7365652061747461636865648201a26b31022a",
"token_ciphertext_hex": "21c3332b61be6a7b6ab8461e155651b17501b6e07532ecf9ab6661bd5a2ca57511223344556677889900aabbccddeeffefa9d24b76e1adf393cfd588214e236eac9dde2777640fdd62e86256d9ddac812eeda9277056b5652b9d83ca7d0da7203a5f51b69f7d35a58da4b6a13562a12145a98810d1b89fec9a70e947c50eee6482f3fda4165fd6eef25819fc5093d5f2aaaf7b8911689a3eeaf0131816bac4041923df9e64a807d9809c35cc7bed027de94dc42a7af10261f14053dc62d77d54e66f60dfb83763a6f66798c51eeabcd2",
"fields_layout": {
"destination_hash_hex": "9695d17f22fa6e45d2b0cd3439a7ca7e",
"source_hash_hex": "c33c40a5b030596d95617dc4ca163aae",
"signature_hex": "f23eb2c325e59493c187b8fa1cc0efb6306f3eeed159a33aeec954576cbd354451caaa176aff84b57cc154b4e4113197b5eb92f1ec6a0e7635fa3d0508c67e01",
"msgpack_payload_hex": "94cb41d954fc40000000c4076d656574696e67c40c7365652061747461636865648201a26b31022a"
}
},
"rns_version_at_generation": "1.2.0",
"lxmf_version_at_generation": "0.9.6",
"generator_script": "tools/regen_lxmf.py",
"verifies_spec_sections": [
"3",
"5.1",
"5.2",
"5.5",
"5.6"
]
}
]
}

72
todo.md
View file

@ -28,26 +28,25 @@ Outstanding work for the spec repo.
- [x] **Bootstrap `test-vectors/identities.json`** — Alice + Bob
identities populated against RNS 1.2.0. Regenerator at
`tools/regen_identities.py`.
- [ ] **Bootstrap remaining test-vectors files** (`announces.json`,
`lxmf.json`, `links.json`) with the existing vectors from
`reticulum-mobile-app/reference/test-vectors.json`. Convert to
the proposed JSON format documented in `test-vectors/README.md`,
adding the regenerator scripts so future contributors can
verify vectors against newer upstream RNS releases.
- [x] **Bootstrap `test-vectors/announces.json`** — two vectors
(no-ratchet + with-ratchet) signed by Alice. Regenerator at
`tools/regen_announces.py` (deterministic via patched
`Identity.get_random_hash` + module-local `time.time` shim).
- [x] **Bootstrap `test-vectors/lxmf.json`** — two opportunistic
LXMF vectors Alice → Bob, full plaintext + Token-encrypted
ciphertext. Regenerator at `tools/regen_lxmf.py` (deterministic
via patched `LXMessage.timestamp`, ephemeral X25519, and
Token CBC IV).
- [x] **Bootstrap `test-vectors/links.json`** — Link handshake
vector with deterministic ephemerals. Regenerator at
`tools/regen_links.py`. Records LINKREQUEST + LRPROOF wire
bytes plus the derived session key both sides must agree on.
- [ ] **Write the priority verifier scripts** listed in
`tools/README.md`, in this order (highest interop value first):
1. [x] `verify_destination_hash.py` — pure-function check, no RNS state needed
2. [x] `verify_packet_header.py` — bit layout + HEADER_1/HEADER_2 round-trip + originator HEADER_1→HEADER_2 conversion
3. [ ] `verify_announce_roundtrip.py` — closes the SPEC.md §4 gap (partial coverage in `verify_announce_app_data.py`)
4. [ ] `verify_token_crypto.py` — closes SPEC.md §3 gap
5. [ ] `verify_lxmf_opportunistic.py` — closes SPEC.md §5 gap
6. [ ] `verify_link_handshake.py` — closes SPEC.md §6 gap
7. [x] `verify_path_request.py` — closes SPEC.md §7.1, §7.2 gaps
8. [ ] `verify_msgpack_quirk.py` — closes SPEC.md §9.3 gap
Each verifier should remove its corresponding `⚠️ UNVERIFIED` /
`🔮 SPECULATION` callout in `SPEC.md` (per `agent.md` §1).
- [x] **Write the priority verifier scripts** listed in
`tools/README.md` — all eight done plus three follow-ons
(`verify_proof_packet.py`, `verify_rnode_split.py`,
`verify_stamps.py`, `verify_ratchet_dedup.py`). Status table
lives in `tools/README.md`.
## Open `⚠️ UNVERIFIED` items in SPEC.md
@ -86,38 +85,9 @@ to remove their markers:
## Open `⚠️` items needing a runtime verifier
- [ ] **`tools/verify_proof_packet.py` to lock in §6.5.** Run two
side-by-side scenarios against upstream RNS: opportunistic DATA
with `use_implicit_proof = True` (default) and with `= False`,
capture the resulting PROOF packet's body length, and assert
it's 64 / 96 respectively with the matching content layout.
Also exercise a Link DATA proof and confirm it's always 96B
regardless of the config setting. Lock in the §6.5 wire shapes.
- [ ] **`tools/verify_rnode_split.py` to lock in §8.3.** The RNode
air-frame split-packet protocol is now documented in SPEC.md §8.3
against direct citations in `markqvist/RNode_Firmware/Framing.h`,
`Config.h`, `Utilities.h`, and `RNode_Firmware.ino`, plus the
clean-room reimplementation in `thatSFguy/reticulum-lora-repeater/src/Radio.cpp`.
A runtime verifier would: build a 300-byte synthetic Reticulum
packet, run it through a Python implementation of the TX-side
header rules, and confirm the byte-level frames match what
`RNode_Firmware.ino:716-742` would emit (header byte high nibble
random + low-nibble FLAG_SPLIT bit, both frames sharing the same
header, split point at 255 bytes total per LoRa frame). RX-side
verifier should drive the state-table at SPEC.md §8.3 and confirm
the four reassembly cases.
- [ ] **Lock in the §6.2 / §6.3 corrections with `verify_link_handshake.py`.**
The wire-byte order of the LRPROOF body (`signature || responder_X25519_pub || signalling`,
not `link_id || responder_X25519_pub || signature || signalling`) and
the `link_id` derivation offsets (`N=2` for HEADER_1, `N=18` for HEADER_2,
not 18/34) were corrected against direct upstream source citations
(`RNS/Link.py:376`, `RNS/Packet.py:354-361`) in `SPEC.md` §6.2/§6.3
while writing `flows/send-link-lxmf.md`. They are source-cited but
not yet exercised by a runtime verifier. Add `tools/verify_link_handshake.py`
that drives an upstream LINKREQUEST → LRPROOF → ACTIVE handshake and
asserts byte-level layouts + `link_id` invariance under HEADER_1↔HEADER_2.
- [x] **`tools/verify_proof_packet.py` locks in §6.5.** Done.
- [x] **`tools/verify_rnode_split.py` locks in §8.3.** Done.
- [x] **`tools/verify_link_handshake.py` locks in §6.2 / §6.3.** Done.
## Spec gaps for a functional client (priority-ordered)

3
tools/README.md
View file

@ -37,5 +37,8 @@ Populated against RNS 1.2.0 / LXMF 0.9.6:
| `verify_stamps.py` | §5.7 — workblock determinism, PoW stamp search/validate, ticket shortcut | ✅ |
| `verify_ratchet_dedup.py` | §7.3 / §4.5 step 6.3 — confirms replay defence is keyed on `random_blob`, NOT on `(dest_hash, ratchet_pub)` | ✅ |
| `regen_identities.py` | regenerates `test-vectors/identities.json` | ✅ |
| `regen_announces.py` | regenerates `test-vectors/announces.json` (deterministic announce wire bytes, with and without ratchet) | ✅ |
| `regen_lxmf.py` | regenerates `test-vectors/lxmf.json` (deterministic opportunistic-LXMF plaintext + Token ciphertext) | ✅ |
| `regen_links.py` | regenerates `test-vectors/links.json` (deterministic LINKREQUEST + LRPROOF + derived session key) | ✅ |
See [`../agent.md`](../agent.md) §5 and [`../todo.md`](../todo.md) for the remaining priority order.

243
tools/regen_announces.py Normal file
View file

@ -0,0 +1,243 @@
"""
Regenerator for test-vectors/announces.json.
Builds two deterministic announce wire-byte vectors using Alice's
identity from `identities.json`:
1. `alice_lxmf_no_ratchet` — context_flag=0, app_data = LXMF
2-element form `[display_name_bytes, stamp_cost]` (S4.3)
2. `alice_lxmf_with_ratchet` — context_flag=1, fixed ratchet priv,
same app_data
Determinism inputs:
- Identity = Alice (private_key from `identities.json`); Ed25519
signing is deterministic per RFC 8032 so the signature byte
sequence is reproducible from the same plaintext + key.
- `random_hash[0:5]` = patched constant prefix.
- `random_hash[5:10]` = patched fixed unix-seconds value (we override
`time.time` inside `RNS.Destination` for the duration of the
announce build).
- `ratchet_priv` = fixed 32 bytes (vector #2 only).
- `app_data` = `umsgpack.packb([b"AliceTest", 0])` per S4.3.
After build we run `RNS.Identity.validate_announce` on the produced
packet and assert it accepts; this proves the recorded bytes are
upstream-valid.
Run from repo root:
python tools/regen_announces.py
Updates `test-vectors/announces.json` in place. Exit 0 on success.
"""
from __future__ import annotations
import json
import os
import sys
import tempfile
import RNS
from RNS.vendor import umsgpack
REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
OUT_PATH = os.path.join(REPO_ROOT, "test-vectors", "announces.json")
IDS_PATH = os.path.join(REPO_ROOT, "test-vectors", "identities.json")
# Stable inputs. These are vector-deterministic; do not change without
# regenerating every announce vector.
FIXED_RANDOM_PREFIX = bytes.fromhex("a1a2a3a4a5") # random_hash[0:5]
FIXED_TIMESTAMP = 1700000000 # 2023-11-14, BE-uint40 -> random_hash[5:10]
FIXED_RATCHET_PRIV = bytes.fromhex("b1b2b3b4b5b6b7b8b9babbbcbdbebfc0c1c2c3c4c5c6c7c8c9cacbcccdcecfd0")
FIXED_APP_DATA = umsgpack.packb([b"AliceTest", 0]) # S4.3 2-element form
def fail(msg: str) -> None:
print(f"FAIL: {msg}")
sys.exit(1)
def init_minimal_rns():
cfg_dir = tempfile.mkdtemp(prefix="rns-regen-announces-")
cfg_path = os.path.join(cfg_dir, "config")
with open(cfg_path, "w", encoding="utf-8") as f:
f.write("[reticulum]\nenable_transport = No\nshare_instance = No\n")
return RNS.Reticulum(configdir=cfg_dir, loglevel=0)
def load_alice() -> RNS.Identity:
with open(IDS_PATH, "r", encoding="utf-8") as f:
ids = json.load(f)
alice = next(v for v in ids["vectors"] if v["label"] == "alice")
prv_bytes = bytes.fromhex(alice["inputs"]["private_key_hex"])
identity = RNS.Identity.from_bytes(prv_bytes)
if identity is None:
fail("Identity.from_bytes returned None for Alice")
return identity, alice
def build_deterministic_announce(identity, with_ratchet: bool) -> bytes:
"""Drive `Destination.announce(send=False)` with patched time/random
sources so the wire bytes are reproducible. Returns packed raw bytes."""
import sys as _sys
# `RNS.Identity` and `RNS.Destination` are the classes (re-exported in
# RNS/__init__.py); the underlying modules live in sys.modules.
dest_mod = _sys.modules["RNS.Destination"]
id_cls = RNS.Identity # the class, where get_random_hash is defined
# Patch the get_random_hash classmethod so random_hash[0:5] is fixed.
# Destination.announce calls it once per announce (Destination.py:282).
real_get_random_hash = id_cls.get_random_hash
def patched_get_random_hash():
return FIXED_RANDOM_PREFIX + b"\x00" * 27
id_cls.get_random_hash = staticmethod(patched_get_random_hash)
# Patch time.time inside the Destination module so
# int(time.time()).to_bytes(5,'big') -> FIXED_TIMESTAMP. Replacing
# `dest_mod.time.time` would mutate the global `time` module; instead
# swap the module-level `time` reference itself.
real_time_module = dest_mod.time
class _FixedTime:
@staticmethod
def time(): return float(FIXED_TIMESTAMP)
dest_mod.time = _FixedTime
try:
# Use a dedicated aspect per vector so the Destination cache
# doesn't reject the second registration in this same process.
aspect = "alice_announce_with_ratchet" if with_ratchet else "alice_announce_no_ratchet"
dest = RNS.Destination(identity, RNS.Destination.IN, RNS.Destination.SINGLE,
"vectors", aspect)
if with_ratchet:
# Force the ratchet without going through enable_ratchets (which
# writes to disk and could pick up stale state). Setting
# latest_ratchet_time = now-eps (>= now-INTERVAL) makes
# rotate_ratchets a no-op so our fixed priv is preserved.
dest.ratchets = [FIXED_RATCHET_PRIV]
dest.ratchet_interval = RNS.Destination.RATCHET_INTERVAL
dest.latest_ratchet_time = float(FIXED_TIMESTAMP)
dest.path_responses = {}
# else: leave dest.ratchets = None so context_flag=0
pkt = dest.announce(app_data=FIXED_APP_DATA, send=False)
pkt.pack()
# Sanity: validate_announce must accept the packet we just built.
# We must rebuild the Packet from raw bytes via the inbound path
# because packet.unpack mutates state.
if not pkt.unpack():
fail(f"Packet.unpack returned False ({aspect})")
if not RNS.Identity.validate_announce(pkt):
fail(f"validate_announce rejected freshly-built vector ({aspect})")
return pkt.raw, dest.hash
finally:
id_cls.get_random_hash = staticmethod(real_get_random_hash)
dest_mod.time = real_time_module
def split_body(raw: bytes, with_ratchet: bool) -> dict:
"""Split announce wire bytes into the named fields per S4.1 so the
JSON vector is human-inspectable."""
KEYSIZE, NAME_HASH, RANDOM_HASH, RATCHET, SIG = 64, 10, 10, 32, 64
flags = raw[0]
hops = raw[1]
dst = raw[2:18]
ctx = raw[18]
body = raw[19:]
pub = body[0:KEYSIZE]
nm = body[KEYSIZE:KEYSIZE+NAME_HASH]
rh = body[KEYSIZE+NAME_HASH:KEYSIZE+NAME_HASH+RANDOM_HASH]
if with_ratchet:
rt = body[KEYSIZE+NAME_HASH+RANDOM_HASH:KEYSIZE+NAME_HASH+RANDOM_HASH+RATCHET]
sigs = KEYSIZE+NAME_HASH+RANDOM_HASH+RATCHET
else:
rt = b""
sigs = KEYSIZE+NAME_HASH+RANDOM_HASH
sig = body[sigs:sigs+SIG]
ad = body[sigs+SIG:]
fields = {
"header_flags_hex": bytes([flags]).hex(),
"header_hops_hex": bytes([hops]).hex(),
"header_dest_hash_hex": dst.hex(),
"header_context_hex": bytes([ctx]).hex(),
"body_public_key_hex": pub.hex(),
"body_name_hash_hex": nm.hex(),
"body_random_hash_hex": rh.hex(),
"body_signature_hex": sig.hex(),
"body_app_data_hex": ad.hex(),
}
if with_ratchet:
fields["body_ratchet_pub_hex"] = rt.hex()
return fields
def main():
print(f"regen_announces.py against RNS {RNS.__version__}")
init_minimal_rns()
try:
identity, alice = load_alice()
vectors = []
for label, with_ratchet in [
("alice_lxmf_no_ratchet", False),
("alice_lxmf_with_ratchet", True),
]:
raw, dest_hash = build_deterministic_announce(identity, with_ratchet)
fields = split_body(raw, with_ratchet)
vectors.append({
"label": label,
"context_flag": 1 if with_ratchet else 0,
"with_ratchet": with_ratchet,
"inputs": {
"identity_label": "alice",
"destination_full_name": "vectors." + ("alice_announce_with_ratchet" if with_ratchet else "alice_announce_no_ratchet"),
"random_hash_prefix_hex": FIXED_RANDOM_PREFIX.hex(),
"random_hash_timestamp": FIXED_TIMESTAMP,
"ratchet_priv_hex": FIXED_RATCHET_PRIV.hex() if with_ratchet else None,
"app_data_msgpack_hex": FIXED_APP_DATA.hex(),
"app_data_decoded": ["AliceTest", 0],
},
"expected": {
"destination_hash_hex": dest_hash.hex(),
"wire_bytes_hex": raw.hex(),
"fields": fields,
},
"rns_version_at_generation": RNS.__version__,
"generator_script": "tools/regen_announces.py",
"verifies_spec_sections": ["2.1", "4.1", "4.2", "4.3", "4.5"],
})
payload = {
"_about": (
"Announce test vectors. Each entry's `expected.wire_bytes_hex` "
"is the full packed announce bytes (header + body). Drop them "
"into a fresh RNS.Packet via the inbound path and call "
"RNS.Identity.validate_announce; it MUST accept. The "
"`expected.fields` block decomposes the wire bytes into "
"named slices per SPEC.md S4.1 for human inspection. "
"Regenerate with the script in `generator_script`."
),
"vectors": vectors,
}
with open(OUT_PATH, "w", encoding="utf-8", newline="\n") as f:
json.dump(payload, f, indent=2, sort_keys=False)
f.write("\n")
print(f"Wrote {OUT_PATH} with {len(vectors)} vectors")
print("ALL PASS")
finally:
try: RNS.Reticulum.exit_handler()
except Exception: pass
if __name__ == "__main__":
main()

295
tools/regen_links.py Normal file
View file

@ -0,0 +1,295 @@
"""
Regenerator for test-vectors/links.json.
Builds a deterministic Link handshake vector pair (LINKREQUEST + LRPROOF)
between Alice (initiator) and Bob (responder) using identities from
`identities.json`.
Determinism inputs:
- Alice and Bob long-term identities (fixed via identities.json).
- Initiator's ephemeral X25519 private key (`Link.prv`,
`RNS/Link.py:285`) — fixed.
- Initiator's ephemeral Ed25519 private key (`Link.sig_prv`,
`RNS/Link.py:286`) — fixed. (This is *not* Alice's long-term key;
Link's signing key is generated fresh per handshake.)
- Responder's ephemeral X25519 private key (`Link.prv`,
`RNS/Link.py:278`) — fixed.
- Both Ed25519 signatures are deterministic per RFC 8032.
Outputs recorded per vector:
- `linkrequest_raw_hex` — full packed LINKREQUEST packet bytes
- `linkrequest_body_hex` — initiator_X25519_pub || initiator_Ed25519_pub
[|| signalling] (S6.1)
- `link_id_hex` — derived per S6.3 (N=2 for HEADER_1)
- `lrproof_raw_hex` — full packed LRPROOF packet bytes
- `lrproof_body_hex` — signature || responder_X25519_pub
[|| signalling] (S6.2)
- `derived_key_hex` — HKDF output for AES256_CBC mode
(length=64, salt=link_id, context=context)
— both sides must arrive at the same
bytes after handshake().
Run from repo root:
python tools/regen_links.py
Updates `test-vectors/links.json` in place. Exit 0 on success.
"""
from __future__ import annotations
import hashlib
import json
import os
import sys
import tempfile
import RNS
from RNS.Link import Link
REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
OUT_PATH = os.path.join(REPO_ROOT, "test-vectors", "links.json")
IDS_PATH = os.path.join(REPO_ROOT, "test-vectors", "identities.json")
# Stable inputs. Three distinct fixed priv key blobs for the three
# ephemeral generations the handshake performs.
INITIATOR_X25519_PRIV = bytes.fromhex("11"*32)
INITIATOR_ED25519_PRIV = bytes.fromhex("22"*32)
RESPONDER_X25519_PRIV = bytes.fromhex("33"*32)
def fail(msg: str) -> None:
print(f"FAIL: {msg}")
sys.exit(1)
def init_minimal_rns():
cfg_dir = tempfile.mkdtemp(prefix="rns-regen-links-")
cfg_path = os.path.join(cfg_dir, "config")
with open(cfg_path, "w", encoding="utf-8") as f:
f.write("[reticulum]\nenable_transport = No\nshare_instance = No\n")
return RNS.Reticulum(configdir=cfg_dir, loglevel=0)
def load_identities():
with open(IDS_PATH, "r", encoding="utf-8") as f:
ids = json.load(f)
alice = next(v for v in ids["vectors"] if v["label"] == "alice")
bob = next(v for v in ids["vectors"] if v["label"] == "bob")
alice_id = RNS.Identity.from_bytes(bytes.fromhex(alice["inputs"]["private_key_hex"]))
bob_id = RNS.Identity.from_bytes(bytes.fromhex(bob ["inputs"]["private_key_hex"]))
if alice_id is None or bob_id is None:
fail("Identity.from_bytes returned None for one of alice/bob")
return alice_id, bob_id
class _StaticPrivQueue:
"""Wraps an upstream X25519/Ed25519 dispatcher class so that
`.generate()` returns a private key seeded with the next blob in a
fixed queue. Each consumed blob is removed; if the queue underflows
we delegate to the real `.generate()` so non-handshake code paths
still work."""
def __init__(self, real_cls, blobs):
self._real_cls = real_cls
self._blobs = list(blobs)
def generate(self):
if self._blobs:
blob = self._blobs.pop(0)
return self._real_cls.from_private_bytes(blob)
return self._real_cls.generate()
# Defer everything else (from_private_bytes, from_public_bytes, etc.)
# to the real class so the rest of upstream's surface keeps working.
def __getattr__(self, name):
return getattr(self._real_cls, name)
def main():
print(f"regen_links.py against RNS {RNS.__version__}")
init_minimal_rns()
try:
alice_id, bob_id = load_identities()
# Bob's destination — the link target. Mark Bob's identity
# discoverable to Alice so Link initiator can find pub keys.
bob_dest_in = RNS.Destination(bob_id, RNS.Destination.IN, RNS.Destination.SINGLE,
"vectors", "link")
bob_dest_out = RNS.Destination(bob_id, RNS.Destination.OUT, RNS.Destination.SINGLE,
"vectors", "link")
RNS.Identity.remember(b"\x00"*32, bob_dest_in.hash, bob_id.get_public_key(), None)
# Patch the X25519 / Ed25519 generators that Link.__init__ uses
# to drive deterministic ephemeral keys. Order of consumption
# follows the source:
# line 285: initiator self.prv = X25519PrivateKey.generate()
# line 286: initiator self.sig_prv = Ed25519PrivateKey.generate()
# line 278: responder self.prv = X25519PrivateKey.generate()
link_mod = sys.modules["RNS.Link"]
real_X25519 = link_mod.X25519PrivateKey
real_Ed25519 = link_mod.Ed25519PrivateKey
link_mod.X25519PrivateKey = _StaticPrivQueue(real_X25519,
[INITIATOR_X25519_PRIV,
RESPONDER_X25519_PRIV])
link_mod.Ed25519PrivateKey = _StaticPrivQueue(real_Ed25519,
[INITIATOR_ED25519_PRIV])
# Capture LINKREQUEST emission. Patch Transport.outbound (not
# Packet.send) so Packet.pack() runs and packet.raw is populated.
captured_lr = {}
captured_pf = {}
real_outbound = RNS.Transport.outbound
def fake_outbound(packet):
if packet.packet_type == RNS.Packet.LINKREQUEST and "raw" not in captured_lr:
captured_lr["raw"] = packet.raw
captured_lr["data"] = packet.data
elif packet.context == RNS.Packet.LRPROOF and "raw" not in captured_pf:
captured_pf["raw"] = packet.raw
captured_pf["data"] = packet.data
return True
RNS.Transport.outbound = staticmethod(fake_outbound)
try:
# 1. Build initiator-side Link → LINKREQUEST emitted via outbound
initiator = Link(destination=bob_dest_out)
# 2. Walk the responder side by hand (Link.validate_request
# inlined). This keeps the script self-contained.
inbound = RNS.Packet(None, captured_lr["raw"])
if not inbound.unpack():
fail("Failed to unpack the captured LINKREQUEST")
inbound.destination = bob_dest_in
request_data = captured_lr["data"]
responder = Link(
owner = bob_dest_in,
peer_pub_bytes = request_data[:Link.ECPUBSIZE//2],
peer_sig_pub_bytes = request_data[Link.ECPUBSIZE//2:Link.ECPUBSIZE],
)
responder.set_link_id(inbound)
if len(request_data) == Link.ECPUBSIZE + Link.LINK_MTU_SIZE:
responder.mtu = Link.mtu_from_lr_packet(inbound) or RNS.Reticulum.MTU
responder.mode = Link.mode_from_lr_packet(inbound)
responder.destination = inbound.destination
responder.handshake() # populates responder.derived_key
responder.prove() # emits LRPROOF via outbound
finally:
RNS.Transport.outbound = real_outbound
link_mod.X25519PrivateKey = real_X25519
link_mod.Ed25519PrivateKey = real_Ed25519
# 3. Drive the initiator's handshake by feeding it the captured
# LRPROOF, so its derived_key is computed too. We assert
# initiator.derived_key == responder.derived_key.
proof_pkt = RNS.Packet(None, captured_pf["raw"])
if not proof_pkt.unpack():
fail("Failed to unpack the captured LRPROOF")
# validate_proof needs the link instance to know its destination
proof_pkt.destination = initiator
# Set up the initiator state so validate_proof can run end-to-end
import time as _time
initiator.request_time = _time.time()
if not initiator.validate_proof(proof_pkt):
# Some versions return None on success, so don't hard-fail
# solely on that — but check derived_key was populated.
pass
if initiator.derived_key is None:
fail("Initiator derived_key not populated after validate_proof")
if responder.derived_key is None:
fail("Responder derived_key not populated after handshake")
if initiator.derived_key != responder.derived_key:
fail("Initiator and responder derived_key disagree:\n"
f" initiator: {initiator.derived_key.hex()}\n"
f" responder: {responder.derived_key.hex()}")
if initiator.link_id != responder.link_id:
fail(f"link_id disagree: ini={initiator.link_id.hex()} res={responder.link_id.hex()}")
# Slice fields per S6.1 / S6.2 for human inspection
lr_data = captured_lr["data"]
ini_x25519_pub = lr_data[:32]
ini_ed25519_pub = lr_data[32:64]
lr_signalling = lr_data[64:] if len(lr_data) > 64 else b""
pf_data = captured_pf["data"]
pf_signature = pf_data[:64]
pf_responder_x25519 = pf_data[64:96]
pf_signalling = pf_data[96:] if len(pf_data) > 96 else b""
vector = {
"label": "alice_to_bob_aes256cbc",
"inputs": {
"initiator_identity_label": "alice",
"responder_identity_label": "bob",
"destination_full_name": "vectors.link",
"initiator_x25519_priv_hex": INITIATOR_X25519_PRIV.hex(),
"initiator_ed25519_priv_hex": INITIATOR_ED25519_PRIV.hex(),
"responder_x25519_priv_hex": RESPONDER_X25519_PRIV.hex(),
"mode": "MODE_AES256_CBC (0x01)",
},
"expected": {
"linkrequest_raw_hex": captured_lr["raw"].hex(),
"linkrequest_body_hex": lr_data.hex(),
"linkrequest_fields": {
"initiator_x25519_pub_hex": ini_x25519_pub.hex(),
"initiator_ed25519_pub_hex": ini_ed25519_pub.hex(),
"signalling_hex": lr_signalling.hex(),
},
"link_id_hex": initiator.link_id.hex(),
"lrproof_raw_hex": captured_pf["raw"].hex(),
"lrproof_body_hex": pf_data.hex(),
"lrproof_fields": {
"signature_hex": pf_signature.hex(),
"responder_x25519_pub_hex": pf_responder_x25519.hex(),
"signalling_hex": pf_signalling.hex(),
},
"shared_secret_hex": initiator.shared_key.hex(),
"derived_key_hex": initiator.derived_key.hex(),
"mtu": initiator.mtu,
"mode": initiator.mode,
},
"rns_version_at_generation": RNS.__version__,
"generator_script": "tools/regen_links.py",
"verifies_spec_sections": ["6.1", "6.2", "6.3", "6.6"],
}
payload = {
"_about": (
"Link handshake test vectors. Each vector records a full "
"Reticulum Link handshake: LINKREQUEST (initiator -> "
"responder) and LRPROOF (responder -> initiator). The "
"ephemeral X25519/Ed25519 keys are pinned via the "
"`inputs.*_priv_hex` blobs; both Ed25519 signatures are "
"RFC 8032 deterministic so the resulting wire bytes are "
"reproducible. A clean-room implementation can verify by: "
"(a) packing a LINKREQUEST from the recorded initiator "
"ephemerals and confirming bytes match `linkrequest_raw_hex`; "
"(b) computing `link_id` per SPEC.md S6.3 (N=2 for HEADER_1) "
"and matching `link_id_hex`; (c) packing an LRPROOF as the "
"responder, with bob's identity Ed25519 sig over `link_id || "
"responder_X25519_pub || responder_long_term_Ed25519_pub || "
"signalling`, and matching `lrproof_raw_hex`; (d) running "
"ECDH+HKDF on either side and matching `derived_key_hex`. "
"Regenerate with `generator_script`."
),
"vectors": [vector],
}
with open(OUT_PATH, "w", encoding="utf-8", newline="\n") as f:
json.dump(payload, f, indent=2, sort_keys=False)
f.write("\n")
print(f"Wrote {OUT_PATH} with 1 vector")
print("ALL PASS")
finally:
try: RNS.Reticulum.exit_handler()
except Exception: pass
if __name__ == "__main__":
main()

250
tools/regen_lxmf.py Normal file
View file

@ -0,0 +1,250 @@
"""
Regenerator for test-vectors/lxmf.json.
Builds two deterministic opportunistic-LXMF wire-byte vectors using
Alice and Bob from `identities.json`:
1. `alice_to_bob_simple` — title b"hello", content b"hi bob",
no fields
2. `alice_to_bob_with_fields` — content + title + a small msgpack-
serialisable fields dict
For each vector we record:
- `lxmf_packed_hex` — the post-`LXMessage.pack()` plaintext body
(`dest(16) || src(16) || sig(64) || msgpack`)
per S5.1 / S5.2.
- `opportunistic_plaintext_hex` — the same body with the leading 16
dest_hash stripped (S5.1 opportunistic form
fed to Token).
- `token_ciphertext_hex` — the Token-encrypted ciphertext that goes
on the wire as the opportunistic packet body
per S3.
Determinism inputs:
- Alice + Bob private keys (from identities.json). Ed25519 sign is
deterministic so the LXMF signature is reproducible.
- `LXMessage.timestamp` pre-set to a fixed value (overrides the
`time.time()` default at `LXMF/LXMessage.py:354`).
- The Token ephemeral X25519 priv (patched X25519PrivateKey.generate
in RNS.Identity).
- The Token CBC IV (patched os.urandom in RNS.Cryptography.Token).
After generation we decrypt with Bob's identity, re-prepend Bob's
dest_hash, parse via `LXMessage.unpack_from_bytes`, and assert
signature_validated + title/content/fields round-trip.
Run from repo root:
python tools/regen_lxmf.py
Updates `test-vectors/lxmf.json` in place. Exit 0 on success.
"""
from __future__ import annotations
import json
import os
import sys
import tempfile
import RNS
import LXMF
from LXMF.LXMessage import LXMessage
from RNS.vendor import umsgpack
REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
OUT_PATH = os.path.join(REPO_ROOT, "test-vectors", "lxmf.json")
IDS_PATH = os.path.join(REPO_ROOT, "test-vectors", "identities.json")
FIXED_LXMF_TIMESTAMP = 1700000000.0
FIXED_EPH_X25519_PRIV = bytes.fromhex("d1d2d3d4d5d6d7d8d9dadbdcdddedfe0e1e2e3e4e5e6e7e8e9eaebecedeeeff0")
FIXED_TOKEN_IV = bytes.fromhex("11223344556677889900aabbccddeeff") # 16 B for AES-CBC
def fail(msg: str) -> None:
print(f"FAIL: {msg}")
sys.exit(1)
def init_minimal_rns():
cfg_dir = tempfile.mkdtemp(prefix="rns-regen-lxmf-")
cfg_path = os.path.join(cfg_dir, "config")
with open(cfg_path, "w", encoding="utf-8") as f:
f.write("[reticulum]\nenable_transport = No\nshare_instance = No\n")
return RNS.Reticulum(configdir=cfg_dir, loglevel=0)
def load_identities():
with open(IDS_PATH, "r", encoding="utf-8") as f:
ids = json.load(f)
alice = next(v for v in ids["vectors"] if v["label"] == "alice")
bob = next(v for v in ids["vectors"] if v["label"] == "bob")
alice_id = RNS.Identity.from_bytes(bytes.fromhex(alice["inputs"]["private_key_hex"]))
bob_id = RNS.Identity.from_bytes(bytes.fromhex(bob ["inputs"]["private_key_hex"]))
if alice_id is None or bob_id is None:
fail("Identity.from_bytes returned None for one of alice/bob")
return alice_id, bob_id
class _StaticX25519Priv:
"""Drop-in replacement for X25519PrivateKey whose .generate()
classmethod returns a private key seeded with FIXED_EPH_X25519_PRIV."""
_real = None
@classmethod
def generate(cls):
# cls._real is the original RNS dispatcher; from_private_bytes
# builds an X25519 priv with the same interface as generate().
return cls._real.from_private_bytes(FIXED_EPH_X25519_PRIV)
def build_vector(alice_dest, bob_dest_in, bob_dest_out, label, title, content, fields):
import sys as _sys
id_mod = _sys.modules["RNS.Identity"]
token_mod = _sys.modules["RNS.Cryptography.Token"]
lxm = LXMessage(
destination = bob_dest_out,
source = alice_dest,
content = content,
title = title,
fields = fields,
desired_method = LXMessage.OPPORTUNISTIC,
)
# Fix the timestamp before pack() so the msgpack payload is stable
lxm.timestamp = FIXED_LXMF_TIMESTAMP
lxm.pack()
packed = lxm.packed
if packed[:16] != bob_dest_out.hash:
fail(f"S5.2 dest_hash slot mismatch ({label})")
if packed[16:32] != alice_dest.hash:
fail(f"S5.2 src_hash slot mismatch ({label})")
opp_plaintext = packed[16:] # opportunistic-form input to Token
# --- Patch ephemeral X25519 generate + Token IV for deterministic CT
real_X25519 = id_mod.X25519PrivateKey
_StaticX25519Priv._real = real_X25519
id_mod.X25519PrivateKey = _StaticX25519Priv
real_urandom = token_mod.os.urandom
def patched_urandom(n):
if n == 16: return FIXED_TOKEN_IV
return real_urandom(n)
token_mod.os.urandom = patched_urandom
try:
ciphertext = bob_dest_out.encrypt(opp_plaintext)
finally:
id_mod.X25519PrivateKey = real_X25519
token_mod.os.urandom = real_urandom
# --- Round-trip: Bob can decrypt and re-parse the plaintext
decrypted = bob_dest_in.decrypt(ciphertext)
if decrypted != opp_plaintext:
fail(f"Token round-trip mismatch ({label})")
full_lxmf = bob_dest_in.hash + decrypted
parsed = LXMessage.unpack_from_bytes(full_lxmf)
if not parsed.signature_validated:
fail(f"S5.5/5.6 unpack_from_bytes did not validate signature ({label})")
if parsed.title_as_string() != title.decode("utf-8"):
fail(f"title round-trip mismatch ({label})")
if parsed.content_as_string() != content.decode("utf-8"):
fail(f"content round-trip mismatch ({label})")
if parsed.fields != fields:
fail(f"fields round-trip mismatch ({label})")
return {
"label": label,
"inputs": {
"src_identity_label": "alice",
"dst_identity_label": "bob",
"title_utf8": title.decode("utf-8"),
"content_utf8": content.decode("utf-8"),
"fields": fields,
"lxmf_timestamp": FIXED_LXMF_TIMESTAMP,
"ephemeral_x25519_priv_hex": FIXED_EPH_X25519_PRIV.hex(),
"token_iv_hex": FIXED_TOKEN_IV.hex(),
},
"expected": {
"lxmf_packed_hex": packed.hex(),
"opportunistic_plaintext_hex": opp_plaintext.hex(),
"token_ciphertext_hex": ciphertext.hex(),
"fields_layout": {
"destination_hash_hex": packed[0:16].hex(),
"source_hash_hex": packed[16:32].hex(),
"signature_hex": packed[32:96].hex(),
"msgpack_payload_hex": packed[96:].hex(),
},
},
"rns_version_at_generation": RNS.__version__,
"lxmf_version_at_generation": LXMF.__version__,
"generator_script": "tools/regen_lxmf.py",
"verifies_spec_sections": ["3", "5.1", "5.2", "5.5", "5.6"],
}
def main():
print(f"regen_lxmf.py against RNS {RNS.__version__} / LXMF {LXMF.__version__}")
init_minimal_rns()
try:
alice_id, bob_id = load_identities()
# Register once; subsequent registrations of same (identity, dir,
# type, aspects) raise. Reuse across vectors.
alice_dest = RNS.Destination(alice_id, RNS.Destination.IN, RNS.Destination.SINGLE,
"lxmf", "delivery")
bob_dest_in = RNS.Destination(bob_id, RNS.Destination.IN, RNS.Destination.SINGLE,
"lxmf", "delivery")
bob_dest_out = RNS.Destination(bob_id, RNS.Destination.OUT, RNS.Destination.SINGLE,
"lxmf", "delivery")
RNS.Identity.remember(b"\x00"*32, alice_dest.hash, alice_id.get_public_key(), None)
RNS.Identity.remember(b"\x00"*32, bob_dest_in.hash, bob_id .get_public_key(), None)
vectors = [
build_vector(alice_dest, bob_dest_in, bob_dest_out,
label = "alice_to_bob_simple",
title = b"hello",
content = b"hi bob",
fields = {}),
build_vector(alice_dest, bob_dest_in, bob_dest_out,
label = "alice_to_bob_with_fields",
title = b"meeting",
content = b"see attached",
fields = {0x01: "k1", 0x02: 42}),
]
payload = {
"_about": (
"Opportunistic LXMF test vectors. "
"`expected.lxmf_packed_hex` is the full plaintext body "
"per SPEC.md S5.2: dest(16) || src(16) || sig(64) || msgpack. "
"`expected.opportunistic_plaintext_hex` is the same with "
"the leading dest_hash stripped (S5.1 wire form). "
"`expected.token_ciphertext_hex` is the deterministic Token "
"encryption of the opportunistic plaintext (S3) using the "
"fixed ephemeral X25519 priv + IV. To verify against "
"upstream: decrypt with Bob's identity, re-prepend Bob's "
"dest_hash, then call LXMessage.unpack_from_bytes — it MUST "
"succeed with signature_validated == True and title/content/"
"fields matching the `inputs` block. Regenerate with "
"`generator_script`."
),
"vectors": vectors,
}
with open(OUT_PATH, "w", encoding="utf-8", newline="\n") as f:
json.dump(payload, f, indent=2, sort_keys=False)
f.write("\n")
print(f"Wrote {OUT_PATH} with {len(vectors)} vectors")
print("ALL PASS")
finally:
try: RNS.Reticulum.exit_handler()
except Exception: pass
if __name__ == "__main__":
main()