mirror of
https://github.com/markqvist/Reticulum.git
synced 2026-10-26 02:48:57 -04:00
Added IFAC tests
This commit is contained in:
Mark Qvist
parent
7347034f9a
commit
929aba0282
1 changed files with 296 additions and 0 deletions
296
tests/ifac.py
Normal file
296
tests/ifac.py
Normal file
|
|
@ -0,0 +1,296 @@
|
|||
import unittest
|
||||
|
||||
import os
|
||||
import time
|
||||
import random
|
||||
|
||||
import RNS
|
||||
from RNS import Transport
|
||||
|
||||
IFAC_TEST_SEED = 0x1FACAD3
|
||||
IFAC_NETNAME = b"rns-ifac-test-net"
|
||||
|
||||
|
||||
def rand_bytes(rng, length):
|
||||
return bytes(rng.randrange(256) for _ in range(length))
|
||||
|
||||
|
||||
class FakeInterface:
|
||||
def __init__(self, ifac_size, ifac_key):
|
||||
self.ifac_size = ifac_size
|
||||
self.ifac_key = ifac_key
|
||||
self.ifac_identity = RNS.Identity.from_bytes(ifac_key)
|
||||
self.violations = []
|
||||
|
||||
def ifac_violation(self, description):
|
||||
self.violations.append(description)
|
||||
|
||||
|
||||
def make_interface(ifac_size):
|
||||
ifac_key = RNS.Cryptography.hkdf(length=64, derive_from=RNS.Identity.full_hash(IFAC_NETNAME),
|
||||
salt=RNS.Reticulum.IFAC_SALT, context=None)
|
||||
|
||||
return FakeInterface(ifac_size, ifac_key)
|
||||
|
||||
|
||||
def make_plain_packet(rng, size, pattern="random"):
|
||||
"""Generate a deterministic protocol-realistic packet: a random or
|
||||
structured payload with the IFAC flag bit of the first header byte
|
||||
clear (as plaintext frames always have)."""
|
||||
if pattern == "random": raw = rand_bytes(rng, size)
|
||||
elif pattern == "zeros": raw = bytes(size)
|
||||
elif pattern == "ones": raw = bytes([0xFF]) * size
|
||||
elif pattern == "seq": raw = bytes((i % 256) for i in range(size))
|
||||
else: raise ValueError(f"Unknown pattern {pattern}")
|
||||
|
||||
raw = bytearray(raw)
|
||||
raw[0] &= 0x7F
|
||||
return bytes(raw)
|
||||
|
||||
|
||||
class TestIFAC(unittest.TestCase):
|
||||
|
||||
def test_01_outgoing_parity(self):
|
||||
print("")
|
||||
|
||||
rng = random.Random(IFAC_TEST_SEED)
|
||||
|
||||
sizes = [8, 16, 32, 100, 500, 1064, 4096, 16384]
|
||||
ifac_sizes = [1, 8, 16]
|
||||
patterns = ["random", "zeros", "ones", "seq"]
|
||||
|
||||
verified = 0
|
||||
for size in sizes:
|
||||
for ifac_size in ifac_sizes:
|
||||
iface = make_interface(ifac_size)
|
||||
for pattern in patterns:
|
||||
raw = make_plain_packet(rng, size, pattern)
|
||||
|
||||
out_new = Transport.handle_outgoing_ifac(iface, raw)
|
||||
out_leg = Transport.handle_outgoing_ifac_legacy(iface, raw)
|
||||
|
||||
self.assertEqual(out_new, out_leg)
|
||||
self.assertEqual(len(out_new), size + ifac_size)
|
||||
verified += 1
|
||||
|
||||
for ifac_size in (8, 16):
|
||||
iface = make_interface(ifac_size)
|
||||
raw = make_plain_packet(rng, 262144, "random")
|
||||
out_new = Transport.handle_outgoing_ifac(iface, raw)
|
||||
out_leg = Transport.handle_outgoing_ifac_legacy(iface, raw)
|
||||
self.assertEqual(out_new, out_leg)
|
||||
verified += 1
|
||||
|
||||
print(f"IFAC outgoing parity: {verified} deterministic cases verified")
|
||||
|
||||
def test_02_incoming_parity(self):
|
||||
print("")
|
||||
|
||||
rng = random.Random(IFAC_TEST_SEED ^ 0x1FAC)
|
||||
|
||||
sizes = [8, 16, 32, 100, 500, 1064, 4096, 16384]
|
||||
ifac_sizes = [1, 8, 16]
|
||||
patterns = ["random", "zeros", "ones", "seq"]
|
||||
|
||||
verified = 0
|
||||
for size in sizes:
|
||||
for ifac_size in ifac_sizes:
|
||||
iface = make_interface(ifac_size)
|
||||
for pattern in patterns:
|
||||
raw = make_plain_packet(rng, size, pattern)
|
||||
frame = Transport.handle_outgoing_ifac(iface, raw)
|
||||
|
||||
ok_new, out_new = Transport.handle_ifac(frame, iface)
|
||||
ok_leg, out_leg = Transport.handle_ifac_legacy(frame, iface)
|
||||
|
||||
self.assertTrue(ok_new)
|
||||
self.assertTrue(ok_leg)
|
||||
self.assertEqual(out_new, out_leg)
|
||||
verified += 1
|
||||
|
||||
for ifac_size in (8, 16):
|
||||
iface = make_interface(ifac_size)
|
||||
raw = make_plain_packet(rng, 262144, "random")
|
||||
frame = Transport.handle_outgoing_ifac(iface, raw)
|
||||
ok_new, out_new = Transport.handle_ifac(frame, iface)
|
||||
ok_leg, out_leg = Transport.handle_ifac_legacy(frame, iface)
|
||||
self.assertTrue(ok_new)
|
||||
self.assertTrue(ok_leg)
|
||||
self.assertEqual(out_new, out_leg)
|
||||
verified += 1
|
||||
|
||||
print(f"IFAC incoming parity: {verified} deterministic cases verified")
|
||||
|
||||
def test_03_roundtrip(self):
|
||||
print("")
|
||||
|
||||
rng = random.Random(IFAC_TEST_SEED ^ 0x0FAC)
|
||||
|
||||
# All four TX/RX handler combinations must recover the original packet
|
||||
combos = [
|
||||
(Transport.handle_outgoing_ifac, Transport.handle_ifac),
|
||||
(Transport.handle_outgoing_ifac, Transport.handle_ifac_legacy),
|
||||
(Transport.handle_outgoing_ifac_legacy, Transport.handle_ifac),
|
||||
(Transport.handle_outgoing_ifac_legacy, Transport.handle_ifac_legacy),
|
||||
]
|
||||
|
||||
sizes = [16, 1064, 16384]
|
||||
ifac_sizes = [1, 8, 16]
|
||||
|
||||
verified = 0
|
||||
for tx, rx in combos:
|
||||
for size in sizes:
|
||||
for ifac_size in ifac_sizes:
|
||||
iface = make_interface(ifac_size)
|
||||
raw = make_plain_packet(rng, size, "random")
|
||||
|
||||
frame = tx(iface, raw)
|
||||
ok, out = rx(frame, iface)
|
||||
|
||||
self.assertTrue(ok)
|
||||
self.assertEqual(out, raw)
|
||||
verified += 1
|
||||
|
||||
print(f"IFAC round-trip: {verified} deterministic TX/RX combinations verified\n All four handler pairings recover the packet exactly.")
|
||||
|
||||
def test_04_invariants(self):
|
||||
print("")
|
||||
|
||||
rng = random.Random(IFAC_TEST_SEED ^ 0xCAFE)
|
||||
iface = make_interface(16)
|
||||
|
||||
for size in (16, 1064, 16384):
|
||||
raw = make_plain_packet(rng, size, "random")
|
||||
|
||||
out_new = Transport.handle_outgoing_ifac(iface, raw)
|
||||
out_leg = Transport.handle_outgoing_ifac_legacy(iface, raw)
|
||||
|
||||
# IFAC flag must be set on the masked frame
|
||||
self.assertEqual(out_new[0] & 0x80, 0x80)
|
||||
|
||||
# The IFAC tag region must be transmitted in the clear and
|
||||
# identical in both implementations
|
||||
self.assertEqual(out_new[2:2+iface.ifac_size], out_leg[2:2+iface.ifac_size])
|
||||
self.assertEqual(out_new[2:2+iface.ifac_size], iface.ifac_identity.sign(raw)[-iface.ifac_size:])
|
||||
|
||||
# Second header byte must be masked identically
|
||||
self.assertEqual(out_new[1], out_leg[1])
|
||||
|
||||
# The incoming handler must clear the IFAC flag in its output
|
||||
ok, out = Transport.handle_ifac(out_new, iface)
|
||||
self.assertTrue(ok)
|
||||
self.assertEqual(out[0] & 0x80, 0)
|
||||
|
||||
print("IFAC invariants: flag set on frame, tag in clear and identical, flag cleared on input recovery")
|
||||
|
||||
def test_05_invalid_tags(self):
|
||||
print("")
|
||||
|
||||
rng = random.Random(IFAC_TEST_SEED ^ 0xBAD)
|
||||
|
||||
sizes = [100, 1064, 16384]
|
||||
ifac_sizes = [1, 8, 16]
|
||||
|
||||
for size in sizes:
|
||||
for ifac_size in ifac_sizes:
|
||||
iface = make_interface(ifac_size)
|
||||
raw = make_plain_packet(rng, size, "random")
|
||||
frame = bytearray(Transport.handle_outgoing_ifac(iface, raw))
|
||||
|
||||
corruptions = []
|
||||
# Corrupt a masked payload byte
|
||||
corrupted = bytearray(frame)
|
||||
corrupted[-1] ^= 0x01
|
||||
corruptions.append(bytes(corrupted))
|
||||
|
||||
# Corrupt the second header byte
|
||||
corrupted = bytearray(frame)
|
||||
corrupted[1] ^= 0x40
|
||||
corruptions.append(bytes(corrupted))
|
||||
|
||||
# Corrupt a byte in the IFAC tag region
|
||||
if ifac_size > 0:
|
||||
corrupted = bytearray(frame)
|
||||
corrupted[2 + ifac_size//2] ^= 0x01
|
||||
corruptions.append(bytes(corrupted))
|
||||
|
||||
for corrupted in corruptions:
|
||||
ok_new, out_new = Transport.handle_ifac(corrupted, iface)
|
||||
ok_leg, out_leg = Transport.handle_ifac_legacy(corrupted, iface)
|
||||
|
||||
# Both implementations must behave identically on any
|
||||
# corrupted frame, including the violation report
|
||||
self.assertEqual(ok_new, ok_leg)
|
||||
self.assertEqual(out_new, out_leg)
|
||||
self.assertEqual(iface.violations[-1:], ["Invalid IFAC on packet"])
|
||||
|
||||
# With >= 8 bytes of tag entropy, rejection is
|
||||
# deterministic. At ifac_size = 1 the truncated tag has
|
||||
# only 8 bits, so a corrupted frame can statistically
|
||||
# collide with the original truncation (1/256) - both
|
||||
# implementations behave identically either way.
|
||||
if ifac_size >= 8: self.assertFalse(ok_new)
|
||||
|
||||
print("IFAC invalid tags: behaviour and violation reporting identical "
|
||||
"between implementations; deterministic rejection verified for "
|
||||
"8+ byte tags")
|
||||
|
||||
@unittest.skipIf(os.getenv("RNS_SKIP_PERF_TESTS") != None, "Skipping performance tests")
|
||||
def test_06_benchmark(self):
|
||||
print("")
|
||||
|
||||
rng = random.Random(IFAC_TEST_SEED ^ 0xBEEF)
|
||||
|
||||
sizes = [1500, 16384, 262144]
|
||||
runs = {1500: 30, 16384: 15, 262144: 5}
|
||||
|
||||
print("IFAC handler benchmark (optimized vs legacy):")
|
||||
print(f"{'size':>8} {'out lgc':>11} {'out opt':>11} {'out spd':>8} | "
|
||||
f"{'in lgc':>11} {'in new':>11} {'in spd':>8}")
|
||||
print("-" * 84)
|
||||
|
||||
total_leg = total_new = 0.0
|
||||
for size in sizes:
|
||||
iface = make_interface(16)
|
||||
raw = make_plain_packet(rng, size, "random")
|
||||
frame = Transport.handle_outgoing_ifac(iface, raw)
|
||||
n = runs[size]
|
||||
|
||||
# Warmup
|
||||
Transport.handle_outgoing_ifac_legacy(iface, raw)
|
||||
Transport.handle_outgoing_ifac(iface, raw)
|
||||
Transport.handle_ifac_legacy(frame, iface)
|
||||
Transport.handle_ifac(frame, iface)
|
||||
|
||||
t0 = time.perf_counter()
|
||||
for _ in range(n): Transport.handle_outgoing_ifac_legacy(iface, raw)
|
||||
out_leg = (time.perf_counter() - t0) / n
|
||||
|
||||
t0 = time.perf_counter()
|
||||
for _ in range(n): Transport.handle_outgoing_ifac(iface, raw)
|
||||
out_new = (time.perf_counter() - t0) / n
|
||||
|
||||
t0 = time.perf_counter()
|
||||
for _ in range(n): Transport.handle_ifac_legacy(frame, iface)
|
||||
in_leg = (time.perf_counter() - t0) / n
|
||||
|
||||
t0 = time.perf_counter()
|
||||
for _ in range(n): Transport.handle_ifac(frame, iface)
|
||||
in_new = (time.perf_counter() - t0) / n
|
||||
|
||||
total_leg += out_leg + in_leg
|
||||
total_new += out_new + in_new
|
||||
|
||||
def ms(t): return f"{t*1000:.3f}m"
|
||||
|
||||
print(f"{size:>8} {ms(out_leg):>11} {ms(out_new):>11} {out_leg/out_new:7.1f}x | "
|
||||
f"{ms(in_leg):>11} {ms(in_new):>11} {in_leg/in_new:7.1f}x")
|
||||
|
||||
print(f"\nCombined TX+RX at {sizes[-1]} bytes: legacy "
|
||||
f"{(out_leg+in_leg)*1000:.2f} ms/pkt, optimized {(out_new+in_new)*1000:.2f} ms/pkt, "
|
||||
f"{(out_leg+in_leg)/(out_new+in_new):.1f}x speedup")
|
||||
print(f"Overall speedup across benchmarks: {total_leg/total_new:.1f}x")
|
||||
print()
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
Loading…
Add table
Add a link
Reference in a new issue