Flow control for KISS interfaces, Transport Identity implemented

RNS/Identity.py

@@ -26,18 +26,6 @@ class Identity:
 	# Storage
 	known_destinations = {}
 
-	def __init__(self,public_only=False):
-		# Initialize keys to none
-		self.prv = None
-		self.pub = None
-		self.prv_bytes = None
-		self.pub_bytes = None
-		self.hash = None
-		self.hexhash = None
-
-		if not public_only:
-			self.createKeys()
-
 	@staticmethod
 	def remember(packet_hash, destination_hash, public_key, app_data = None):
 		RNS.log("Remembering "+RNS.prettyhexrep(destination_hash), RNS.LOG_VERBOSE)
@@ -126,6 +114,27 @@ class Identity:
 		Identity.saveKnownDestinations()
 
 
+	@staticmethod
+	def from_file(path):
+		identity = Identity(public_only=True)
+		if identity.load(path):
+			return identity
+		else:
+			return None
+
+
+	def __init__(self,public_only=False):
+		# Initialize keys to none
+		self.prv = None
+		self.pub = None
+		self.prv_bytes = None
+		self.pub_bytes = None
+		self.hash = None
+		self.hexhash = None
+
+		if not public_only:
+			self.createKeys()
+
 	def createKeys(self):
 		self.prv = rsa.generate_private_key(
 			public_exponent=65337,
@@ -145,7 +154,7 @@ class Identity:
 
 		self.updateHashes()
 
-		RNS.log("Identity keys created for "+RNS.prettyhexrep(self.hash), RNS.LOG_INFO)
+		RNS.log("Identity keys created for "+RNS.prettyhexrep(self.hash), RNS.LOG_VERBOSE)
 
 	def getPrivateKey(self):
 		return self.prv_bytes
@@ -153,9 +162,14 @@ class Identity:
 	def getPublicKey(self):
 		return self.pub_bytes
 
-	def loadPrivateKey(self, key):
+	def loadPrivateKey(self, prv_bytes):
-		self.prv_bytes = key
+		try:
-		self.prv = serialization.load_der_private_key(self.prv_bytes, password=None,backend=default_backend())
+			self.prv_bytes = prv_bytes
+			self.prv = serialization.load_der_private_key(
+				self.prv_bytes,
+				password=None,
+				backend=default_backend()
+			)
 			self.pub = self.prv.public_key()
 			self.pub_bytes = self.pub.public_bytes(
 				encoding=serialization.Encoding.DER,
@@ -163,6 +177,13 @@ class Identity:
 			)
 			self.updateHashes()
 
+			return True
+
+		except Exception as e:
+			RNS.log("Failed to load identity key", RNS.LOG_ERROR)
+			RNS.log("The contained exception was: "+str(e))
+			return False
+
 	def loadPublicKey(self, key):
 		try:
 			self.pub_bytes = key
@@ -175,11 +196,25 @@ class Identity:
 		self.hash = Identity.truncatedHash(self.pub_bytes)
 		self.hexhash = self.hash.encode("hex_codec")
 
-	def saveIdentity(self):
+	def save(self, path):
-		pass
+		try:
+			with open(path, "w") as key_file:
+				key_file.write(self.prv_bytes)
+				return True
+			return False
+		except Exception as e:
+			RNS.log("Error while saving identity to "+str(path), RNS.LOG_ERROR)
+			RNS.log("The contained exception was: "+str(e))
 
-	def loadIdentity(self):
+	def load(self, path):
-		pass
+		try:
+			with open(path, "r") as key_file:
+				prv_bytes = key_file.read()
+				return self.loadPrivateKey(prv_bytes)
+			return False
+		except Exception as e:
+			RNS.log("Error while loading identity from "+str(path), RNS.LOG_ERROR)
+			RNS.log("The contained exception was: "+str(e))
 
 	def encrypt(self, plaintext):
 		if self.pub != None:
@@ -280,3 +315,6 @@ class Identity:
 
 		proof = RNS.Packet(destination, proof_data, RNS.Packet.PROOF)
 		proof.send()
+
+	def __str__(self):
+		return RNS.prettyhexrep(self.hash)

RNS/Interfaces/AX25KISSInterface.py

@@ -20,6 +20,7 @@ class KISS():
 	CMD_TXTAIL		  = chr(0x04)
 	CMD_FULLDUPLEX	  = chr(0x05)
 	CMD_SETHARDWARE	  = chr(0x06)
+	CMD_READY         = chr(0x0F)
 	CMD_RETURN		  = chr(0xFF)
 
 class AX25():
@@ -39,7 +40,7 @@ class AX25KISSInterface(Interface):
 	stopbits = None
 	serial   = None
 
-	def __init__(self, owner, name, callsign, ssid, port, speed, databits, parity, stopbits, preamble, txtail, persistence, slottime):
+	def __init__(self, owner, name, callsign, ssid, port, speed, databits, parity, stopbits, preamble, txtail, persistence, slottime, flow_control):
 		self.serial   = None
 		self.owner    = owner
 		self.name	  = name
@@ -55,6 +56,10 @@ class AX25KISSInterface(Interface):
 		self.timeout  = 100
 		self.online   = False
 
+		self.packet_queue    = []
+		self.flow_control    = flow_control
+		self.interface_ready = False
+
 		if (len(self.src_call) < 3 or len(self.src_call) > 6):
 			raise ValueError("Invalid callsign for "+str(self))
 
@@ -104,6 +109,8 @@ class AX25KISSInterface(Interface):
 			self.setTxTail(self.txtail)
 			self.setPersistence(self.persistence)
 			self.setSlotTime(self.slottime)
+			self.setFlowControl(self.flow_control)
+			self.interface_ready = True
 			RNS.log("AX.25 KISS interface configured")
 			sleep(2)
 		else:
@@ -160,6 +167,15 @@ class AX25KISSInterface(Interface):
 		if written != len(kiss_command):
 			raise IOError("Could not configure AX.25 KISS interface slot time to "+str(slottime_ms)+" (command value "+str(slottime)+")")
 
+	def setFlowControl(self, flow_control):
+		kiss_command = KISS.FEND+KISS.CMD_READY+chr(0x01)+KISS.FEND
+		written = self.serial.write(kiss_command)
+		if written != len(kiss_command):
+			if (flow_control):
+				raise IOError("Could not enable AX.25 KISS interface flow control")
+			else:
+				raise IOError("Could not enable AX.25 KISS interface flow control")
+
 
 	def processIncoming(self, data):
 		if (len(data) > 16):
@@ -168,7 +184,9 @@ class AX25KISSInterface(Interface):
 
 	def processOutgoing(self,data):
 		if self.online:
-			# gen src/dst
+			if self.interface_ready:
+				if self.flow_control:
+					self.interface_ready = False
 
 				encoded_dst_ssid = 0x60 | (self.dst_ssid << 1)
 				encoded_src_ssid = 0x60 | (self.src_ssid << 1) | 0x01
@@ -197,8 +215,22 @@ class AX25KISSInterface(Interface):
 
 				written = self.serial.write(kiss_frame)
 				if written != len(kiss_frame):
+					if self.flow_control:
+						self.interface_ready = True
 					raise IOError("AX.25 interface only wrote "+str(written)+" bytes of "+str(len(kiss_frame)))
+			else:
+				self.queue(data)
 
+	def queue(self, data):
+		self.packet_queue.append(data)
+
+	def process_queue(self):
+		if len(self.packet_queue) > 0:
+			data = self.packet_queue.pop(0)
+			self.interface_ready = True
+			self.processOutgoing(data)
+		elif len(self.packet_queue) == 0:
+			self.interface_ready = True
 
 	def readLoop(self):
 		try:
@@ -237,6 +269,10 @@ class AX25KISSInterface(Interface):
 										byte = KISS.FESC
 									escape = False
 								data_buffer = data_buffer+byte
+						elif (command == KISS.CMD_READY):
+							# TODO: add timeout and reset if ready
+							# command never arrives
+							self.process_queue()
 				else:
 					time_since_last = int(time.time()*1000) - last_read_ms
 					if len(data_buffer) > 0 and time_since_last > self.timeout:

RNS/Interfaces/KISSInterface.py

@@ -20,6 +20,7 @@ class KISS():
 	CMD_TXTAIL		= chr(0x04)
 	CMD_FULLDUPLEX	= chr(0x05)
 	CMD_SETHARDWARE	= chr(0x06)
+	CMD_READY       = chr(0x0F)
 	CMD_RETURN		= chr(0xFF)
 
 class KISSInterface(Interface):
@@ -33,7 +34,7 @@ class KISSInterface(Interface):
 	stopbits = None
 	serial   = None
 
-	def __init__(self, owner, name, port, speed, databits, parity, stopbits, preamble, txtail, persistence, slottime):
+	def __init__(self, owner, name, port, speed, databits, parity, stopbits, preamble, txtail, persistence, slottime, flow_control):
 		self.serial   = None
 		self.owner    = owner
 		self.name     = name
@@ -45,6 +46,10 @@ class KISSInterface(Interface):
 		self.timeout  = 100
 		self.online   = False
 
+		self.packet_queue    = []
+		self.flow_control = flow_control
+		self.interface_ready = False
+
 		self.preamble    = preamble if preamble != None else 350;
 		self.txtail      = txtail if txtail != None else 20;
 		self.persistence = persistence if persistence != None else 64;
@@ -88,6 +93,8 @@ class KISSInterface(Interface):
 			self.setTxTail(self.txtail)
 			self.setPersistence(self.persistence)
 			self.setSlotTime(self.slottime)
+			self.setFlowControl(self.flow_control)
+			self.interface_ready = True
 			RNS.log("KISS interface configured")
 		else:
 			raise IOError("Could not open serial port")
@@ -144,6 +151,15 @@ class KISSInterface(Interface):
 		if written != len(kiss_command):
 			raise IOError("Could not configure KISS interface slot time to "+str(slottime_ms)+" (command value "+str(slottime)+")")
 
+		def setFlowControl(self, flow_control):
+		kiss_command = KISS.FEND+KISS.CMD_READY+chr(0x01)+KISS.FEND
+		written = self.serial.write(kiss_command)
+		if written != len(kiss_command):
+			if (flow_control):
+				raise IOError("Could not enable KISS interface flow control")
+			else:
+				raise IOError("Could not enable KISS interface flow control")
+
 
 	def processIncoming(self, data):
 		self.owner.inbound(data, self)
@@ -151,6 +167,10 @@ class KISSInterface(Interface):
 
 	def processOutgoing(self,data):
 		if self.online:
+			if self.interface_ready:
+				if self.flow_control:
+					self.interface_ready = False
+
 				data = data.replace(chr(0xdb), chr(0xdb)+chr(0xdd))
 				data = data.replace(chr(0xc0), chr(0xdb)+chr(0xdc))
 				frame = chr(0xc0)+chr(0x00)+data+chr(0xc0)
@@ -158,6 +178,19 @@ class KISSInterface(Interface):
 				if written != len(frame):
 					raise IOError("Serial interface only wrote "+str(written)+" bytes of "+str(len(data)))
 
+			else:
+				self.queue(data)
+
+	def queue(self, data):
+		self.packet_queue.append(data)
+
+	def process_queue(self):
+		if len(self.packet_queue) > 0:
+			data = self.packet_queue.pop(0)
+			self.interface_ready = True
+			self.processOutgoing(data)
+		elif len(self.packet_queue) == 0:
+			self.interface_ready = True
 
 	def readLoop(self):
 		try:
@@ -196,6 +229,10 @@ class KISSInterface(Interface):
 										byte = KISS.FESC
 									escape = False
 								data_buffer = data_buffer+byte
+						elif (command == KISS.CMD_READY):
+							# TODO: add timeout and reset if ready
+							# command never arrives
+							self.process_queue()
 				else:
 					time_since_last = int(time.time()*1000) - last_read_ms
 					if len(data_buffer) > 0 and time_since_last > self.timeout:

RNS/Reticulum.py

@@ -51,7 +51,7 @@ class Reticulum:
 		RNS.Identity.loadKnownDestinations()
 		Reticulum.router = self
 
-		RNS.Transport.scheduleJobs()
+		RNS.Transport.start()
 
 		atexit.register(RNS.Identity.exitHandler)
 
@@ -141,6 +141,7 @@ class Reticulum:
 					txtail = int(c["txtail"]) if "txtail" in c else None
 					persistence = int(c["persistence"]) if "persistence" in c else None
 					slottime = int(c["slottime"]) if "slottime" in c else None
+					flow_control = (True if c["flow_control"] == "true" else False) if "flow_control" in c else False
 
 					port = c["port"] if "port" in c else None
 					speed = int(c["speed"]) if "speed" in c else 9600
@@ -162,7 +163,8 @@ class Reticulum:
 						preamble,
 						txtail,
 						persistence,
-						slottime
+						slottime,
+						flow_control
 					)
 
 					if "outgoing" in c and c["outgoing"].lower() == "true":
