mirror of
https://github.com/markqvist/Reticulum.git
synced 2025-07-22 14:31:00 -04:00
Updated tests for link modes
This commit is contained in:
Mark Qvist
parent
bc7a8cd09f
commit
3cbcbec942
2 changed files with 140 additions and 29 deletions
|
|
@ -7,8 +7,11 @@ import os
|
|||
signed_message = "e51a008b8b8ba855993d8892a40daad84a6fb69a7138e1b5f69b427fe03449826ab6ccb81f0d72b4725e8d55c814d3e8e151b495cf5b59702f197ec366d935ad04a98ca519d6964f96ea09910b020351d1cdff3befbad323a2a28a6ec7ced4d0d67f02c525f93b321d9b076d704408475bd2d123cd51916f7e49039246ac56add37ef87e32d7f9853ac44a7f77d26fedc83e4e67a45742b751c2599309f5eda6efa0dafd957f61af1f0e86c4d6c5052e0e5fa577db99846f2b7a0204c31cef4013ca51cb307506c9209fd18d0195a7c9ae628af1a1d9ee7a4cf30037ed190a9fdcaa4ce5bb7bea19803cb5b5cea8c21fdb98d8f73ff5aaad87f5f6c3b7bcfe8974e5b063cc1113d77b9e96bec1c9d10ed37b780c3f7349a34092bb3968daeced40eb0b5130c0d11595e30b9671896385d04289d067f671599386536eed8430a72e186fb95023d5ac5dd442443bfabfe13a84a38d060af73bf20f921f38a768672fdbcb1dfece7458166e2e15948d6b4fa81f42db48747d283c670f576a0b410b31a70d2594823d0e29135a488cb0408c9e5bc1e197ff99aef471924231ccc8e3eddc82dbcea4801f14c5fc7a389a26a52cc93cfe0770953ef595ff410b7033a6ed5c975dd922b3f48f9dffcfb412eeed5758f3aa51de7eb47cd2cb"
|
||||
sig_from_key_0 = "3020ef58f861591826a61c3d2d4a25b949cdb3094085ba6b1177a6f2a05f3cdd24d1095d6fdd078f0b2826e80b261c93c1ff97fbfd4857f25706d57dd073590c"
|
||||
|
||||
encrypted_message_legacy = "71884a271ead43558fcf1e331c5aebcd43498f16da16f8056b0893ce6b15d521eaa4f31639cd34da1b57995944076c4f14f300f2d2612111d21a3429a9966ac1da68545c00c7887d8b26f6c1ab9defa020b9519849ca41b7904199882802b6542771df85144a79890289d3c02daef6c26652c5ce9de231a2"
|
||||
fixed_token_legacy = "54d6ba347f3f2fe74fa52d6844a9090c049a6f437d7d151b9bd7db3e6785dd40286c451babda82660cbb4827517365b740675adf60d4b82778d7f7815a0e9818f2f2d3f15c0365e9d4f08df4f8261e5549c8c398e92bc66750fcd4ce7ea150f8a8761936341129e89afd22eaa57c303ccbe045d0b2fc7b8637946e16627419ef1fea0a0fef974c418a98af046d61e8e064f42c4948b0c81701106583c8f224329c0b475cb2168dc2e3fbf649edb79c58b7c839a509e146ec8d26589cb990c76c756fdefd0110410a6ab84fa3a722db74"
|
||||
|
||||
encrypted_message = "71884a271ead43558fcf1e331c5aebcd43498f16da16f8056b0893ce6b15d521eaa4f31639cd34da1b57995944076c4f14f300f2d2612111d21a3429a9966ac1da68545c00c7887d8b26f6c1ab9defa020b9519849ca41b7904199882802b6542771df85144a79890289d3c02daef6c26652c5ce9de231a2"
|
||||
fixed_token = "54d6ba347f3f2fe74fa52d6844a9090c049a6f437d7d151b9bd7db3e6785dd40286c451babda82660cbb4827517365b740675adf60d4b82778d7f7815a0e9818f2f2d3f15c0365e9d4f08df4f8261e5549c8c398e92bc66750fcd4ce7ea150f8a8761936341129e89afd22eaa57c303ccbe045d0b2fc7b8637946e16627419ef1fea0a0fef974c418a98af046d61e8e064f42c4948b0c81701106583c8f224329c0b475cb2168dc2e3fbf649edb79c58b7c839a509e146ec8d26589cb990c76c756fdefd0110410a6ab84fa3a722db74"
|
||||
fixed_token = "e028a7d7b757241e53c65f20214365a74b83b4529e83291b391c614703d8a218708d1b47666c277c41cd9bb64b36ba5688801945df0b2470ea05e215eb9ffac8fad658f4d7fbc28688712daf9066e662b8f941cc766ee394e57b12eb629e3e807b3af9ec867170b5cd40fc17f29ff478fa28e419772020bd7c141f732851fb2397b150f30c2f9e347dabe01e80e5d857aaee7087d557f6027922fa814a67f417fba5e600975cf7502686d4112ba7814b9661067a7c900d5ed62f3125b126dc3c7ffea21d4702e6d205614b5da09e4b4d"
|
||||
|
||||
fixed_keys = [
|
||||
("f8953ffaf607627e615603ff1530c82c434cf87c07179dd7689ea776f30b964cfb7ba6164af00c5111a45e69e57d885e1285f8dbfe3a21e95ae17cf676b0f8b7", "650b5d76b6bec0390d1f8cfca5bd33f9"),
|
||||
|
|
@ -148,9 +151,15 @@ class TestIdentity(unittest.TestCase):
|
|||
def test_2_encrypt(self):
|
||||
print("")
|
||||
|
||||
# Test decryption of known token
|
||||
# Test loading identity from bytes
|
||||
fid = RNS.Identity.from_bytes(bytes.fromhex(fixed_keys[0][0]))
|
||||
self.assertEqual(fid.hash, bytes.fromhex(fixed_keys[0][1]))
|
||||
|
||||
# Test decryption of known AES-128 token
|
||||
plaintext = fid.decrypt(bytes.fromhex(fixed_token_legacy))
|
||||
self.assertEqual(plaintext, bytes.fromhex(encrypted_message_legacy))
|
||||
|
||||
# Test decryption of known AES-256 token
|
||||
plaintext = fid.decrypt(bytes.fromhex(fixed_token))
|
||||
self.assertEqual(plaintext, bytes.fromhex(encrypted_message))
|
||||
|
||||
|
|
|
|||
156
tests/link.py
156
tests/link.py
|
|
@ -25,6 +25,7 @@ fixed_keys = [
|
|||
]
|
||||
|
||||
BUFFER_TEST_TARGET = 32000
|
||||
LINK_UP_WAIT = 0.050
|
||||
|
||||
def targets_job(caller):
|
||||
cmd = "python -c \"from tests.link import targets; targets()\""
|
||||
|
|
@ -65,10 +66,11 @@ class TestLink(unittest.TestCase):
|
|||
close_rns()
|
||||
|
||||
@skipIf(os.getenv('SKIP_NORMAL_TESTS') != None, "Skipping")
|
||||
def test_0_valid_announce(self):
|
||||
def test_00_valid_announce(self):
|
||||
init_rns(self)
|
||||
print("")
|
||||
|
||||
print("Testing valid announce...")
|
||||
fid = RNS.Identity.from_bytes(bytes.fromhex(fixed_keys[3][0]))
|
||||
dst = RNS.Destination(fid, RNS.Destination.IN, RNS.Destination.SINGLE, "test", "announce")
|
||||
ap = dst.announce(send=False)
|
||||
|
|
@ -76,10 +78,11 @@ class TestLink(unittest.TestCase):
|
|||
self.assertEqual(RNS.Identity.validate_announce(ap), True)
|
||||
|
||||
@skipIf(os.getenv('SKIP_NORMAL_TESTS') != None, "Skipping")
|
||||
def test_1_invalid_announce(self):
|
||||
def test_01_invalid_announce(self):
|
||||
init_rns(self)
|
||||
print("")
|
||||
|
||||
print("Testing invalid announce...")
|
||||
fid = RNS.Identity.from_bytes(bytes.fromhex(fixed_keys[4][0]))
|
||||
dst = RNS.Destination(fid, RNS.Destination.IN, RNS.Destination.SINGLE, "test", "announce")
|
||||
ap = dst.announce(send=False)
|
||||
|
|
@ -92,7 +95,7 @@ class TestLink(unittest.TestCase):
|
|||
self.assertEqual(RNS.Identity.validate_announce(ap), False)
|
||||
|
||||
@skipIf(os.getenv('SKIP_NORMAL_TESTS') != None, "Skipping")
|
||||
def test_2_establish(self):
|
||||
def test_02_establish(self):
|
||||
init_rns(self)
|
||||
print("")
|
||||
|
||||
|
|
@ -106,18 +109,42 @@ class TestLink(unittest.TestCase):
|
|||
|
||||
self.assertEqual(dest.hash, bytes.fromhex("fb48da0e82e6e01ba0c014513f74540d"))
|
||||
|
||||
print("Testing default mode link establishment...")
|
||||
l1 = RNS.Link(dest)
|
||||
time.sleep(0.5)
|
||||
time.sleep(LINK_UP_WAIT)
|
||||
self.assertEqual(l1.status, RNS.Link.ACTIVE)
|
||||
|
||||
l1.teardown()
|
||||
time.sleep(0.5)
|
||||
time.sleep(LINK_UP_WAIT)
|
||||
self.assertEqual(l1.status, RNS.Link.CLOSED)
|
||||
|
||||
print("Testing AES_128_CBC mode link establishment...")
|
||||
l2 = RNS.Link(dest, mode=RNS.Link.MODE_AES128_CBC)
|
||||
time.sleep(LINK_UP_WAIT)
|
||||
self.assertEqual(l2.status, RNS.Link.ACTIVE)
|
||||
self.assertEqual(l2.mode, RNS.Link.MODE_AES128_CBC)
|
||||
self.assertEqual(len(l2.derived_key), 32)
|
||||
|
||||
l2.teardown()
|
||||
time.sleep(LINK_UP_WAIT)
|
||||
self.assertEqual(l2.status, RNS.Link.CLOSED)
|
||||
|
||||
print("Testing AES_256_CBC mode link establishment...")
|
||||
l3 = RNS.Link(dest, mode=RNS.Link.MODE_AES256_CBC)
|
||||
time.sleep(LINK_UP_WAIT)
|
||||
self.assertEqual(l3.status, RNS.Link.ACTIVE)
|
||||
self.assertEqual(l3.mode, RNS.Link.MODE_AES256_CBC)
|
||||
self.assertEqual(len(l3.derived_key), 64)
|
||||
|
||||
l3.teardown()
|
||||
time.sleep(LINK_UP_WAIT)
|
||||
self.assertEqual(l3.status, RNS.Link.CLOSED)
|
||||
|
||||
@skipIf(os.getenv('SKIP_NORMAL_TESTS') != None, "Skipping")
|
||||
def test_3_packets(self):
|
||||
def test_03a_packets(self):
|
||||
init_rns(self)
|
||||
print("")
|
||||
print("Testing default mode link packets...")
|
||||
|
||||
# TODO: Load this from public bytes only
|
||||
id1 = RNS.Identity.from_bytes(bytes.fromhex(fixed_keys[0][0]))
|
||||
|
|
@ -131,7 +158,7 @@ class TestLink(unittest.TestCase):
|
|||
self.assertEqual(dest.hash, bytes.fromhex("fb48da0e82e6e01ba0c014513f74540d"))
|
||||
|
||||
l1 = RNS.Link(dest)
|
||||
time.sleep(0.5)
|
||||
time.sleep(LINK_UP_WAIT)
|
||||
self.assertEqual(l1.status, RNS.Link.ACTIVE)
|
||||
|
||||
b = 0
|
||||
|
|
@ -165,6 +192,7 @@ class TestLink(unittest.TestCase):
|
|||
if not r.status == RNS.PacketReceipt.DELIVERED:
|
||||
all_ok = False
|
||||
break
|
||||
time.sleep(0.01)
|
||||
|
||||
pduration = time.time()-pstart
|
||||
|
||||
|
|
@ -182,11 +210,85 @@ class TestLink(unittest.TestCase):
|
|||
print("Single packet and proof round-trip throughput is "+self.size_str(b/pduration, "b")+"ps")
|
||||
|
||||
l1.teardown()
|
||||
time.sleep(0.5)
|
||||
time.sleep(LINK_UP_WAIT)
|
||||
self.assertEqual(l1.status, RNS.Link.CLOSED)
|
||||
|
||||
@skipIf(os.getenv('SKIP_NORMAL_TESTS') != None, "Skipping")
|
||||
def test_4_micro_resource(self):
|
||||
def test_03b_packets(self):
|
||||
init_rns(self)
|
||||
print("")
|
||||
|
||||
print("Testing AES_256_CBC mode link packets...")
|
||||
|
||||
# TODO: Load this from public bytes only
|
||||
id1 = RNS.Identity.from_bytes(bytes.fromhex(fixed_keys[0][0]))
|
||||
self.assertEqual(id1.hash, bytes.fromhex(fixed_keys[0][1]))
|
||||
|
||||
RNS.Transport.request_path(bytes.fromhex("fb48da0e82e6e01ba0c014513f74540d"))
|
||||
time.sleep(0.2)
|
||||
|
||||
dest = RNS.Destination(id1, RNS.Destination.OUT, RNS.Destination.SINGLE, APP_NAME, "link", "establish")
|
||||
|
||||
self.assertEqual(dest.hash, bytes.fromhex("fb48da0e82e6e01ba0c014513f74540d"))
|
||||
|
||||
l1 = RNS.Link(dest, mode=RNS.Link.MODE_AES256_CBC)
|
||||
time.sleep(LINK_UP_WAIT)
|
||||
self.assertEqual(l1.status, RNS.Link.ACTIVE)
|
||||
|
||||
b = 0
|
||||
pr_t = 0
|
||||
receipts = []
|
||||
if RNS.Cryptography.backend() == "internal" or RNS.Reticulum.MTU > 500:
|
||||
num_packets = 50
|
||||
else:
|
||||
num_packets = 500
|
||||
|
||||
packet_size = RNS.Link.MDU
|
||||
pstart = time.time()
|
||||
print("Sending "+str(num_packets)+" link packets of "+str(packet_size)+" bytes...")
|
||||
for i in range(0, num_packets):
|
||||
time.sleep(0.003)
|
||||
b += packet_size
|
||||
data = os.urandom(packet_size)
|
||||
start = time.time()
|
||||
p = RNS.Packet(l1, data)
|
||||
receipts.append(p.send())
|
||||
pr_t += time.time() - start
|
||||
|
||||
print("Sent "+self.size_str(b)+", "+self.size_str(b/pr_t, "b")+"ps")
|
||||
print("Checking receipts...", end=" ")
|
||||
|
||||
all_ok = False
|
||||
receipt_timeout = time.time() + 35
|
||||
while not all_ok and time.time() < receipt_timeout:
|
||||
for r in receipts:
|
||||
all_ok = True
|
||||
if not r.status == RNS.PacketReceipt.DELIVERED:
|
||||
all_ok = False
|
||||
break
|
||||
time.sleep(0.01)
|
||||
|
||||
pduration = time.time()-pstart
|
||||
|
||||
n_failed = 0
|
||||
for r in receipts:
|
||||
if not r.status == RNS.PacketReceipt.DELIVERED:
|
||||
n_failed += 1
|
||||
|
||||
if n_failed > 0:
|
||||
ns = "s" if n_failed != 1 else ""
|
||||
print("Failed to receive proof for "+str(n_failed)+" packet"+ns)
|
||||
|
||||
self.assertEqual(all_ok, True)
|
||||
print("OK!")
|
||||
print("Single packet and proof round-trip throughput is "+self.size_str(b/pduration, "b")+"ps")
|
||||
|
||||
l1.teardown()
|
||||
time.sleep(LINK_UP_WAIT)
|
||||
self.assertEqual(l1.status, RNS.Link.CLOSED)
|
||||
|
||||
@skipIf(os.getenv('SKIP_NORMAL_TESTS') != None, "Skipping")
|
||||
def test_04_micro_resource(self):
|
||||
init_rns(self)
|
||||
print("")
|
||||
print("Micro resource test")
|
||||
|
|
@ -203,7 +305,7 @@ class TestLink(unittest.TestCase):
|
|||
self.assertEqual(dest.hash, bytes.fromhex("fb48da0e82e6e01ba0c014513f74540d"))
|
||||
|
||||
l1 = RNS.Link(dest)
|
||||
time.sleep(0.5)
|
||||
time.sleep(LINK_UP_WAIT)
|
||||
self.assertEqual(l1.status, RNS.Link.ACTIVE)
|
||||
|
||||
resource_timeout = 120
|
||||
|
|
@ -226,7 +328,7 @@ class TestLink(unittest.TestCase):
|
|||
self.assertEqual(l1.status, RNS.Link.CLOSED)
|
||||
|
||||
@skipIf(os.getenv('SKIP_NORMAL_TESTS') != None, "Skipping")
|
||||
def test_5_mini_resource(self):
|
||||
def test_05_mini_resource(self):
|
||||
init_rns(self)
|
||||
print("")
|
||||
print("Mini resource test")
|
||||
|
|
@ -243,7 +345,7 @@ class TestLink(unittest.TestCase):
|
|||
self.assertEqual(dest.hash, bytes.fromhex("fb48da0e82e6e01ba0c014513f74540d"))
|
||||
|
||||
l1 = RNS.Link(dest)
|
||||
time.sleep(0.5)
|
||||
time.sleep(LINK_UP_WAIT)
|
||||
self.assertEqual(l1.status, RNS.Link.ACTIVE)
|
||||
|
||||
resource_timeout = 120
|
||||
|
|
@ -266,7 +368,7 @@ class TestLink(unittest.TestCase):
|
|||
self.assertEqual(l1.status, RNS.Link.CLOSED)
|
||||
|
||||
@skipIf(os.getenv('SKIP_NORMAL_TESTS') != None, "Skipping")
|
||||
def test_6_small_resource(self):
|
||||
def test_06_small_resource(self):
|
||||
init_rns(self)
|
||||
print("")
|
||||
print("Small resource test")
|
||||
|
|
@ -282,7 +384,7 @@ class TestLink(unittest.TestCase):
|
|||
self.assertEqual(dest.hash, bytes.fromhex("fb48da0e82e6e01ba0c014513f74540d"))
|
||||
|
||||
l1 = RNS.Link(dest)
|
||||
time.sleep(0.5)
|
||||
time.sleep(LINK_UP_WAIT)
|
||||
self.assertEqual(l1.status, RNS.Link.ACTIVE)
|
||||
|
||||
resource_timeout = 120
|
||||
|
|
@ -301,12 +403,12 @@ class TestLink(unittest.TestCase):
|
|||
print("Resource completed at "+self.size_str(resource_size/t, "b")+"ps")
|
||||
|
||||
l1.teardown()
|
||||
time.sleep(0.5)
|
||||
time.sleep(LINK_UP_WAIT)
|
||||
self.assertEqual(l1.status, RNS.Link.CLOSED)
|
||||
|
||||
|
||||
@skipIf(os.getenv('SKIP_NORMAL_TESTS') != None, "Skipping")
|
||||
def test_7_medium_resource(self):
|
||||
def test_07_medium_resource(self):
|
||||
if RNS.Cryptography.backend() == "internal":
|
||||
print("Skipping medium resource test...")
|
||||
return
|
||||
|
|
@ -326,7 +428,7 @@ class TestLink(unittest.TestCase):
|
|||
self.assertEqual(dest.hash, bytes.fromhex("fb48da0e82e6e01ba0c014513f74540d"))
|
||||
|
||||
l1 = RNS.Link(dest)
|
||||
time.sleep(0.5)
|
||||
time.sleep(LINK_UP_WAIT)
|
||||
self.assertEqual(l1.status, RNS.Link.ACTIVE)
|
||||
|
||||
resource_timeout = 120
|
||||
|
|
@ -345,7 +447,7 @@ class TestLink(unittest.TestCase):
|
|||
print("Resource completed at "+self.size_str(resource_size/t, "b")+"ps")
|
||||
|
||||
l1.teardown()
|
||||
time.sleep(0.5)
|
||||
time.sleep(LINK_UP_WAIT)
|
||||
self.assertEqual(l1.status, RNS.Link.CLOSED)
|
||||
|
||||
large_resource_status = None
|
||||
|
|
@ -353,7 +455,7 @@ class TestLink(unittest.TestCase):
|
|||
TestLink.large_resource_status = resource.status
|
||||
|
||||
@skipIf(os.getenv('SKIP_NORMAL_TESTS') != None, "Skipping")
|
||||
def test_9_large_resource(self):
|
||||
def test_09_large_resource(self):
|
||||
if RNS.Cryptography.backend() == "internal":
|
||||
print("Skipping large resource test...")
|
||||
return
|
||||
|
|
@ -373,7 +475,7 @@ class TestLink(unittest.TestCase):
|
|||
self.assertEqual(dest.hash, bytes.fromhex("fb48da0e82e6e01ba0c014513f74540d"))
|
||||
|
||||
l1 = RNS.Link(dest)
|
||||
time.sleep(0.5)
|
||||
time.sleep(LINK_UP_WAIT)
|
||||
self.assertEqual(l1.status, RNS.Link.ACTIVE)
|
||||
|
||||
resource_timeout = 120
|
||||
|
|
@ -392,7 +494,7 @@ class TestLink(unittest.TestCase):
|
|||
print("Resource completed at "+self.size_str(resource_size/t, "b")+"ps")
|
||||
|
||||
l1.teardown()
|
||||
time.sleep(0.5)
|
||||
time.sleep(LINK_UP_WAIT)
|
||||
self.assertEqual(l1.status, RNS.Link.CLOSED)
|
||||
|
||||
#@skipIf(os.getenv('SKIP_NORMAL_TESTS') != None, "Skipping")
|
||||
|
|
@ -430,7 +532,7 @@ class TestLink(unittest.TestCase):
|
|||
channel.add_message_handler(handle_message)
|
||||
channel.send(test_message)
|
||||
|
||||
time.sleep(0.5)
|
||||
time.sleep(LINK_UP_WAIT)
|
||||
|
||||
self.assertEqual(1, len(received))
|
||||
|
||||
|
|
@ -443,7 +545,7 @@ class TestLink(unittest.TestCase):
|
|||
self.assertEqual(0, len(l1._channel._rx_ring))
|
||||
|
||||
l1.teardown()
|
||||
time.sleep(0.5)
|
||||
time.sleep(LINK_UP_WAIT)
|
||||
self.assertEqual(l1.status, RNS.Link.CLOSED)
|
||||
self.assertEqual(0, len(l1._channel._rx_ring))
|
||||
|
||||
|
|
@ -466,7 +568,7 @@ class TestLink(unittest.TestCase):
|
|||
self.assertEqual(dest.hash, bytes.fromhex("fb48da0e82e6e01ba0c014513f74540d"))
|
||||
|
||||
l1 = RNS.Link(dest)
|
||||
time.sleep(1)
|
||||
time.sleep(LINK_UP_WAIT)
|
||||
self.assertEqual(l1.status, RNS.Link.ACTIVE)
|
||||
buffer = None
|
||||
|
||||
|
|
@ -481,7 +583,7 @@ class TestLink(unittest.TestCase):
|
|||
buffer.write("Hi there".encode("utf-8"))
|
||||
buffer.flush()
|
||||
|
||||
time.sleep(0.5)
|
||||
time.sleep(LINK_UP_WAIT)
|
||||
|
||||
self.assertEqual(1 , len(received))
|
||||
|
||||
|
|
@ -490,7 +592,7 @@ class TestLink(unittest.TestCase):
|
|||
self.assertEqual("Hi there back at you", rx_message)
|
||||
|
||||
l1.teardown()
|
||||
time.sleep(0.5)
|
||||
time.sleep(LINK_UP_WAIT)
|
||||
self.assertEqual(l1.status, RNS.Link.CLOSED)
|
||||
|
||||
# @skipIf(os.getenv('SKIP_NORMAL_TESTS') != None and os.getenv('RUN_SLOW_TESTS') == None, "Skipping")
|
||||
|
|
@ -595,7 +697,7 @@ class TestLink(unittest.TestCase):
|
|||
self.assertEqual(expected_rx_message, rx_message)
|
||||
|
||||
l1.teardown()
|
||||
time.sleep(0.5)
|
||||
time.sleep(LINK_UP_WAIT)
|
||||
self.assertEqual(l1.status, RNS.Link.CLOSED)
|
||||
finally:
|
||||
local_interface.bitrate = original_bitrate
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue