lora-irrigation/lora.py

# Copyright 2021 John-Mark Gurney.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions
# are met:
# 1. Redistributions of source code must retain the above copyright
#    notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
#    notice, this list of conditions and the following disclaimer in the
#    documentation and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
# OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
# HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
# OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
# SUCH DAMAGE.
#

import asyncio
import codecs
import contextlib
import functools
import itertools
import os
import pathlib
import shutil
import subprocess
import sys
import tempfile
import unittest

from unittest.mock import patch
from subprocess import DEVNULL, PIPE

from Strobe.Strobe import Strobe, KeccakF
from Strobe.Strobe import AuthenticationFailed

import syote_comms
from syote_comms import make_pktbuf, X25519
import multicast
from util import *

# Response to command will be the CMD and any arguments if needed.
# The command is encoded as an unsigned byte
CMD_TERMINATE = 1	# no args: terminate the sesssion, reply confirms

# The follow commands are queue up, but will be acknoledged when queued
CMD_WAITFOR = 2		# arg: (length): waits for length seconds
CMD_RUNFOR = 3		# arg: (chan, length): turns on chan for length seconds
CMD_PING = 4		# arg: (): a no op command
CMD_SETUNSET = 5	# arg: (chan, val): sets chan to val
CMD_ADV = 6		# arg: ([cnt]): advances to the next cnt (default 1) command
CMD_CLEAR = 7		# arg: (): clears all future commands, but keeps current running
CMD_KEY = 8		# arg: ([key reports]): standard USB HID key reports, return is number added

_directmap = [
	(40, '\n\x1b\b\t -=[]\\#;\'`,./'),
	(40, '\r'),
]
_keymap = { chr(x): x - ord('a') + 4 for x in range(ord('a'), ord('z') + 1) } | \
	{ '0': 39 } | \
	{ chr(x): x - ord('1') + 30 for x in range(ord('1'), ord('9') + 1) } | \
	{ x: i + base for base, keys in _directmap for i, x in enumerate(keys) }
_shiftmaplist = '!1@2#3$4%5^6&7*8(9)0_-+=~`{[}]|\\:;"\'<,>.?/'
_completemap = { x: (False, y) for x, y in _keymap.items() } | \
	{ x.upper(): (True, y) for x, y in _keymap.items() if x != x.upper() } | \
	{ x: (True, _keymap[y]) for x, y in zip(_shiftmaplist[::2], _shiftmaplist[1::2]) }


def makekeybuf(s):
	blank = b'\x00' * 8

	# start clean
	ret = [ blank ]
	templ = bytearray([ 0 ] * 8)

	for i in s:
		shift, key = _completemap[i]

		if ret[-1][2] == key:
			ret.append(blank)

		templ[2] = key
		templ[0] = 2 if shift else 0

		ret.append(templ[:])

	# end clean
	ret.append(blank)

	return b''.join(ret)

class LORANode(object):
	'''Implement a LORANode initiator.

	There are currently two implemented modes, one is shared, and then
	a shared key must be provided to the shared keyword argument.

	The other is ecdhe mode, which requires an X25519 key to be passed
	in to init_key, and the respondent's public key to be passed in to
	resp_pub.
	'''

	SHARED_DOMAIN = b'com.funkthat.lora.irrigation.shared.v0.0.1'
	ECDHE_DOMAIN = b'com.funkthat.lora.irrigation.ecdhe.v0.0.1'

	MAC_LEN = 8

	def __init__(self, syncdatagram, shared=None, init_key=None, resp_pub=None):
		self.sd = syncdatagram
		if shared is not None:
			self.st = Strobe(self.SHARED_DOMAIN, F=KeccakF(800))
			self.st.key(shared)
			self.start = self.shared_start
		elif init_key is not None and resp_pub is not None:
			self.st = Strobe(self.ECDHE_DOMAIN, F=KeccakF(800))
			self.key = init_key
			self.resp_pub = resp_pub
			self.st.key(init_key.getpub() + resp_pub)
			self.start = self.ecdhe_start
		else:
			raise RuntimeError('invalid combination of keys provided')

	async def shared_start(self):
		resp = await self.sendrecvvalid(os.urandom(16) + b'reqreset')

		self.st.ratchet()

		pkt = await self.sendrecvvalid(b'confirm')

		if pkt != b'confirmed':
			raise RuntimeError('got invalid response: %s' %
			    repr(pkt))

	async def ecdhe_start(self):
		ephkey = X25519.gen()

		resp = await self.sendrecvvalid(ephkey.getpub() + b'reqreset',
		    fun=lambda: self.st.key(ephkey.dh(self.resp_pub) + self.key.dh(self.resp_pub)))

		self.st.key(ephkey.dh(resp) + self.key.dh(resp))

		pkt = await self.sendrecvvalid(b'confirm')

		if pkt != b'confirmed':
			raise RuntimeError('got invalid response: %s' %
			    repr(pkt))

	async def sendrecvvalid(self, msg, fun=None):
		msg = self.st.send_enc(msg) + self.st.send_mac(self.MAC_LEN)

		if fun is not None:
			fun()

		origstate = self.st.copy()

		while True:
			resp = await self.sd.sendtillrecv(msg, .50)
			#_debprint('got:', resp)

			# skip empty messages
			if len(resp) == 0:
				continue

			try:
				decmsg = self.st.recv_enc(resp[:-self.MAC_LEN])
				self.st.recv_mac(resp[-self.MAC_LEN:])
				break
			except AuthenticationFailed:
				# didn't get a valid packet, restore
				# state and retry

				#_debprint('failed')
				self.st.set_state_from(origstate)

		#_debprint('got rep:', repr(resp), repr(decmsg))
		return decmsg

	@staticmethod
	def _encodeargs(*args):
		r = []
		for i in args:
			if isinstance(i, bytes):
				r.append(i)
			else:
				r.append(i.to_bytes(4, byteorder='little'))

		return b''.join(r)

	async def _sendcmd(self, cmd, *args):
		cmdbyte = cmd.to_bytes(1, byteorder='little')
		resp = await self.sendrecvvalid(cmdbyte + self._encodeargs(*args))

		if resp[0:1] != cmdbyte:
			raise RuntimeError(
			    'response does not match, got: %s, expected: %s' %
			    (repr(resp[0:1]), repr(cmdbyte)))

		r = resp[1:]
		if r:
			return r

	async def waitfor(self, length):
		return await self._sendcmd(CMD_WAITFOR, length)

	async def runfor(self, chan, length):
		return await self._sendcmd(CMD_RUNFOR, chan, length)

	async def setunset(self, chan, val):
		return await self._sendcmd(CMD_SETUNSET, chan, val)

	async def ping(self):
		return await self._sendcmd(CMD_PING)

	async def type(self, s):
		keys = makekeybuf(s)

		while keys:
			r = await self._sendcmd(CMD_KEY, keys[:8 * 6])
			r = int.from_bytes(r, byteorder='little')
			keys = keys[r * 8:]

	async def adv(self, cnt=None):
		args = ()
		if cnt is not None:
			args = (cnt, )
		return await self._sendcmd(CMD_ADV, *args)

	async def clear(self):
		return await self._sendcmd(CMD_CLEAR)

	async def terminate(self):
		return await self._sendcmd(CMD_TERMINATE)