@@ -177,6 +179,7 @@ class Reticulum:
 					txtail = int(c["txtail"]) if "txtail" in c else None
 					persistence = int(c["persistence"]) if "persistence" in c else None
 					slottime = int(c["slottime"]) if "slottime" in c else None
+					flow_control = (True if c["flow_control"] == "true" else False) if "flow_control" in c else False
 
 					port = c["port"] if "port" in c else None
 					speed = int(c["speed"]) if "speed" in c else 9600
@@ -203,7 +206,8 @@ class Reticulum:
 						preamble,
 						txtail,
 						persistence,
-						slottime
+						slottime,
+						flow_control
 					)
 
 					if "outgoing" in c and c["outgoing"].lower() == "true":

RNS/Transport.py

@@ -26,12 +26,29 @@ class Transport:
 	receipts_check_interval = 1.0
 	hashlist_maxsize        = 1000000
 
+	identity = None
+
 	@staticmethod
-	def scheduleJobs():
+	def start():
+		if Transport.identity == None:
+			transport_identity_path = RNS.Reticulum.configdir+"/transportidentity"
+			if os.path.isfile(transport_identity_path):
+				Transport.identity = RNS.Identity.from_file(transport_identity_path)				
+
+			if Transport.identity == None:
+				RNS.log("No valid Transport Identity on disk, creating...", RNS.LOG_VERBOSE)
+				Transport.identity = RNS.Identity()
+				Transport.identity.save(transport_identity_path)
+			else:
+				RNS.log("Loaded Transport Identity from disk", RNS.LOG_VERBOSE)
+
+
 		thread = threading.Thread(target=Transport.jobloop)
 		thread.setDaemon(True)
 		thread.start()
 
+		RNS.log("Transport instance "+str(Transport.identity)+" started")
+
 	@staticmethod
 	def jobloop():
 		while (True):