reticiulum-specification/tools/regen_announces.py

"""
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()