class SyncDatagram(object):
	'''Base interface for a more simple synchronous interface.'''

	async def recv(self, timeout=None): #pragma: no cover
		'''Receive a datagram.  If timeout is not None, wait that many
		seconds, and if nothing is received in that time, raise an
		asyncio.TimeoutError exception.'''

		raise NotImplementedError

	async def send(self, data): #pragma: no cover

		raise NotImplementedError

	async def sendtillrecv(self, data, freq):
		'''Send the datagram in data, every freq seconds until a datagram
		is received.  If timeout seconds happen w/o receiving a datagram,
		then raise an TimeoutError exception.'''

		while True:
			#_debprint('sending:', repr(data))
			await self.send(data)
			try:
				return await self.recv(freq)
			except asyncio.TimeoutError:
				pass

class MulticastSyncDatagram(SyncDatagram):
	'''
	An implementation of SyncDatagram that uses the provided
	multicast address maddr as the source/sink of the packets.

	Note that once created, the start coroutine needs to be
	await'd before being passed to a LORANode so that everything
	is running.
	'''

	# Note: sent packets will be received.  A similar method to
	# what was done in multicast.{to,from}_loragw could be done
	# here as well, that is passing in a set of packets to not
	# pass back up.

	def __init__(self, maddr):
		self.maddr = maddr
		self._ignpkts = set()

	async def start(self):
		self.mr = await multicast.create_multicast_receiver(self.maddr)
		self.mt = await multicast.create_multicast_transmitter(
		    self.maddr)

	async def _recv(self):
		while True:
			pkt = await self.mr.recv()
			pkt = pkt[0]
			if pkt not in self._ignpkts:
				return pkt

			self._ignpkts.remove(pkt)

	async def recv(self, timeout=None): #pragma: no cover
		r = await asyncio.wait_for(self._recv(), timeout=timeout)

		return r

	async def send(self, data): #pragma: no cover
		self._ignpkts.add(bytes(data))
		await self.mt.send(data)

	def close(self):
		'''Shutdown communications.'''

		self.mr.close()
		self.mr = None
		self.mt.close()
		self.mt = None

def listsplit(lst, item):
	try:
		idx = lst.index(item)
	except ValueError:
		return lst, []

	return lst[:idx], lst[idx + 1:]

