""" Verifier for DIRECT LXMF delivery over an established Reticulum Link. Validates deterministic PACKET and Resource vectors, the exact upstream content_size boundary (319/320), full-body DIRECT parsing, wrong-link-key rejection, and LXMRouter's DIRECT receive dispatch. """ from __future__ import annotations import json import os import sys import tempfile import LXMF import RNS from LXMF.LXMessage import LXMessage from LXMF.LXMRouter import LXMRouter from RNS.Cryptography.Token import Token from RNS.Resource import Resource, ResourceAdvertisement REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) VECTORS_PATH = os.path.join(REPO_ROOT, "test-vectors", "link-lxmf.json") IDS_PATH = os.path.join(REPO_ROOT, "test-vectors", "identities.json") LINKS_PATH = os.path.join(REPO_ROOT, "test-vectors", "links.json") def fail(message: str) -> None: print(f"FAIL: {message}") sys.exit(1) def init_minimal_rns(): config_dir = tempfile.mkdtemp(prefix="rns-verify-link-lxmf-") config_path = os.path.join(config_dir, "config") with open(config_path, "w", encoding="utf-8") as config: config.write("[reticulum]\nenable_transport = No\nshare_instance = No\n") return RNS.Reticulum(configdir=config_dir, loglevel=0) def load_json(path: str): with open(path, "r", encoding="utf-8") as input_file: return json.load(input_file) def content_size(packed: bytes) -> int: payload = packed[2 * LXMessage.DESTINATION_LENGTH + LXMessage.SIGNATURE_LENGTH :] return len(payload) - LXMessage.TIMESTAMP_SIZE - LXMessage.STRUCT_OVERHEAD def parse_and_check(packed: bytes, expected_size: int) -> LXMessage: parsed = LXMessage.unpack_from_bytes(packed) if not parsed.signature_validated: fail(f"DIRECT LXMF signature invalid at content_size {expected_size}") if len(parsed.content) != expected_size or parsed.content != b"x" * expected_size: fail(f"DIRECT LXMF content mismatch at content_size {expected_size}") if parsed.method != LXMessage.UNKNOWN: fail("unpack_from_bytes unexpectedly assigned a delivery method") return parsed def verify_vectors(derived_key: bytes, link_id: bytes) -> tuple[bytes, bytes]: vectors = load_json(VECTORS_PATH)["vectors"] packet_vector = next(vector for vector in vectors if vector["expected"]["representation"] == LXMessage.PACKET) resource_vector = next(vector for vector in vectors if vector["expected"]["representation"] == LXMessage.RESOURCE) token = Token(derived_key) packet_packed = bytes.fromhex(packet_vector["expected"]["lxmf_packed_hex"]) packet_raw = bytes.fromhex(packet_vector["expected"]["link_packet_raw_hex"]) packet_ciphertext = bytes.fromhex(packet_vector["expected"]["link_packet_ciphertext_hex"]) if content_size(packet_packed) != 319 or len(packet_packed) != RNS.Link.MDU: fail("319 boundary vector does not fill the Link MDU exactly") if len(packet_raw) != 19 + len(packet_ciphertext) or len(packet_raw) > RNS.Reticulum.MTU: fail("319 boundary vector has an invalid Link packet wire length") if packet_raw[:2] != bytes([RNS.Destination.LINK << 2, 0]): fail("DIRECT PACKET vector flags/hops are not HEADER_1 DATA LINK") if packet_raw[2:18] != link_id or packet_raw[18] != RNS.Packet.NONE: fail("DIRECT PACKET vector link_id/context mismatch") if packet_raw[19:] != packet_ciphertext: fail("DIRECT PACKET raw body differs from recorded ciphertext") if token.decrypt(packet_ciphertext) != packet_packed: fail("DIRECT PACKET did not decrypt to the complete LXMF body") parse_and_check(packet_packed, 319) resource_packed = bytes.fromhex(resource_vector["expected"]["lxmf_packed_hex"]) parts = [bytes.fromhex(part["body_hex"]) for part in resource_vector["expected"]["resource_parts"]] stream = b"".join(parts) decrypted = token.decrypt(stream) prefix = bytes.fromhex(resource_vector["inputs"]["throwaway_prefix_hex"]) if decrypted[: Resource.RANDOM_HASH_SIZE] != prefix: fail("DIRECT Resource throwaway prefix mismatch") if decrypted[Resource.RANDOM_HASH_SIZE :] != resource_packed: fail("DIRECT Resource did not decrypt to the complete LXMF body") if content_size(resource_packed) != 320: fail("Resource boundary vector computed content_size is not 320") parse_and_check(resource_packed, 320) adv = ResourceAdvertisement.unpack( bytes.fromhex(resource_vector["expected"]["resource_advertisement_plaintext_hex"]) ) if adv.d != len(resource_packed) or adv.h.hex() != resource_vector["expected"]["resource_hash_hex"]: fail("DIRECT Resource advertisement size/hash mismatch") try: Token(bytes(reversed(derived_key))).decrypt(packet_ciphertext) except Exception: pass else: fail("DIRECT packet decrypted under the wrong Link key") print("PASS S5.2/S10.1 DIRECT PACKET/RESOURCE boundary vectors: 319 -> PACKET, 320 -> RESOURCE") print("PASS S3/S5.5/S5.6 DIRECT Link decrypt and signed full-body parse; wrong Link key rejected") return packet_packed, resource_packed def verify_upstream_boundary(alice_identity, bob_identity) -> None: alice = RNS.Destination( alice_identity, RNS.Destination.IN, RNS.Destination.SINGLE, "verify_link_lxmf", "alice" ) bob = RNS.Destination( bob_identity, RNS.Destination.OUT, RNS.Destination.SINGLE, "verify_link_lxmf", "bob" ) for size, expected in [(319, LXMessage.PACKET), (320, LXMessage.RESOURCE)]: message = LXMessage(bob, alice, b"x" * size, b"", {}, LXMessage.DIRECT) message.timestamp = 1700000000.0 message.pack() if content_size(message.packed) != size or message.representation != expected: fail(f"upstream boundary mismatch at {size}") print("PASS S10.1 upstream selection uses computed LXMF content_size at the 319/320 boundary") def verify_router_direct_dispatch(packet_packed: bytes) -> None: router = object.__new__(LXMRouter) captured = {} router.lxmf_delivery = lambda data, destination_type, **kwargs: captured.update( data=data, destination_type=destination_type, method=kwargs.get("method") ) class ImmediateThread: def __init__(self, target, daemon=None): self.target = target def start(self): self.target() class FakePacket: destination_type = RNS.Destination.LINK rssi = -80 snr = 7 q = 90 ratchet_id = bytes.fromhex("00112233445566778899aabbccddeeff") proved = False def prove(self): self.proved = True packet = FakePacket() router_mod = sys.modules["LXMF.LXMRouter"] real_thread = router_mod.threading.Thread router_mod.threading.Thread = ImmediateThread try: router.delivery_packet(packet_packed, packet) finally: router_mod.threading.Thread = real_thread if not packet.proved: fail("LXMRouter.delivery_packet did not prove DIRECT Link DATA") if captured.get("data") != packet_packed: fail("LXMRouter.delivery_packet altered or prepended DIRECT LXMF data") if captured.get("destination_type") != RNS.Destination.LINK: fail("LXMRouter.delivery_packet lost LINK destination type") if captured.get("method") != LXMessage.DIRECT: fail("LXMRouter.delivery_packet did not classify LINK data as DIRECT") print("PASS S5.2 DIRECT receive dispatch proves DATA and passes full LXMF body unchanged") def main() -> None: print(f"verify_link_lxmf.py against RNS {RNS.__version__} / LXMF {LXMF.__version__}") init_minimal_rns() try: identities = load_json(IDS_PATH)["vectors"] links = load_json(LINKS_PATH)["vectors"][0] alice = next(vector for vector in identities if vector["label"] == "alice") bob = next(vector for vector in identities if vector["label"] == "bob") alice_identity = RNS.Identity.from_bytes(bytes.fromhex(alice["inputs"]["private_key_hex"])) bob_identity = RNS.Identity.from_bytes(bytes.fromhex(bob["inputs"]["private_key_hex"])) RNS.Identity.remember(bytes(32), bytes.fromhex(alice["expected"]["destination_hash_hex"]), alice_identity.get_public_key(), None) derived_key = bytes.fromhex(links["expected"]["derived_key_hex"]) link_id = bytes.fromhex(links["expected"]["link_id_hex"]) packet_packed, _ = verify_vectors(derived_key, link_id) verify_upstream_boundary(alice_identity, bob_identity) verify_router_direct_dispatch(packet_packed) print("ALL PASS") finally: try: RNS.Reticulum.exit_handler() except Exception: pass if __name__ == "__main__": main()