|
|
@@ -33,8 +33,8 @@ import unittest |
|
|
|
from Strobe.Strobe import Strobe, KeccakF |
|
|
|
from Strobe.Strobe import AuthenticationFailed |
|
|
|
|
|
|
|
import lora_comms |
|
|
|
from lora_comms import make_pktbuf, X25519 |
|
|
|
import syote_comms |
|
|
|
from syote_comms import make_pktbuf, X25519 |
|
|
|
import multicast |
|
|
|
from util import * |
|
|
|
|
|
|
@@ -291,11 +291,11 @@ async def main(): |
|
|
|
|
|
|
|
# seed the RNG |
|
|
|
prngseed = os.urandom(64) |
|
|
|
lora_comms.strobe_seed_prng((c_uint8 * |
|
|
|
syote_comms.strobe_seed_prng((c_uint8 * |
|
|
|
len(prngseed))(*prngseed), len(prngseed)) |
|
|
|
|
|
|
|
# Create the state for testing |
|
|
|
commstate = lora_comms.CommsState() |
|
|
|
commstate = syote_comms.CommsState() |
|
|
|
|
|
|
|
import util_load |
|
|
|
client_func = util_load.load_application(args.client) |
|
|
@@ -310,17 +310,17 @@ async def main(): |
|
|
|
for i in range(outbuf[0].pktlen): |
|
|
|
outbuf[0].pkt[i] = ret[i] |
|
|
|
|
|
|
|
cb = lora_comms.process_msgfunc_t(client_call) |
|
|
|
cb = syote_comms.process_msgfunc_t(client_call) |
|
|
|
|
|
|
|
# Initialize everything |
|
|
|
lora_comms.comms_init(commstate, cb, make_pktbuf(shared_key)) |
|
|
|
syote_comms.comms_init(commstate, cb, make_pktbuf(shared_key)) |
|
|
|
|
|
|
|
try: |
|
|
|
while True: |
|
|
|
pkt = await mr.recv() |
|
|
|
msg = pkt[0] |
|
|
|
|
|
|
|
out = lora_comms.comms_process_wrap( |
|
|
|
out = syote_comms.comms_process_wrap( |
|
|
|
commstate, msg) |
|
|
|
|
|
|
|
if out: |
|
|
@@ -917,19 +917,19 @@ class TestLORANode(unittest.IsolatedAsyncioTestCase): |
|
|
|
# seed the RNG |
|
|
|
prngseed = b'abc123' |
|
|
|
from ctypes import c_uint8 |
|
|
|
lora_comms.strobe_seed_prng((c_uint8 * |
|
|
|
syote_comms.strobe_seed_prng((c_uint8 * |
|
|
|
len(prngseed))(*prngseed), len(prngseed)) |
|
|
|
|
|
|
|
# Create the state for testing |
|
|
|
commstate = lora_comms.CommsState() |
|
|
|
commstate = syote_comms.CommsState() |
|
|
|
|
|
|
|
cb = lora_comms.process_msgfunc_t(lambda msg, outbuf: None) |
|
|
|
cb = syote_comms.process_msgfunc_t(lambda msg, outbuf: None) |
|
|
|
|
|
|
|
# Generate shared key |
|
|
|
shared_key = os.urandom(32) |
|
|
|
|
|
|
|
# Initialize everything |
|
|
|
lora_comms.comms_init(commstate, cb, make_pktbuf(shared_key), None, None) |
|
|
|
syote_comms.comms_init(commstate, cb, make_pktbuf(shared_key), None, None) |
|
|
|
|
|
|
|
# Create test fixture, only use it to init crypto state |
|
|
|
tsd = SyncDatagram() |
|
|
@@ -942,13 +942,13 @@ class TestLORANode(unittest.IsolatedAsyncioTestCase): |
|
|
|
msg = os.urandom(16) + b'othre' |
|
|
|
msg = cstate.send_enc(msg) + cstate.send_mac(l.MAC_LEN) |
|
|
|
|
|
|
|
out = lora_comms.comms_process_wrap(commstate, msg) |
|
|
|
out = syote_comms.comms_process_wrap(commstate, msg) |
|
|
|
|
|
|
|
self.assertFalse(out) |
|
|
|
|
|
|
|
# that varous short messages don't cause problems |
|
|
|
for i in range(10): |
|
|
|
out = lora_comms.comms_process_wrap(commstate, b'0' * i) |
|
|
|
out = syote_comms.comms_process_wrap(commstate, b'0' * i) |
|
|
|
|
|
|
|
self.assertFalse(out) |
|
|
|
|
|
|
@@ -959,7 +959,7 @@ class TestLORANode(unittest.IsolatedAsyncioTestCase): |
|
|
|
msg = os.urandom(16) + b' eqreset' |
|
|
|
msg = cstate.send_enc(msg) + cstate.send_mac(l.MAC_LEN) |
|
|
|
|
|
|
|
out = lora_comms.comms_process_wrap(commstate, msg) |
|
|
|
out = syote_comms.comms_process_wrap(commstate, msg) |
|
|
|
|
|
|
|
self.assertFalse(out) |
|
|
|
|
|
|
@@ -967,7 +967,7 @@ class TestLORANode(unittest.IsolatedAsyncioTestCase): |
|
|
|
msg = os.urandom(16) + b'reqreset' |
|
|
|
msg = l.st.send_enc(msg) + l.st.send_mac(l.MAC_LEN) |
|
|
|
|
|
|
|
out = lora_comms.comms_process_wrap(commstate, msg) |
|
|
|
out = syote_comms.comms_process_wrap(commstate, msg) |
|
|
|
|
|
|
|
l.st.recv_enc(out[:-l.MAC_LEN]) |
|
|
|
l.st.recv_mac(out[-l.MAC_LEN:]) |
|
|
@@ -981,7 +981,7 @@ class TestLORANode(unittest.IsolatedAsyncioTestCase): |
|
|
|
msg = b'onfirm' |
|
|
|
msg = cstate.send_enc(msg) + cstate.send_mac(l.MAC_LEN) |
|
|
|
|
|
|
|
out = lora_comms.comms_process_wrap(commstate, msg) |
|
|
|
out = syote_comms.comms_process_wrap(commstate, msg) |
|
|
|
|
|
|
|
self.assertFalse(out) |
|
|
|
|
|
|
@@ -992,7 +992,7 @@ class TestLORANode(unittest.IsolatedAsyncioTestCase): |
|
|
|
msg = b' onfirm' |
|
|
|
msg = cstate.send_enc(msg) + cstate.send_mac(l.MAC_LEN) |
|
|
|
|
|
|
|
out = lora_comms.comms_process_wrap(commstate, msg) |
|
|
|
out = syote_comms.comms_process_wrap(commstate, msg) |
|
|
|
|
|
|
|
self.assertFalse(out) |
|
|
|
|
|
|
@@ -1003,11 +1003,11 @@ class TestLORANode(unittest.IsolatedAsyncioTestCase): |
|
|
|
|
|
|
|
# seed the RNG |
|
|
|
prngseed = b'abc123' |
|
|
|
lora_comms.strobe_seed_prng((c_uint8 * |
|
|
|
syote_comms.strobe_seed_prng((c_uint8 * |
|
|
|
len(prngseed))(*prngseed), len(prngseed)) |
|
|
|
|
|
|
|
# Create the state for testing |
|
|
|
commstate = lora_comms.CommsState() |
|
|
|
commstate = syote_comms.CommsState() |
|
|
|
|
|
|
|
# These are the expected messages and their arguments |
|
|
|
exptmsgs = [ |
|
|
@@ -1031,7 +1031,7 @@ class TestLORANode(unittest.IsolatedAsyncioTestCase): |
|
|
|
raise RuntimeError('cmd not found') |
|
|
|
|
|
|
|
# wrap the callback function |
|
|
|
cb = lora_comms.process_msgfunc_t(procmsg) |
|
|
|
cb = syote_comms.process_msgfunc_t(procmsg) |
|
|
|
|
|
|
|
class CCodeSD(MockSyncDatagram): |
|
|
|
async def runner(self): |
|
|
@@ -1040,7 +1040,7 @@ class TestLORANode(unittest.IsolatedAsyncioTestCase): |
|
|
|
inmsg = await self.get() |
|
|
|
|
|
|
|
# process the test message |
|
|
|
out = lora_comms.comms_process_wrap( |
|
|
|
out = syote_comms.comms_process_wrap( |
|
|
|
commstate, inmsg) |
|
|
|
|
|
|
|
# make sure the reply matches length |
|
|
@@ -1050,7 +1050,7 @@ class TestLORANode(unittest.IsolatedAsyncioTestCase): |
|
|
|
origmsg = out |
|
|
|
|
|
|
|
# pretend that the reply didn't make it |
|
|
|
out = lora_comms.comms_process_wrap( |
|
|
|
out = syote_comms.comms_process_wrap( |
|
|
|
commstate, inmsg) |
|
|
|
|
|
|
|
# make sure that the reply matches |
|
|
@@ -1065,7 +1065,7 @@ class TestLORANode(unittest.IsolatedAsyncioTestCase): |
|
|
|
respkey = X25519.gen() |
|
|
|
|
|
|
|
# Initialize everything |
|
|
|
lora_comms.comms_init(commstate, cb, None, make_pktbuf(respkey.getpriv()), make_pktbuf(initkey.getpub())) |
|
|
|
syote_comms.comms_init(commstate, cb, None, make_pktbuf(respkey.getpriv()), make_pktbuf(initkey.getpub())) |
|
|
|
|
|
|
|
# Create test fixture |
|
|
|
tsd = CCodeSD() |
|
|
@@ -1100,11 +1100,11 @@ class TestLORANode(unittest.IsolatedAsyncioTestCase): |
|
|
|
|
|
|
|
# seed the RNG |
|
|
|
prngseed = b'abc123' |
|
|
|
lora_comms.strobe_seed_prng((c_uint8 * |
|
|
|
syote_comms.strobe_seed_prng((c_uint8 * |
|
|
|
len(prngseed))(*prngseed), len(prngseed)) |
|
|
|
|
|
|
|
# Create the state for testing |
|
|
|
commstate = lora_comms.CommsState() |
|
|
|
commstate = syote_comms.CommsState() |
|
|
|
|
|
|
|
# These are the expected messages and their arguments |
|
|
|
exptmsgs = [ |
|
|
@@ -1128,7 +1128,7 @@ class TestLORANode(unittest.IsolatedAsyncioTestCase): |
|
|
|
raise RuntimeError('cmd not found') |
|
|
|
|
|
|
|
# wrap the callback function |
|
|
|
cb = lora_comms.process_msgfunc_t(procmsg) |
|
|
|
cb = syote_comms.process_msgfunc_t(procmsg) |
|
|
|
|
|
|
|
class CCodeSD(MockSyncDatagram): |
|
|
|
async def runner(self): |
|
|
@@ -1137,7 +1137,7 @@ class TestLORANode(unittest.IsolatedAsyncioTestCase): |
|
|
|
inmsg = await self.get() |
|
|
|
|
|
|
|
# process the test message |
|
|
|
out = lora_comms.comms_process_wrap( |
|
|
|
out = syote_comms.comms_process_wrap( |
|
|
|
commstate, inmsg) |
|
|
|
|
|
|
|
# make sure the reply matches length |
|
|
@@ -1147,7 +1147,7 @@ class TestLORANode(unittest.IsolatedAsyncioTestCase): |
|
|
|
origmsg = out |
|
|
|
|
|
|
|
# pretend that the reply didn't make it |
|
|
|
out = lora_comms.comms_process_wrap( |
|
|
|
out = syote_comms.comms_process_wrap( |
|
|
|
commstate, inmsg) |
|
|
|
|
|
|
|
# make sure that the reply matches |
|
|
@@ -1161,7 +1161,7 @@ class TestLORANode(unittest.IsolatedAsyncioTestCase): |
|
|
|
shared_key = os.urandom(32) |
|
|
|
|
|
|
|
# Initialize everything |
|
|
|
lora_comms.comms_init(commstate, cb, make_pktbuf(shared_key), None, None) |
|
|
|
syote_comms.comms_init(commstate, cb, make_pktbuf(shared_key), None, None) |
|
|
|
|
|
|
|
# Create test fixture |
|
|
|
tsd = CCodeSD() |
|
|
@@ -1203,11 +1203,11 @@ class TestLORANode(unittest.IsolatedAsyncioTestCase): |
|
|
|
|
|
|
|
# seed the RNG |
|
|
|
prngseed = b'abc123' |
|
|
|
lora_comms.strobe_seed_prng((c_uint8 * |
|
|
|
syote_comms.strobe_seed_prng((c_uint8 * |
|
|
|
len(prngseed))(*prngseed), len(prngseed)) |
|
|
|
|
|
|
|
# Create the state for testing |
|
|
|
commstate = lora_comms.CommsState() |
|
|
|
commstate = syote_comms.CommsState() |
|
|
|
|
|
|
|
# These are the expected messages and their arguments |
|
|
|
exptmsgs = [ |
|
|
@@ -1231,7 +1231,7 @@ class TestLORANode(unittest.IsolatedAsyncioTestCase): |
|
|
|
raise RuntimeError('cmd not found: %d' % cmd) |
|
|
|
|
|
|
|
# wrap the callback function |
|
|
|
cb = lora_comms.process_msgfunc_t(procmsg) |
|
|
|
cb = syote_comms.process_msgfunc_t(procmsg) |
|
|
|
|
|
|
|
class FlipMsg(object): |
|
|
|
async def flipmsg(self): |
|
|
@@ -1239,7 +1239,7 @@ class TestLORANode(unittest.IsolatedAsyncioTestCase): |
|
|
|
inmsg = await self.get() |
|
|
|
|
|
|
|
# process the test message |
|
|
|
out = lora_comms.comms_process_wrap( |
|
|
|
out = syote_comms.comms_process_wrap( |
|
|
|
commstate, inmsg) |
|
|
|
|
|
|
|
# pass the reply back |
|
|
@@ -1257,7 +1257,7 @@ class TestLORANode(unittest.IsolatedAsyncioTestCase): |
|
|
|
inmsg = b'0'*24 |
|
|
|
|
|
|
|
# process the bogus message |
|
|
|
out = lora_comms.comms_process_wrap( |
|
|
|
out = syote_comms.comms_process_wrap( |
|
|
|
commstate, inmsg) |
|
|
|
|
|
|
|
# make sure there was not a response |
|
|
@@ -1290,7 +1290,7 @@ class TestLORANode(unittest.IsolatedAsyncioTestCase): |
|
|
|
shared_key = os.urandom(32) |
|
|
|
|
|
|
|
# Initialize everything |
|
|
|
lora_comms.comms_init(commstate, cb, make_pktbuf(shared_key), None, None) |
|
|
|
syote_comms.comms_init(commstate, cb, make_pktbuf(shared_key), None, None) |
|
|
|
|
|
|
|
# Create test fixture |
|
|
|
tsd = CCodeSD1() |
|
|
@@ -1340,11 +1340,11 @@ class TestLoRaNodeMulticast(unittest.IsolatedAsyncioTestCase): |
|
|
|
|
|
|
|
# seed the RNG |
|
|
|
prngseed = b'abc123' |
|
|
|
lora_comms.strobe_seed_prng((c_uint8 * |
|
|
|
syote_comms.strobe_seed_prng((c_uint8 * |
|
|
|
len(prngseed))(*prngseed), len(prngseed)) |
|
|
|
|
|
|
|
# Create the state for testing |
|
|
|
commstate = lora_comms.CommsState() |
|
|
|
commstate = syote_comms.CommsState() |
|
|
|
|
|
|
|
# These are the expected messages and their arguments |
|
|
|
exptmsgs = [ |
|
|
@@ -1367,13 +1367,13 @@ class TestLoRaNodeMulticast(unittest.IsolatedAsyncioTestCase): |
|
|
|
raise RuntimeError('cmd not found') |
|
|
|
|
|
|
|
# wrap the callback function |
|
|
|
cb = lora_comms.process_msgfunc_t(procmsg) |
|
|
|
cb = syote_comms.process_msgfunc_t(procmsg) |
|
|
|
|
|
|
|
# Generate shared key |
|
|
|
shared_key = os.urandom(32) |
|
|
|
|
|
|
|
# Initialize everything |
|
|
|
lora_comms.comms_init(commstate, cb, make_pktbuf(shared_key), None, None) |
|
|
|
syote_comms.comms_init(commstate, cb, make_pktbuf(shared_key), None, None) |
|
|
|
|
|
|
|
# create the object we are testing |
|
|
|
msd = MulticastSyncDatagram(self.maddr) |
|
|
@@ -1394,7 +1394,7 @@ class TestLoRaNodeMulticast(unittest.IsolatedAsyncioTestCase): |
|
|
|
pkt = await mr.recv() |
|
|
|
msg = pkt[0] |
|
|
|
|
|
|
|
out = lora_comms.comms_process_wrap( |
|
|
|
out = syote_comms.comms_process_wrap( |
|
|
|
commstate, msg) |
|
|
|
|
|
|
|
if out: |
|
|
|