async def main():
	import argparse

	from loraserv import DEFAULT_MADDR as maddr

	parser = argparse.ArgumentParser(description='This is an implementation of both the server and client implementing syote secure IoT protocol.',
	    epilog='Both -k and -p MUST be used together.  One of either -s or the combone -k & -p MUST be specified.')

	parser.add_argument('-f', dest='schedfile', metavar='filename', type=str,
	    help='Use commands from the file.  One command per line.')
	parser.add_argument('-k', dest='privkey', metavar='privfile', type=str,
	    help='File containing a hex encoded private key.')
	parser.add_argument('-p', dest='pubkey', metavar='pubfile', type=str,
	    help='File containing a hex encoded public key.')
	parser.add_argument('-r', dest='client', metavar='module:function', type=str,
	    help='Create a respondant instead of sending commands.  Commands will be passed to the function.')
	parser.add_argument('-s', dest='shared_key', metavar='shared_key', type=str,
	    help='File containing the shared key to use (note, any white space is included).')
	parser.add_argument('args', metavar='CMD_ARG', type=str, nargs='*',
	    help='Various commands to send to the device.')

	args = parser.parse_args()

	# make sure a key is specified.
	if args.privkey is None and args.pubkey is None and \
	    args.shared_key is None:
		parser.error('a key must be specified (either -k and -p, or -s)')

	# make sure if one is specified, both are.
	if (args.privkey is not None or args.pubkey is not None) and \
	    (args.privkey is None or args.pubkey is None):
		parser.error('both -k and -p MUST be specified if one is.')

	if args.shared_key is not None:
		skdata = pathlib.Path(args.shared_key).read_bytes()
		lorakwargs = dict(shared=skdata)
		commsinitargs = (make_pktbuf(lorakwargs['shared']), None, None)
	else:
		privkeydata = pathlib.Path(args.privkey).read_text().strip()
		privkey = X25519.frombytes(codecs.decode(privkeydata, 'hex'))

		pubkeydata = pathlib.Path(args.pubkey).read_text().strip()
		pubkey = codecs.decode(pubkeydata, 'hex')

		lorakwargs = dict(init_key=privkey, resp_pub=pubkey)
		commsinitargs = (None, make_pktbuf(privkey.getpriv()),
		    make_pktbuf(pubkey))

	if args.client:
		# Run a client

		mr = await multicast.create_multicast_receiver(maddr)
		mt = await multicast.create_multicast_transmitter(maddr)

		from ctypes import c_uint8

		# seed the RNG
		prngseed = os.urandom(64)
		syote_comms._lib.strobe_seed_prng((c_uint8 *
		    len(prngseed))(*prngseed), len(prngseed))

		# Create the state for testing
		commstate = syote_comms.CommsState()

		import util_load
		client_func = util_load.load_application(args.client)

		def client_call(msg, outbuf):
			ret = client_func(msg._from())

			if len(ret) > outbuf[0].pktlen:
				ret = b'error, too long buffer: %d' % len(ret)

			outbuf[0].pktlen = min(len(ret), outbuf[0].pktlen)
			for i in range(outbuf[0].pktlen):
				outbuf[0].pkt[i] = ret[i]

		cb = syote_comms.process_msgfunc_t(client_call)

		# Initialize everything
		syote_comms.comms_init(commstate, cb, *commsinitargs)

		try:
			while True:
				pkt = await mr.recv()
				msg = pkt[0]

				#_debprint('procmsg:', repr(msg))

				out = syote_comms.comms_process_wrap(
				    commstate, msg)

				#_debprint('resp:', repr(out))

				if out:
					await mt.send(out)
		finally:
			mr.close()
			mt.close()
		sys.exit(0)

	msd = MulticastSyncDatagram(maddr)
	await msd.start()

	l = LORANode(msd, **lorakwargs)

	await l.start()

	valid_cmds = {
	    'waitfor', 'setunset', 'runfor', 'ping', 'adv', 'clear',
	    'terminate', 'type',
	}

	if args.args and args.schedfile:
		parser.error('only one of -f or arguments can be specified.')

	if args.args:
		cmds = list(args.args)
		cmdargs = []
		while cmds:
			a, cmds = listsplit(cmds, '--')
			cmdargs.append(a)
	else:
		with open(args.schedfile) as fp:
			cmdargs = [ x.split() for x in fp.readlines() ]

	while cmdargs:
		cmd, *args = cmdargs.pop(0)

		if cmd not in valid_cmds:
			print('invalid command:', repr(cmd))
			sys.exit(1)

		fun = getattr(l, cmd)

		try:
			await fun(*(int(x) for x in args))
		except:
			await fun(*args)

if __name__ == '__main__':
	asyncio.run(main())

class MockSyncDatagram(SyncDatagram):
	'''A testing version of SyncDatagram.  Define a method runner which
	implements part of the sequence.  In the function, await on either
	self.get, to wait for the other side to send something, or await
	self.put w/ data to send.'''

	def __init__(self):
		self.sendq = asyncio.Queue()
		self.recvq = asyncio.Queue()
		self.task = asyncio.create_task(self.runner())

		self.get = self.sendq.get
		self.put = self.recvq.put

	async def drain(self):
		'''Wait for the runner thread to finish up.'''

		return await self.task

	async def runner(self): #pragma: no cover
		raise NotImplementedError

	async def recv(self, timeout=None):
		return await self.recvq.get()

	async def send(self, data):
		return await self.sendq.put(data)

	def __del__(self): #pragma: no cover
		if self.task is not None and not self.task.done():
			self.task.cancel()

class TestSyncData(unittest.IsolatedAsyncioTestCase):
	async def test_syncsendtillrecv(self):
		class MySync(SyncDatagram):
			def __init__(self):
				self.sendq = []
				self.resp = [ asyncio.TimeoutError(), b'a' ]

			async def recv(self, timeout=None):
				assert timeout == 1
				r = self.resp.pop(0)
				if isinstance(r, Exception):
					raise r

				return r

			async def send(self, data):
				self.sendq.append(data)

		ms = MySync()

		r = await ms.sendtillrecv(b'foo', 1)

		self.assertEqual(r, b'a')
		self.assertEqual(ms.sendq, [ b'foo', b'foo' ])

class AsyncSequence(object):
	'''
	Object used for sequencing async functions.  To use, use the
	asynchronous context manager created by the sync method.  For
	example:
	seq = AsyncSequence()
	async func1():
		async with seq.sync(1):
			second_fun()

	async func2():
		async with seq.sync(0):
			first_fun()

	This will make sure that function first_fun is run before running
	the function second_fun.  If a previous block raises an Exception,
	it will be passed up, and all remaining blocks (and future ones)
	will raise a CancelledError to help ensure that any tasks are
	properly cleaned up.
	'''

	def __init__(self, positerfactory=lambda: itertools.count()):
		'''The argument positerfactory, is a factory that will
		create an iterator that will be used for the values that
		are passed to the sync method.'''

		self.positer = positerfactory()
		self.token = object()
		self.die = False
		self.waiting = {
			next(self.positer): self.token
		}

	async def simpsync(self, pos):
		async with self.sync(pos):
			pass

	@contextlib.asynccontextmanager
	async def sync(self, pos):
		'''An async context manager that will be run when it's
		turn arrives.  It will only run when all the previous
		items in the iterator has been successfully run.'''

		if self.die:
			raise asyncio.CancelledError('seq cancelled')

		if pos in self.waiting:
			if self.waiting[pos] is not self.token:
				raise RuntimeError('pos already waiting!')
		else:
			fut = asyncio.Future()
			self.waiting[pos] = fut
			await fut

		# our time to shine!
		del self.waiting[pos]

		try:
			yield None
		except Exception as e:
			# if we got an exception, things went pear shaped,
			# shut everything down, and any future calls.

			#_debprint('dieing...', repr(e))
			self.die = True

			# cancel existing blocks
			while self.waiting:
				k, v = self.waiting.popitem()
				#_debprint('canceling: %s' % repr(k))
				if v is self.token:
					continue

				# for Python 3.9:
				# msg='pos %s raised exception: %s' %
				#     (repr(pos), repr(e))
				v.cancel()

			# populate real exception up
			raise
		else:
			# handle next
			nextpos = next(self.positer)

			if nextpos in self.waiting:
				#_debprint('np:', repr(self), nextpos,
				#    repr(self.waiting[nextpos]))
				self.waiting[nextpos].set_result(None)
			else:
				self.waiting[nextpos] = self.token

