reticiulum-specification/tools/regen_propagation_peer.py

"""
Regenerator for test-vectors/propagation-peer.json.

Builds deterministic propagation-node announce app_data, directional peering
key and /offer payloads, and the stamped peer-sync Resource plaintext.
"""

from __future__ import annotations

import json
import os
import sys

import LXMF
import RNS
from LXMF import LXStamper
from LXMF.LXMRouter import LXMRouter
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", "propagation-peer.json")
IDS_PATH = os.path.join(REPO_ROOT, "test-vectors", "identities.json")
PROPAGATED_PATH = os.path.join(REPO_ROOT, "test-vectors", "propagated-lxmf.json")

FIXED_ANNOUNCE_TIME = 1700000120
FIXED_SYNC_TIME = 1700000180.0
PEERING_COST = 4


def load_json(path: str):
    with open(path, "r", encoding="utf-8") as input_file:
        return json.load(input_file)


def load_identities():
    vectors = load_json(IDS_PATH)["vectors"]
    alice = next(vector for vector in vectors if vector["label"] == "alice")
    bob = next(vector for vector in vectors if vector["label"] == "bob")
    return (
        RNS.Identity.from_bytes(bytes.fromhex(alice["inputs"]["private_key_hex"])),
        RNS.Identity.from_bytes(bytes.fromhex(bob["inputs"]["private_key_hex"])),
    )


def build_announce() -> bytes:
    class FakeRouter:
        name = "Vector PN"
        propagation_node = True
        from_static_only = False
        propagation_stamp_cost = 16
        propagation_stamp_cost_flexibility = 3
        peering_cost = 18
        propagation_per_transfer_limit = 256
        propagation_per_sync_limit = 10240

        def get_propagation_node_announce_metadata(self):
            return LXMRouter.get_propagation_node_announce_metadata(self)

    router_module = sys.modules["LXMF.LXMRouter"]
    real_time = router_module.time.time
    router_module.time.time = lambda: FIXED_ANNOUNCE_TIME
    try:
        return LXMRouter.get_propagation_node_app_data(FakeRouter())
    finally:
        router_module.time.time = real_time


def find_peering_key(peering_id: bytes) -> tuple[bytes, int]:
    workblock = LXStamper.stamp_workblock(
        peering_id, expand_rounds=LXStamper.WORKBLOCK_EXPAND_ROUNDS_PEERING
    )
    for candidate_number in range(1_000_000):
        candidate = candidate_number.to_bytes(32, "big")
        value = LXStamper.stamp_value(workblock, candidate)
        if value >= PEERING_COST:
            return candidate, value
    raise RuntimeError("could not find deterministic peering key")


def main() -> None:
    print(f"regen_propagation_peer.py against RNS {RNS.__version__} / LXMF {LXMF.__version__}")
    alice_id, bob_id = load_identities()
    propagated = load_json(PROPAGATED_PATH)
    first = propagated["boundary"]["largest_packet"]
    second = propagated["boundary"]["first_resource"]
    transient_ids = [
        bytes.fromhex(first["transient_id_hex"]),
        bytes.fromhex(second["transient_id_hex"]),
    ]
    submitted_entries = [
        bytes.fromhex(first["submitted_entry_hex"]),
        bytes.fromhex(second["submitted_entry_hex"]),
    ]

    # Receiver validates receiver_identity_hash || offering_identity_hash.
    peering_id = bob_id.hash + alice_id.hash
    peering_key, peering_value = find_peering_key(peering_id)
    announce_data = build_announce()
    offer_data = [peering_key, transient_ids]
    sync_data = [FIXED_SYNC_TIME, submitted_entries]

    payload = {
        "_about": (
            "Deterministic propagation-node announce, directional peering-key, "
            "/offer, and peer-sync Resource plaintext vectors."
        ),
        "inputs": {
            "offering_identity_label": "alice",
            "receiving_identity_label": "bob",
            "announce_time": FIXED_ANNOUNCE_TIME,
            "sync_time": FIXED_SYNC_TIME,
            "peering_cost": PEERING_COST,
        },
        "announce": {
            "app_data_hex": announce_data.hex(),
            "decoded": {
                "legacy_support": False,
                "timebase": FIXED_ANNOUNCE_TIME,
                "propagation_enabled": True,
                "transfer_limit_kb": 256,
                "sync_limit_kb": 10240,
                "stamp_costs": [16, 3, 18],
                "name_utf8": "Vector PN",
            },
        },
        "peering": {
            "peering_id_hex": peering_id.hex(),
            "peering_key_hex": peering_key.hex(),
            "peering_key_value": peering_value,
            "offer_data_hex": umsgpack.packb(offer_data).hex(),
            "offered_transient_ids_hex": [transient_id.hex() for transient_id in transient_ids],
            "expected_responses": {
                "receiver_has_none": True,
                "receiver_has_all": False,
                "receiver_has_first_hex": [transient_ids[1].hex()],
            },
            "sync_resource_plaintext_hex": umsgpack.packb(sync_data).hex(),
            "submitted_entries_hex": [entry.hex() for entry in submitted_entries],
        },
        "rns_version_at_generation": RNS.__version__,
        "lxmf_version_at_generation": LXMF.__version__,
        "generator_script": "tools/regen_propagation_peer.py",
        "verifies_spec_sections": ["5.8.2", "5.8.4", "5.8.5"],
    }
    with open(OUT_PATH, "w", encoding="utf-8", newline="\n") as output:
        json.dump(payload, output, indent=2, sort_keys=False)
        output.write("\n")
    print(f"Wrote {OUT_PATH}")
    print("ALL PASS")


if __name__ == "__main__":
    main()