class TestSequencing(unittest.IsolatedAsyncioTestCase):
	@timeout(2)
	async def test_seq_alreadywaiting(self):
		waitseq = AsyncSequence()

		seq = AsyncSequence()

		async def fun1():
			async with waitseq.sync(1):
				pass

		async def fun2():
			async with seq.sync(1):
				async with waitseq.sync(1): # pragma: no cover
					pass

		task1 = asyncio.create_task(fun1())
		task2 = asyncio.create_task(fun2())

		# spin things to make sure things advance
		await asyncio.sleep(0)

		async with seq.sync(0):
			pass

		with self.assertRaises(RuntimeError):
			await task2

		async with waitseq.sync(0):
			pass

		await task1

	@timeout(2)
	async def test_seqexc(self):
		seq = AsyncSequence()

		excseq = AsyncSequence()

		async def excfun1():
			async with seq.sync(1):
				pass

			async with excseq.sync(0):
				raise ValueError('foo')

		# that a block that enters first, but runs after
		# raises an exception
		async def excfun2():
			async with seq.sync(0):
				pass

			async with excseq.sync(1): # pragma: no cover
				pass

		# that a block that enters after, raises an
		# exception
		async def excfun3():
			async with seq.sync(2):
				pass

			async with excseq.sync(2): # pragma: no cover
				pass

		task1 = asyncio.create_task(excfun1())
		task2 = asyncio.create_task(excfun2())
		task3 = asyncio.create_task(excfun3())

		with self.assertRaises(ValueError):
			await task1

		with self.assertRaises(asyncio.CancelledError):
			await task2

		with self.assertRaises(asyncio.CancelledError):
			await task3

	@timeout(2)
	async def test_seq(self):
		# test that a seq object when created
		seq = AsyncSequence(lambda: itertools.count(1))

		col = []

		async def fun1():
			async with seq.sync(1):
				col.append(1)

			async with seq.sync(2):
				col.append(2)

			async with seq.sync(4):
				col.append(4)

		async def fun2():
			async with seq.sync(3):
				col.append(3)

			async with seq.sync(6):
				col.append(6)

		async def fun3():
			async with seq.sync(5):
				col.append(5)

		# and various functions are run
		task1 = asyncio.create_task(fun1())
		task2 = asyncio.create_task(fun2())
		task3 = asyncio.create_task(fun3())

		# and the functions complete
		await task3
		await task2
		await task1

		# that the order they ran in was correct
		self.assertEqual(col, list(range(1, 7)))

class TestLORANode(unittest.IsolatedAsyncioTestCase):
	shared_domain = b'com.funkthat.lora.irrigation.shared.v0.0.1'
	ecdhe_domain = b'com.funkthat.lora.irrigation.ecdhe.v0.0.1'

	def test_initparams(self):
		# make sure no keys fails
		with self.assertRaises(RuntimeError):
			l = LORANode(None)

	@timeout(2)
	async def test_lora_ecdhe(self):
		_self = self
		initkey = X25519.gen()
		respkey = X25519.gen()

		class TestSD(MockSyncDatagram):
			async def sendgettest(self, msg):
				'''Send the message, but make sure that if a
				bad message is sent afterward, that it replies
				w/ the same previous message.
				'''

				await self.put(msg)
				resp = await self.get()

				await self.put(b'bogusmsg' * 5)

				resp2 = await self.get()

				_self.assertEqual(resp, resp2)

				return resp

			async def runner(self):
				# as respondant

				l = Strobe(_self.ecdhe_domain, F=KeccakF(800))

				l.key(initkey.getpub() + respkey.getpub())

				# start handshake
				r = await self.get()

				# get eph key w/ reqreset
				pkt = l.recv_enc(r[:-8])
				l.recv_mac(r[-8:])

				assert pkt.endswith(b'reqreset')

				ephpub = pkt[:-len(b'reqreset')]

				# make sure junk gets ignored
				await self.put(b'sdlfkj')

				# and that the packet remains the same
				_self.assertEqual(r, await self.get())

				# and a couple more times
				await self.put(b'0' * 24)
				_self.assertEqual(r, await self.get())
				await self.put(b'0' * 32)
				_self.assertEqual(r, await self.get())

				# update the keys
				l.key(respkey.dh(ephpub) + respkey.dh(initkey.getpub()))

				# generate our eph key
				ephkey = X25519.gen()

				# send the response
				await self.put(l.send_enc(ephkey.getpub()) +
				    l.send_mac(8))

				l.key(ephkey.dh(ephpub) + ephkey.dh(initkey.getpub()))

				# get the confirmation message
				r = await self.get()

				# test the resend capabilities
				await self.put(b'0' * 24)
				_self.assertEqual(r, await self.get())

				# decode confirmation message
				c = l.recv_enc(r[:-8])
				l.recv_mac(r[-8:])

				# assert that we got it
				_self.assertEqual(c, b'confirm')

				# send confirmed reply
				r = await self.sendgettest(l.send_enc(
				    b'confirmed') + l.send_mac(8))

				# test and decode remaining command messages
				cmd = l.recv_enc(r[:-8])
				l.recv_mac(r[-8:])

				assert cmd[0] == CMD_WAITFOR
				assert int.from_bytes(cmd[1:],
				     byteorder='little') == 30

				r = await self.sendgettest(l.send_enc(
				    cmd[0:1]) + l.send_mac(8))

				cmd = l.recv_enc(r[:-8])
				l.recv_mac(r[-8:])

				assert cmd[0] == CMD_RUNFOR
				assert int.from_bytes(cmd[1:5],
				     byteorder='little') == 1
				assert int.from_bytes(cmd[5:],
				     byteorder='little') == 50

				r = await self.sendgettest(l.send_enc(
				    cmd[0:1]) + l.send_mac(8))

				cmd = l.recv_enc(r[:-8])
				l.recv_mac(r[-8:])

				assert cmd[0] == CMD_TERMINATE

				await self.put(l.send_enc(cmd[0:1]) +
				    l.send_mac(8))

		tsd = TestSD()

		# make sure it fails w/o both specified
		with self.assertRaises(RuntimeError):
			l = LORANode(tsd, init_key=initkey)

		with self.assertRaises(RuntimeError):
			l = LORANode(tsd, resp_pub=respkey.getpub())

		l = LORANode(tsd, init_key=initkey, resp_pub=respkey.getpub())

		await l.start()

		await l.waitfor(30)

		await l.runfor(1, 50)

		await l.terminate()

		await tsd.drain()

		# Make sure all messages have been processed
		self.assertTrue(tsd.sendq.empty())
		self.assertTrue(tsd.recvq.empty())
		#_debprint('done')

	@timeout(2)
	async def test_lora_shared(self):
		_self = self
		shared_key = os.urandom(32)

		class TestSD(MockSyncDatagram):
			async def sendgettest(self, msg):
				'''Send the message, but make sure that if a
				bad message is sent afterward, that it replies
				w/ the same previous message.
				'''

				await self.put(msg)
				resp = await self.get()

				await self.put(b'bogusmsg' * 5)

				resp2 = await self.get()

				_self.assertEqual(resp, resp2)

				return resp

			async def runner(self):
				l = Strobe(TestLORANode.shared_domain, F=KeccakF(800))

				l.key(shared_key)

				# start handshake
				r = await self.get()

				pkt = l.recv_enc(r[:-8])
				l.recv_mac(r[-8:])

				assert pkt.endswith(b'reqreset')

				# make sure junk gets ignored
				await self.put(b'sdlfkj')

				# and that the packet remains the same
				_self.assertEqual(r, await self.get())

				# and a couple more times
				await self.put(b'0' * 24)
				_self.assertEqual(r, await self.get())
				await self.put(b'0' * 32)
				_self.assertEqual(r, await self.get())

				# send the response
				await self.put(l.send_enc(os.urandom(16)) +
				    l.send_mac(8))

				# require no more back tracking at this point
				l.ratchet()

				# get the confirmation message
				r = await self.get()

				# test the resend capabilities
				await self.put(b'0' * 24)
				_self.assertEqual(r, await self.get())

				# decode confirmation message
				c = l.recv_enc(r[:-8])
				l.recv_mac(r[-8:])

				# assert that we got it
				_self.assertEqual(c, b'confirm')

				# send confirmed reply
				r = await self.sendgettest(l.send_enc(
				    b'confirmed') + l.send_mac(8))

				# test and decode remaining command messages
				cmd = l.recv_enc(r[:-8])
				l.recv_mac(r[-8:])

				assert cmd[0] == CMD_WAITFOR
				assert int.from_bytes(cmd[1:],
				     byteorder='little') == 30

				r = await self.sendgettest(l.send_enc(
				    cmd[0:1]) + l.send_mac(8))

				cmd = l.recv_enc(r[:-8])
				l.recv_mac(r[-8:])

				assert cmd[0] == CMD_RUNFOR
				assert int.from_bytes(cmd[1:5],
				     byteorder='little') == 1
				assert int.from_bytes(cmd[5:],
				     byteorder='little') == 50

				r = await self.sendgettest(l.send_enc(
				    cmd[0:1]) + l.send_mac(8))

				cmd = l.recv_enc(r[:-8])
				l.recv_mac(r[-8:])

				assert cmd[0] == CMD_TERMINATE

				await self.put(l.send_enc(cmd[0:1]) +
				    l.send_mac(8))

		tsd = TestSD()
		l = LORANode(tsd, shared=shared_key)

		await l.start()

		await l.waitfor(30)

		await l.runfor(1, 50)

		await l.terminate()

		await tsd.drain()

		# Make sure all messages have been processed
		self.assertTrue(tsd.sendq.empty())
		self.assertTrue(tsd.recvq.empty())
		#_debprint('done')

	@timeout(2)
	async def test_ccode_badmsgs(self):
		# Test to make sure that various bad messages in the
		# handshake process are rejected even if the attacker
		# has the correct key.  This just keeps the protocol
		# tight allowing for variations in the future.

		# seed the RNG
		prngseed = b'abc123'
		from ctypes import c_uint8
		syote_comms.strobe_seed_prng((c_uint8 *
		    len(prngseed))(*prngseed), len(prngseed))

		# Create the state for testing
		commstate = syote_comms.CommsState()

		cb = syote_comms.process_msgfunc_t(lambda msg, outbuf: None)

		# Generate shared key
		shared_key = os.urandom(32)

		# Initialize everything
		syote_comms.comms_init(commstate, cb, make_pktbuf(shared_key), None, None)

		# Create test fixture, only use it to init crypto state
		tsd = SyncDatagram()
		l = LORANode(tsd, shared=shared_key)

		# copy the crypto state
		cstate = l.st.copy()

		# compose an incorrect init message
		msg = os.urandom(16) + b'othre'
		msg = cstate.send_enc(msg) + cstate.send_mac(l.MAC_LEN)

		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 = syote_comms.comms_process_wrap(commstate, b'0' * i)

			self.assertFalse(out)

		# copy the crypto state
		cstate = l.st.copy()

		# compose an incorrect init message
		msg = os.urandom(16) + b' eqreset'
		msg = cstate.send_enc(msg) + cstate.send_mac(l.MAC_LEN)

		out = syote_comms.comms_process_wrap(commstate, msg)

		self.assertFalse(out)

		# compose the correct init message
		msg = os.urandom(16) + b'reqreset'
		msg = l.st.send_enc(msg) + l.st.send_mac(l.MAC_LEN)

		out = syote_comms.comms_process_wrap(commstate, msg)

		l.st.recv_enc(out[:-l.MAC_LEN])
		l.st.recv_mac(out[-l.MAC_LEN:])

		l.st.ratchet()

		# copy the crypto state
		cstate = l.st.copy()

		# compose an incorrect confirmed message
		msg = b'onfirm'
		msg = cstate.send_enc(msg) + cstate.send_mac(l.MAC_LEN)

		out = syote_comms.comms_process_wrap(commstate, msg)

		self.assertFalse(out)

		# copy the crypto state
		cstate = l.st.copy()

		# compose an incorrect confirmed message
		msg = b' onfirm'
		msg = cstate.send_enc(msg) + cstate.send_mac(l.MAC_LEN)

		out = syote_comms.comms_process_wrap(commstate, msg)

		self.assertFalse(out)

	@timeout(2)
	async def test_ccode_ecdhe(self):
		_self = self
		from ctypes import c_uint8

		# seed the RNG
		prngseed = b'abc123'
		syote_comms.strobe_seed_prng((c_uint8 *
		    len(prngseed))(*prngseed), len(prngseed))

		# Create the state for testing
		commstate = syote_comms.CommsState()

		# These are the expected messages and their arguments
		exptmsgs = [
			(CMD_WAITFOR, [ 30 ]),
			(CMD_RUNFOR, [ 1, 50 ]),
			(CMD_PING, [ ]),
			(CMD_TERMINATE, [ ]),
		]
		def procmsg(msg, outbuf):
			msgbuf = msg._from()
			cmd = msgbuf[0]
			args = [ int.from_bytes(msgbuf[x:x + 4],
			    byteorder='little') for x in range(1, len(msgbuf),
			    4) ]

			if exptmsgs[0] == (cmd, args):
				exptmsgs.pop(0)
				outbuf[0].pkt[0] = cmd
				outbuf[0].pktlen = 1
			else: #pragma: no cover
				raise RuntimeError('cmd not found')

		# wrap the callback function
		cb = syote_comms.process_msgfunc_t(procmsg)

		class CCodeSD(MockSyncDatagram):
			async def runner(self):
				for expectlen in [ 40, 17, 9, 9, 9, 9 ]:
					# get message
					inmsg = await self.get()

					# process the test message
					out = syote_comms.comms_process_wrap(
					    commstate, inmsg)

					# make sure the reply matches length
					_self.assertEqual(expectlen, len(out))

					# save what was originally replied
					origmsg = out

					# pretend that the reply didn't make it
					out = syote_comms.comms_process_wrap(
					    commstate, inmsg)

					# make sure that the reply matches
					# the previous
					_self.assertEqual(origmsg, out)

					# pass the reply back
					await self.put(out)

		# Generate keys
		initkey = X25519.gen()
		respkey = X25519.gen()

		# Initialize everything
		syote_comms.comms_init(commstate, cb, None, make_pktbuf(respkey.getpriv()), make_pktbuf(initkey.getpub()))

		# Create test fixture
		tsd = CCodeSD()
		l = LORANode(tsd, init_key=initkey, resp_pub=respkey.getpub())

		# Send various messages
		await l.start()

		await l.waitfor(30)

		await l.runfor(1, 50)

		await l.ping()

		await l.terminate()

		await tsd.drain()

		# Make sure all messages have been processed
		self.assertTrue(tsd.sendq.empty())
		self.assertTrue(tsd.recvq.empty())

		# Make sure all the expected messages have been
		# processed.
		self.assertFalse(exptmsgs)
		#_debprint('done')

	@timeout(2)
	async def test_ccode(self):
		_self = self
		from ctypes import c_uint8, memmove

		# seed the RNG
		prngseed = b'abc123'
		syote_comms.strobe_seed_prng((c_uint8 *
		    len(prngseed))(*prngseed), len(prngseed))

		# Create the state for testing
		commstate = syote_comms.CommsState()

		# These are the expected messages and their arguments
		exptmsgs = [
			(CMD_WAITFOR, [ 30 ]),
			(CMD_RUNFOR, [ 1, 50 ]),
			(CMD_PING, [ ]),
			# using big here, because Python is stupid.
			(CMD_KEY, b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x16\x00\x00\x00\x00\x00\x02\x00\x0b\x00\x00\x00\x00\x00\x00\x00\x17\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00', CMD_KEY.to_bytes(1, byteorder='big') + (3).to_bytes(4, byteorder='little')),
			(CMD_KEY, b'\x00\x00\x17\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00', CMD_KEY.to_bytes(1, byteorder='big') + (2).to_bytes(4, byteorder='little')),
			(CMD_TERMINATE, [ ]),
		]
		def procmsg(msg, outbuf):
			msgbuf = msg._from()
			cmd = msgbuf[0]
			if isinstance(exptmsgs[0][1], bytes):
				args = msgbuf[1:]
			else:
				args = [ int.from_bytes(msgbuf[x:x + 4],
				    byteorder='little') for x in range(1, len(msgbuf),
				    4) ]

			if exptmsgs[0][:2] == (cmd, args):
				if len(exptmsgs[0]) > 2:
					rmsg = exptmsgs[0][2]
					memmove(outbuf[0].pkt, rmsg, len(rmsg))
					outbuf[0].pktlen = len(rmsg)
				else:
					outbuf[0].pkt[0] = cmd
					outbuf[0].pktlen = 1

				exptmsgs.pop(0)
			else: #pragma: no cover
				raise RuntimeError('cmd not found, got %d, expected %d' % (cmd, exptmsgs[0][0]))

		# wrap the callback function
		cb = syote_comms.process_msgfunc_t(procmsg)

		class CCodeSD(MockSyncDatagram):
			async def runner(self):
				for expectlen in [ 24, 17, 9, 9, 9, 13, 13, 9 ]:
					# get message
					inmsg = await self.get()

					# process the test message
					out = syote_comms.comms_process_wrap(
					    commstate, inmsg)

					# make sure the reply matches length
					_self.assertEqual(expectlen, len(out))

					# save what was originally replied
					origmsg = out

					# pretend that the reply didn't make it
					out = syote_comms.comms_process_wrap(
					    commstate, inmsg)

					# make sure that the reply matches
					# the previous
					_self.assertEqual(origmsg, out)

					# pass the reply back
					await self.put(out)

		# Generate shared key
		shared_key = os.urandom(32)

		# Initialize everything
		syote_comms.comms_init(commstate, cb, make_pktbuf(shared_key), None, None)

		# Create test fixture
		tsd = CCodeSD()
		l = LORANode(tsd, shared=shared_key)

		# Send various messages
		await l.start()

		await l.waitfor(30)

		await l.runfor(1, 50)

		await l.ping()

		await l.type('sHt')

		await l.terminate()

		await tsd.drain()

		# Make sure all messages have been processed
		self.assertTrue(tsd.sendq.empty())
		self.assertTrue(tsd.recvq.empty())

		# Make sure all the expected messages have been
		# processed.
		self.assertFalse(exptmsgs)
		#_debprint('done')

	@timeout(2)
	async def test_ccode_newsession(self):
		'''This test is to make sure that if an existing session
		is running, that a new session can be established, and that
		when it does, the old session becomes inactive.
		'''

		_self = self
		from ctypes import c_uint8

		seq = AsyncSequence()

		# seed the RNG
		prngseed = b'abc123'
		syote_comms.strobe_seed_prng((c_uint8 *
		    len(prngseed))(*prngseed), len(prngseed))

		# Create the state for testing
		commstate = syote_comms.CommsState()

		# These are the expected messages and their arguments
		exptmsgs = [
			(CMD_WAITFOR, [ 30 ]),
			(CMD_WAITFOR, [ 70 ]),
			(CMD_WAITFOR, [ 40 ]),
			(CMD_TERMINATE, [ ]),
		]
		def procmsg(msg, outbuf):
			msgbuf = msg._from()
			cmd = msgbuf[0]
			args = [ int.from_bytes(msgbuf[x:x + 4],
			    byteorder='little') for x in range(1, len(msgbuf),
			    4) ]

			if exptmsgs[0] == (cmd, args):
				exptmsgs.pop(0)
				outbuf[0].pkt[0] = cmd
				outbuf[0].pktlen = 1
			else: #pragma: no cover
				raise RuntimeError('cmd not found: %d' % cmd)

		# wrap the callback function
		cb = syote_comms.process_msgfunc_t(procmsg)

		class FlipMsg(object):
			async def flipmsg(self):
				# get message
				inmsg = await self.get()

				# process the test message
				out = syote_comms.comms_process_wrap(
				    commstate, inmsg)

				# pass the reply back
				await self.put(out)

		# this class always passes messages, this is
		# used for the first session.
		class CCodeSD1(MockSyncDatagram, FlipMsg):
			async def runner(self):
				for i in range(3):
					await self.flipmsg()

				async with seq.sync(0):
					# create bogus message
					inmsg = b'0'*24

					# process the bogus message
					out = syote_comms.comms_process_wrap(
					    commstate, inmsg)

					# make sure there was not a response
					_self.assertFalse(out)

				await self.flipmsg()

		# this one is special in that it will pause after the first
		# message to ensure that the previous session will continue
		# to work, AND that if a new "new" session comes along, it
		# will override the previous new session that hasn't been
		# confirmed yet.
		class CCodeSD2(MockSyncDatagram, FlipMsg):
			async def runner(self):
				# pass one message from the new session
				async with seq.sync(1):
					# There might be a missing case
					# handled for when the confirmed
					# message is generated, but lost.
					await self.flipmsg()

					# and the old session is still active
					await l.waitfor(70)

				async with seq.sync(2):
					for i in range(3):
						await self.flipmsg()

		# Generate shared key
		shared_key = os.urandom(32)

		# Initialize everything
		syote_comms.comms_init(commstate, cb, make_pktbuf(shared_key), None, None)

		# Create test fixture
		tsd = CCodeSD1()
		l = LORANode(tsd, shared=shared_key)

		# Send various messages
		await l.start()

		await l.waitfor(30)

		# Ensure that a new one can take over
		tsd2 = CCodeSD2()

		l2 = LORANode(tsd2, shared=shared_key)

		# Send various messages
		await l2.start()

		await l2.waitfor(40)

		await l2.terminate()

		await tsd.drain()
		await tsd2.drain()

		# Make sure all messages have been processed
		self.assertTrue(tsd.sendq.empty())
		self.assertTrue(tsd.recvq.empty())
		self.assertTrue(tsd2.sendq.empty())
		self.assertTrue(tsd2.recvq.empty())

		# Make sure all the expected messages have been
		# processed.
		self.assertFalse(exptmsgs)

class TestLoRaNodeMulticast(unittest.IsolatedAsyncioTestCase):
	# see: https://www.iana.org/assignments/multicast-addresses/multicast-addresses.xhtml#multicast-addresses-1
	maddr = ('224.0.0.198', 48542)

	@timeout(2)
	async def test_multisyncdgram(self):
		# Test the implementation of the multicast version of
		# SyncDatagram

		_self = self
		from ctypes import c_uint8

		# seed the RNG
		prngseed = b'abc123'
		syote_comms.strobe_seed_prng((c_uint8 *
		    len(prngseed))(*prngseed), len(prngseed))

		# Create the state for testing
		commstate = syote_comms.CommsState()

		# These are the expected messages and their arguments
		exptmsgs = [
			(CMD_WAITFOR, [ 30 ]),
			(CMD_PING, [ ]),
			(CMD_TERMINATE, [ ]),
		]
		def procmsg(msg, outbuf):
			msgbuf = msg._from()
			cmd = msgbuf[0]
			args = [ int.from_bytes(msgbuf[x:x + 4],
			    byteorder='little') for x in range(1, len(msgbuf),
			    4) ]

			if exptmsgs[0] == (cmd, args):
				exptmsgs.pop(0)
				outbuf[0].pkt[0] = cmd
				outbuf[0].pktlen = 1
			else: #pragma: no cover
				raise RuntimeError('cmd not found')

		# wrap the callback function
		cb = syote_comms.process_msgfunc_t(procmsg)

		# Generate shared key
		shared_key = os.urandom(32)

		# Initialize everything
		syote_comms.comms_init(commstate, cb, make_pktbuf(shared_key), None, None)

		# create the object we are testing
		msd = MulticastSyncDatagram(self.maddr)

		seq = AsyncSequence()

		async def clienttask():
			mr = await multicast.create_multicast_receiver(
			    self.maddr)
			mt = await multicast.create_multicast_transmitter(
			    self.maddr)

			try:
				# make sure the above threads are running
				await seq.simpsync(0)

				while True:
					pkt = await mr.recv()
					msg = pkt[0]

					out = syote_comms.comms_process_wrap(
					    commstate, msg)

					if out:
						await mt.send(out)
			finally:
				mr.close()
				mt.close()

		task = asyncio.create_task(clienttask())

		# start it
		await msd.start()

		# pass it to a node
		l = LORANode(msd, shared=shared_key)

		await seq.simpsync(1)

		# Send various messages
		await l.start()

		await l.waitfor(30)

		await l.ping()

		await l.terminate()

		# shut things down
		ln = None
		msd.close()

		task.cancel()

		with self.assertRaises(asyncio.CancelledError):
			await task

class TestLORAmain(unittest.TestCase):
	def setUp(self):
		# setup temporary directory
		d = pathlib.Path(tempfile.mkdtemp()).resolve()
		self.basetempdir = d
		self.tempdir = d / 'subdir'
		self.tempdir.mkdir()
		self.origcwd = os.getcwd()

		os.chdir(pathlib.Path(__file__).parent)

	def tearDown(self):
		btd = self.basetempdir
		self.tempdir = None
		self.basetempdir = None
		shutil.rmtree(btd)

		os.chdir(self.origcwd)
		self.origcwd = None

	def test_sharedkey(self):
		keys = []

		# shared key test
		keyfile = self.basetempdir / 'shared.key'
		with keyfile.open('w') as fp:
			fp.write(os.urandom(16).hex())

		keys.append(('shared', [ [ '-s', keyfile ] ] * 2))

		# pub key test
		init_key = X25519.gen()
		resp_key = X25519.gen()

		f = {}
		for i in [ 'init', 'resp' ]:
			key = locals()['%s_key' % i]

			for j in [ 'pub', 'priv' ]:
				data = getattr(key, 'get%s' % j)().hex()
				file = self.basetempdir / ('%s_%s.key' % (i, j))
				f['%s_%s_key_file' % (i, j)] = file

				with file.open('w') as fp:
					fp.write(data)

		keys.append(('pk', (
		    [ '-k', f['init_priv_key_file'],
			'-p', f['resp_pub_key_file'] ],
		    [ '-k', f['resp_priv_key_file'],
			'-p', f['init_pub_key_file'] ]
		)))
		for name, (initargs, respargs) in keys:
			# XXX - macosx specific library path var
			with self.subTest(name=name), patch.dict(os.environ,
			    dict(DYLD_LIBRARY_PATH=self.origcwd)):
				self.run_program_keyargs(initargs, respargs)

	def run_program_keyargs(self, initargs, respargs):
		try:
			resp = subprocess.Popen([ 'python', __file__,
			    '-r', 'dummy:TestAttr.dummy_func' ] +
			    respargs, stdin=DEVNULL, stderr=PIPE,
			    stdout=PIPE)

			init = subprocess.Popen([ 'python', __file__ ] +
			    initargs + [ 'ping' ], stdin=DEVNULL, stderr=PIPE,
			    stdout=DEVNULL)

			# make sure the command completes
			initret = init.wait(2)
			self.assertEqual(initret, 0)

			# terminate responder
			respret = resp.terminate()
			stdout, errout = resp.communicate()

			# make sure we got the message
			self.assertEqual(stdout, b"got msg: b'\\x04'\n")

			self.assertEqual(errout, b'')

		finally:
			init.kill()
			_, errout = init.communicate()
			#print(repr(errout))
			#sys.stdout.flush()
			self.assertEqual(errout, b'')

			resp.kill()
			_, errout = resp.communicate()
			self.assertEqual(errout, b'')