mirror of
https://github.com/markqvist/Reticulum.git
synced 2026-07-22 07:58:10 -07:00
Compare commits
15 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| be36abd857 | |||
| fa4d4c6716 | |||
| a5728be421 | |||
| 032b1aa3b2 | |||
| fb7479a6f6 | |||
| f81b267509 | |||
| 6527423526 | |||
| d8bc20d4cb | |||
| 6c6238ce29 | |||
| cbba3502b7 | |||
| 2b79db03c0 | |||
| 2c9edc43f1 | |||
| ef9244f79f | |||
| 2d811dc398 | |||
| e64d815033 |
+42
-15
@@ -8,6 +8,7 @@ import ipaddress
|
|||||||
import subprocess
|
import subprocess
|
||||||
from threading import Lock
|
from threading import Lock
|
||||||
from .vendor import umsgpack as msgpack
|
from .vendor import umsgpack as msgpack
|
||||||
|
from collections import deque
|
||||||
|
|
||||||
NAME = 0xFF
|
NAME = 0xFF
|
||||||
TRANSPORT_ID = 0xFE
|
TRANSPORT_ID = 0xFE
|
||||||
@@ -31,7 +32,7 @@ APP_NAME = "rnstransport"
|
|||||||
|
|
||||||
class InterfaceAnnouncer():
|
class InterfaceAnnouncer():
|
||||||
JOB_INTERVAL = 60
|
JOB_INTERVAL = 60
|
||||||
DEFAULT_STAMP_VALUE = 14
|
DEFAULT_STAMP_VALUE = 16
|
||||||
WORKBLOCK_EXPAND_ROUNDS = 20
|
WORKBLOCK_EXPAND_ROUNDS = 20
|
||||||
|
|
||||||
DISCOVERABLE_INTERFACE_TYPES = ["BackboneInterface", "TCPServerInterface", "TCPClientInterface",
|
DISCOVERABLE_INTERFACE_TYPES = ["BackboneInterface", "TCPServerInterface", "TCPClientInterface",
|
||||||
@@ -224,10 +225,14 @@ class InterfaceAnnounceHandler:
|
|||||||
RNS.log("You can install it with the command: pip install lxmf", RNS.LOG_CRITICAL)
|
RNS.log("You can install it with the command: pip install lxmf", RNS.LOG_CRITICAL)
|
||||||
RNS.panic()
|
RNS.panic()
|
||||||
|
|
||||||
self.aspect_filter = APP_NAME+".discovery.interface"
|
self.aspect_filter = APP_NAME+".discovery.interface"
|
||||||
self.required_value = required_value
|
self.required_value = required_value
|
||||||
self.callback = callback
|
self.callback = callback
|
||||||
self.stamper = LXStamper
|
self.stamper = LXStamper
|
||||||
|
self.valid_cache = {}
|
||||||
|
self.valid_cache_max = 2048
|
||||||
|
self.invalid_cache = deque(maxlen=2048)
|
||||||
|
self.validation_lock = threading.Lock()
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def sanitize_name(name):
|
def sanitize_name(name):
|
||||||
@@ -250,21 +255,43 @@ class InterfaceAnnounceHandler:
|
|||||||
app_data = app_data[1:]
|
app_data = app_data[1:]
|
||||||
signed = flags & self.FLAG_SIGNED
|
signed = flags & self.FLAG_SIGNED
|
||||||
encrypted = flags & self.FLAG_ENCRYPTED
|
encrypted = flags & self.FLAG_ENCRYPTED
|
||||||
|
fullhash = RNS.Identity.full_hash(app_data)
|
||||||
|
|
||||||
if encrypted:
|
if fullhash in self.invalid_cache:
|
||||||
if not RNS.Transport.has_network_identity(): return
|
RNS.log(f"Ignored previously discovered interface with insufficient stamp value", RNS.LOG_PATHING) if RNS.sl(RNS.LOG_PATHING) else None
|
||||||
app_data = RNS.Transport.network_identity.decrypt(app_data)
|
return
|
||||||
if not app_data: return
|
|
||||||
|
|
||||||
stamp = app_data[-self.stamper.STAMP_SIZE:]
|
if fullhash in self.valid_cache:
|
||||||
packed = app_data[:-self.stamper.STAMP_SIZE]
|
RNS.log(f"Discovery announce cache hit", RNS.LOG_PATHING) if RNS.sl(RNS.LOG_PATHING) else None
|
||||||
infohash = RNS.Identity.full_hash(packed)
|
cache_entry = self.valid_cache[fullhash]
|
||||||
workblock = self.stamper.stamp_workblock(infohash, expand_rounds=InterfaceAnnouncer.WORKBLOCK_EXPAND_ROUNDS)
|
stamp = cache_entry["stamp"]
|
||||||
value = self.stamper.stamp_value(workblock, stamp)
|
packed = cache_entry["packed"]
|
||||||
valid = self.stamper.stamp_valid(stamp, self.required_value, workblock)
|
value = cache_entry["value"]
|
||||||
|
valid = cache_entry["valid"]
|
||||||
|
|
||||||
|
else:
|
||||||
|
if encrypted:
|
||||||
|
if not RNS.Transport.has_network_identity(): return
|
||||||
|
app_data = RNS.Transport.network_identity.decrypt(app_data)
|
||||||
|
if not app_data: return
|
||||||
|
|
||||||
|
if self.validation_lock.locked():
|
||||||
|
RNS.log(f"Dropping received interface discovery announce, already validating other discovery stamp", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
|
||||||
|
return
|
||||||
|
|
||||||
|
with self.validation_lock:
|
||||||
|
stamp = app_data[-self.stamper.STAMP_SIZE:]
|
||||||
|
packed = app_data[:-self.stamper.STAMP_SIZE]
|
||||||
|
infohash = RNS.Identity.full_hash(packed)
|
||||||
|
workblock = self.stamper.stamp_workblock(infohash, expand_rounds=InterfaceAnnouncer.WORKBLOCK_EXPAND_ROUNDS)
|
||||||
|
value = self.stamper.stamp_value(workblock, stamp)
|
||||||
|
valid = self.stamper.stamp_valid(stamp, self.required_value, workblock)
|
||||||
|
self.valid_cache[fullhash] = {"valid": valid, "value": value, "packed": packed, "stamp": stamp}
|
||||||
|
while len(self.valid_cache) > self.valid_cache_max: self.valid_cache.pop(next(iter(self.valid_cache)))
|
||||||
|
|
||||||
if not valid:
|
if not valid:
|
||||||
RNS.log(f"Ignored discovered interface with insufficient stamp value {value}", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
|
RNS.log(f"Ignored discovered interface with insufficient stamp value {value}", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
|
||||||
|
self.invalid_cache.append(fullhash)
|
||||||
return
|
return
|
||||||
|
|
||||||
if value < self.required_value: RNS.log(f"Ignored discovered interface with stamp value {value}", RNS.LOG_DEBUG)
|
if value < self.required_value: RNS.log(f"Ignored discovered interface with stamp value {value}", RNS.LOG_DEBUG)
|
||||||
|
|||||||
+13
-34
@@ -174,12 +174,9 @@ class Identity:
|
|||||||
else: return None
|
else: return None
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def save_known_destinations(background=False, recombine=True):
|
def save_known_destinations(background=False, recombine=False):
|
||||||
# TODO: Improve the storage method so we don't have to
|
if recombine: RNS.log(f"Recombining known destinations from disk cache on persist is deprecated, argument ignored", RNS.LOG_WARNING)
|
||||||
# deserialize and serialize the entire table on every
|
if RNS.Transport.owner.is_connected_to_shared_instance: return
|
||||||
# save, but the only changes. It might be possible to
|
|
||||||
# simply overwrite on exit now that every local client
|
|
||||||
# disconnect triggers a data persist.
|
|
||||||
|
|
||||||
try:
|
try:
|
||||||
if hasattr(Identity, "saving_known_destinations"):
|
if hasattr(Identity, "saving_known_destinations"):
|
||||||
@@ -194,28 +191,9 @@ class Identity:
|
|||||||
|
|
||||||
Identity.saving_known_destinations = True
|
Identity.saving_known_destinations = True
|
||||||
save_start = time.time()
|
save_start = time.time()
|
||||||
|
|
||||||
if recombine:
|
|
||||||
storage_known_destinations = {}
|
|
||||||
if os.path.isfile(RNS.Reticulum.storagepath+"/known_destinations"):
|
|
||||||
try:
|
|
||||||
with open(RNS.Reticulum.storagepath+"/known_destinations","rb") as file:
|
|
||||||
storage_known_destinations = umsgpack.load(file)
|
|
||||||
|
|
||||||
except: pass
|
|
||||||
|
|
||||||
try:
|
|
||||||
for destination_hash in storage_known_destinations:
|
|
||||||
if not destination_hash in Identity.known_destinations:
|
|
||||||
with Identity.known_destinations_lock:
|
|
||||||
Identity.known_destinations[destination_hash] = storage_known_destinations[destination_hash]
|
|
||||||
|
|
||||||
except Exception as e:
|
|
||||||
RNS.log("Skipped recombining known destinations from disk, since an error occurred: "+str(e), RNS.LOG_WARNING)
|
|
||||||
|
|
||||||
RNS.log("Saving "+str(len(Identity.known_destinations))+" known destinations to storage...", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
|
RNS.log("Saving "+str(len(Identity.known_destinations))+" known destinations to storage...", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
|
||||||
temp_file = RNS.Reticulum.storagepath+f"/known_destinations.tmp.{time.time()}"
|
|
||||||
|
|
||||||
|
temp_file = RNS.Reticulum.storagepath+f"/known_destinations.tmp.{time.time()}"
|
||||||
try:
|
try:
|
||||||
with open(temp_file,"wb") as file: umsgpack.dump(Identity.known_destinations.copy(), file)
|
with open(temp_file,"wb") as file: umsgpack.dump(Identity.known_destinations.copy(), file)
|
||||||
os.replace(temp_file, RNS.Reticulum.storagepath+f"/known_destinations")
|
os.replace(temp_file, RNS.Reticulum.storagepath+f"/known_destinations")
|
||||||
@@ -226,17 +204,13 @@ class Identity:
|
|||||||
except Exception as e: RNS.log(f"Could not clean up temporary file {temp_file}: {e}", RNS.LOG_WARNING)
|
except Exception as e: RNS.log(f"Could not clean up temporary file {temp_file}: {e}", RNS.LOG_WARNING)
|
||||||
raise e
|
raise e
|
||||||
|
|
||||||
save_time = time.time() - save_start
|
RNS.log(f"Saved known destinations to storage in {RNS.prettyshorttime(time.time()-save_start)}", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
|
||||||
if save_time < 1: time_str = str(round(save_time*1000,2))+"ms"
|
|
||||||
else: time_str = str(round(save_time,2))+"s"
|
|
||||||
|
|
||||||
RNS.log("Saved known destinations to storage in "+time_str, RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
|
|
||||||
|
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
RNS.log("Error while saving known destinations to disk, the contained exception was: "+str(e), RNS.LOG_ERROR)
|
RNS.log("Error while saving known destinations to disk, the contained exception was: "+str(e), RNS.LOG_ERROR)
|
||||||
RNS.trace_exception(e)
|
RNS.trace_exception(e)
|
||||||
|
|
||||||
Identity.saving_known_destinations = False
|
finally: Identity.saving_known_destinations = False
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def load_known_destinations():
|
def load_known_destinations():
|
||||||
@@ -308,7 +282,7 @@ class Identity:
|
|||||||
except Exception as e: RNS.log(f"Error while retaining identity {RNS.prettyhexrep(identity_hash)}: {e}", RNS.LOG_ERROR)
|
except Exception as e: RNS.log(f"Error while retaining identity {RNS.prettyhexrep(identity_hash)}: {e}", RNS.LOG_ERROR)
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def clean_known_destinations():
|
def clean_known_destinations(background=False):
|
||||||
now = time.time()
|
now = time.time()
|
||||||
st = now
|
st = now
|
||||||
total = len(Identity.known_destinations)
|
total = len(Identity.known_destinations)
|
||||||
@@ -318,9 +292,13 @@ class Identity:
|
|||||||
never_used = 0
|
never_used = 0
|
||||||
ratchetdir = RNS.Reticulum.storagepath+"/ratchets"
|
ratchetdir = RNS.Reticulum.storagepath+"/ratchets"
|
||||||
|
|
||||||
|
RNS.log(f"Cleaning known destinations{' at background priority' if background else ''}...", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
|
||||||
|
|
||||||
with Identity.known_destinations_lock: destination_hashes = list(Identity.known_destinations.keys())
|
with Identity.known_destinations_lock: destination_hashes = list(Identity.known_destinations.keys())
|
||||||
for destination_hash in destination_hashes:
|
for destination_hash in destination_hashes:
|
||||||
try:
|
try:
|
||||||
|
if background: time.sleep(0.001) # Low priority, yield thread
|
||||||
|
RNS.Transport.destinations_last_cleaned = time.time()
|
||||||
if RNS.Transport.has_path(destination_hash): has_path = True
|
if RNS.Transport.has_path(destination_hash): has_path = True
|
||||||
else:
|
else:
|
||||||
has_path = False
|
has_path = False
|
||||||
@@ -366,7 +344,8 @@ class Identity:
|
|||||||
if os.path.isfile(ratchet_path): os.unlink(ratchet_path)
|
if os.path.isfile(ratchet_path): os.unlink(ratchet_path)
|
||||||
except Exception as e: RNS.log(f"Could not clean stale ratchets for {RNS.prettyhexrep(destination_hash)}: {e}", RNS.LOG_WARNING)
|
except Exception as e: RNS.log(f"Could not clean stale ratchets for {RNS.prettyhexrep(destination_hash)}: {e}", RNS.LOG_WARNING)
|
||||||
|
|
||||||
# RNS.log(f"Total destinations: {total}, stale: {len(stale)}, removed: {removed}, no path: {no_path}, never used: {never_used}, with path: {total-no_path}, used: {total-never_used}, retained: {retained}. Completed in {RNS.prettyshorttime(time.time()-st)}", RNS.LOG_WARNING) # TODO: Remove
|
RNS.log(f"Cleaned known destinations in {RNS.prettyshorttime(time.time()-st)}", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
|
||||||
|
RNS.log(f"Total: {total}, stale: {len(stale)}, removed: {removed}, no path: {no_path}, never used: {never_used}, with path: {total-no_path}, used: {total-never_used}, retained: {retained}", RNS.LOG_PATHING) if RNS.sl(RNS.LOG_PATHING) else None
|
||||||
if not RNS.Transport.owner.is_connected_to_shared_instance: Identity.save_known_destinations(recombine=False)
|
if not RNS.Transport.owner.is_connected_to_shared_instance: Identity.save_known_destinations(recombine=False)
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
|
|||||||
@@ -276,7 +276,9 @@ class BackboneInterface(Interface):
|
|||||||
|
|
||||||
try: BackboneInterface.epoll.unregister(fileno)
|
try: BackboneInterface.epoll.unregister(fileno)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
RNS.log(f"An error occurred while deregistering file descriptor {fileno}: {e}", RNS.LOG_DEBUG)
|
if str(e).endswith("No such file or directory"): pass
|
||||||
|
elif str(e).endswith("Bad file descriptor"): pass
|
||||||
|
else: RNS.log(f"An error occurred while deregistering file descriptor {fileno}: {e}", RNS.LOG_DEBUG)
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def deregister_listeners():
|
def deregister_listeners():
|
||||||
@@ -296,7 +298,7 @@ class BackboneInterface(Interface):
|
|||||||
try: BackboneInterface.epoll.modify(fileno, select.EPOLLOUT)
|
try: BackboneInterface.epoll.modify(fileno, select.EPOLLOUT)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
if str(e).endswith("No such file or directory"): pass
|
if str(e).endswith("No such file or directory"): pass
|
||||||
elif str(e).endswith("Bad file descriptor"): pass
|
elif str(e).endswith("Bad file descriptor"): pass
|
||||||
else: RNS.log(f"Error occurred on {interface} while modifying socket EPOLL state: {e}", RNS.LOG_WARNING)
|
else: RNS.log(f"Error occurred on {interface} while modifying socket EPOLL state: {e}", RNS.LOG_WARNING)
|
||||||
raise e
|
raise e
|
||||||
|
|
||||||
@@ -342,8 +344,10 @@ class BackboneInterface(Interface):
|
|||||||
written = 0
|
written = 0
|
||||||
if not spawned_interface.detached:
|
if not spawned_interface.detached:
|
||||||
if RNS.sl(RNS.LOG_DEBUG):
|
if RNS.sl(RNS.LOG_DEBUG):
|
||||||
if str(e).endswith("Connection timed out"): pass
|
if str(e).endswith("Connection timed out"): pass
|
||||||
elif str(e).endswith("Connection reset by peer"): pass
|
elif str(e).endswith("Connection reset by peer"): pass
|
||||||
|
elif str(e).endswith("No route to host"): pass
|
||||||
|
elif str(e).endswith("Broken pipe"): pass
|
||||||
else: RNS.log(f"Error while writing to {spawned_interface}: {e}", RNS.LOG_DEBUG)
|
else: RNS.log(f"Error while writing to {spawned_interface}: {e}", RNS.LOG_DEBUG)
|
||||||
BackboneInterface.deregister_fileno(fileno)
|
BackboneInterface.deregister_fileno(fileno)
|
||||||
|
|
||||||
@@ -398,7 +402,7 @@ class BackboneInterface(Interface):
|
|||||||
try: client_socket.close()
|
try: client_socket.close()
|
||||||
except Exception as e: RNS.log(f"Error while closing socket for failed incoming connection: {e}", RNS.LOG_WARNING)
|
except Exception as e: RNS.log(f"Error while closing socket for failed incoming connection: {e}", RNS.LOG_WARNING)
|
||||||
|
|
||||||
except:
|
except Exception as e:
|
||||||
RNS.log(f"Accepting socket failed for incoming connection: {e}", RNS.LOG_WARNING)
|
RNS.log(f"Accepting socket failed for incoming connection: {e}", RNS.LOG_WARNING)
|
||||||
try: client_socket.close()
|
try: client_socket.close()
|
||||||
except Exception as e: RNS.log(f"Error while closing socket for failed incoming socket accept: {e}", RNS.LOG_WARNING)
|
except Exception as e: RNS.log(f"Error while closing socket for failed incoming socket accept: {e}", RNS.LOG_WARNING)
|
||||||
@@ -523,6 +527,11 @@ class BackboneInterface(Interface):
|
|||||||
elif str(e).endswith("Bad file descriptor"): pass
|
elif str(e).endswith("Bad file descriptor"): pass
|
||||||
else: RNS.log("Error while shutting down socket for "+str(self)+": "+str(e), RNS.LOG_ERROR)
|
else: RNS.log("Error while shutting down socket for "+str(self)+": "+str(e), RNS.LOG_ERROR)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def blocked_ip_list(self):
|
||||||
|
if not self.block_fast_flapping: return []
|
||||||
|
else: return list(self.fast_flapping.keys())
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def blocked_ip_count(self):
|
def blocked_ip_count(self):
|
||||||
if not self.block_fast_flapping: return 0
|
if not self.block_fast_flapping: return 0
|
||||||
@@ -659,7 +668,8 @@ class BackboneClientInterface(Interface):
|
|||||||
try:
|
try:
|
||||||
if self.socket != None: self.socket.shutdown(socket.SHUT_RDWR)
|
if self.socket != None: self.socket.shutdown(socket.SHUT_RDWR)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
if str(e).endswith("Transport endpoint is not connected"): pass
|
if str(e).endswith("Transport endpoint is not connected"): pass
|
||||||
|
elif str(e).endswith("Bad file descriptor"): pass
|
||||||
else: RNS.log("Error while shutting down socket for "+str(self)+": "+str(e), RNS.LOG_ERROR)
|
else: RNS.log("Error while shutting down socket for "+str(self)+": "+str(e), RNS.LOG_ERROR)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
@@ -824,7 +834,7 @@ class BackboneClientInterface(Interface):
|
|||||||
|
|
||||||
else:
|
else:
|
||||||
RNS.log("The interface "+str(self)+" is being torn down.", RNS.LOG_PATHING) if RNS.sl(RNS.LOG_PATHING) else None
|
RNS.log("The interface "+str(self)+" is being torn down.", RNS.LOG_PATHING) if RNS.sl(RNS.LOG_PATHING) else None
|
||||||
if self.parent_interface.block_fast_flapping and hasattr(self, "spawned_at"):
|
if self.parent_interface and self.parent_interface.block_fast_flapping and hasattr(self, "spawned_at"):
|
||||||
connected_time = time.time() - self.spawned_at
|
connected_time = time.time() - self.spawned_at
|
||||||
if connected_time < self.parent_interface.fast_flap_threshold:
|
if connected_time < self.parent_interface.fast_flap_threshold:
|
||||||
try:
|
try:
|
||||||
|
|||||||
@@ -102,6 +102,7 @@ class Interface:
|
|||||||
self.online = False
|
self.online = False
|
||||||
self.bitrate = 62500
|
self.bitrate = 62500
|
||||||
self.HW_MTU = None
|
self.HW_MTU = None
|
||||||
|
self.__hash = None
|
||||||
|
|
||||||
self.supports_discovery = False
|
self.supports_discovery = False
|
||||||
self.discoverable = False
|
self.discoverable = False
|
||||||
@@ -139,7 +140,8 @@ class Interface:
|
|||||||
self.op_freq_deque = deque(maxlen=Interface.OA_FREQ_SAMPLES)
|
self.op_freq_deque = deque(maxlen=Interface.OA_FREQ_SAMPLES)
|
||||||
|
|
||||||
def get_hash(self):
|
def get_hash(self):
|
||||||
return RNS.Identity.full_hash(str(self).encode("utf-8"))
|
if not self.__hash: self.__hash = RNS.Identity.full_hash(str(self).encode("utf-8"))
|
||||||
|
return self.__hash
|
||||||
|
|
||||||
# This is a generic function for determining when an interface
|
# This is a generic function for determining when an interface
|
||||||
# should activate ingress limiting. Since this can vary for
|
# should activate ingress limiting. Since this can vary for
|
||||||
@@ -364,11 +366,9 @@ class Interface:
|
|||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def get_config_obj(config_in):
|
def get_config_obj(config_in):
|
||||||
if type(config_in) == ConfigObj:
|
if type(config_in) == ConfigObj: return config_in
|
||||||
return config_in
|
|
||||||
else:
|
else:
|
||||||
try:
|
try: return ConfigObj(config_in)
|
||||||
return ConfigObj(config_in)
|
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
RNS.log(f"Could not parse supplied configuration data. The contained exception was: {e}", RNS.LOG_ERROR)
|
RNS.log(f"Could not parse supplied configuration data. The contained exception was: {e}", RNS.LOG_ERROR)
|
||||||
raise SystemError("Invalid configuration data supplied")
|
raise SystemError("Invalid configuration data supplied")
|
||||||
+9
-8
@@ -746,17 +746,17 @@ class Link:
|
|||||||
last_inbound = max(max(self.last_inbound, self.last_proof), activated_at)
|
last_inbound = max(max(self.last_inbound, self.last_proof), activated_at)
|
||||||
now = time.time()
|
now = time.time()
|
||||||
|
|
||||||
if now >= last_inbound + self.keepalive:
|
if now >= last_inbound + self.keepalive or now >= self.last_outbound + self.keepalive:
|
||||||
if self.initiator and now >= self.last_keepalive + self.keepalive:
|
if self.initiator and now >= self.last_keepalive + self.keepalive:
|
||||||
self.send_keepalive()
|
self.send_keepalive()
|
||||||
|
|
||||||
if time.time() >= last_inbound + self.stale_time:
|
if now >= last_inbound + self.stale_time:
|
||||||
sleep_time = self.rtt * self.keepalive_timeout_factor + Link.STALE_GRACE
|
sleep_time = self.rtt * self.keepalive_timeout_factor + Link.STALE_GRACE
|
||||||
self.status = Link.STALE
|
self.status = Link.STALE
|
||||||
|
|
||||||
else: sleep_time = self.keepalive
|
else: sleep_time = self.keepalive
|
||||||
|
|
||||||
else: sleep_time = (last_inbound + self.keepalive) - time.time()
|
else: sleep_time = (last_inbound + self.keepalive) - now
|
||||||
|
|
||||||
elif self.status == Link.STALE:
|
elif self.status == Link.STALE:
|
||||||
sleep_time = 0.001
|
sleep_time = 0.001
|
||||||
@@ -766,9 +766,9 @@ class Link:
|
|||||||
self.link_closed()
|
self.link_closed()
|
||||||
|
|
||||||
|
|
||||||
if sleep_time == 0: RNS.log("Warning! Link watchdog sleep time of 0!", RNS.LOG_ERROR)
|
if sleep_time == 0: RNS.log(f"Link watchdog sleep time of 0 on {self}", RNS.LOG_ERROR)
|
||||||
if sleep_time == None or sleep_time < 0:
|
if sleep_time == None or sleep_time < 0:
|
||||||
RNS.log("Timing error! Tearing down link "+str(self)+" now.", RNS.LOG_ERROR)
|
RNS.log(f"Timing error, tearing down link {self} now", RNS.LOG_ERROR)
|
||||||
self.teardown()
|
self.teardown()
|
||||||
sleep_time = 0.1
|
sleep_time = 0.1
|
||||||
|
|
||||||
@@ -1099,9 +1099,10 @@ class Link:
|
|||||||
|
|
||||||
elif packet.context == RNS.Packet.KEEPALIVE:
|
elif packet.context == RNS.Packet.KEEPALIVE:
|
||||||
if not self.initiator and packet.data == bytes([0xFF]):
|
if not self.initiator and packet.data == bytes([0xFF]):
|
||||||
keepalive_packet = RNS.Packet(self, bytes([0xFE]), context=RNS.Packet.KEEPALIVE)
|
if time.time() >= self.last_outbound + self.keepalive:
|
||||||
keepalive_packet.send()
|
keepalive_packet = RNS.Packet(self, bytes([0xFE]), context=RNS.Packet.KEEPALIVE)
|
||||||
self.had_outbound(is_keepalive = True)
|
keepalive_packet.send()
|
||||||
|
self.had_outbound(is_keepalive = True)
|
||||||
|
|
||||||
|
|
||||||
# TODO: find the most efficient way to allow multiple
|
# TODO: find the most efficient way to allow multiple
|
||||||
|
|||||||
@@ -1418,6 +1418,9 @@ class Reticulum:
|
|||||||
if hasattr(interface, "blocked_ip_count"):
|
if hasattr(interface, "blocked_ip_count"):
|
||||||
ifstats["blocked_ips"] = interface.blocked_ip_count
|
ifstats["blocked_ips"] = interface.blocked_ip_count
|
||||||
|
|
||||||
|
if hasattr(interface, "blocked_ip_list"):
|
||||||
|
ifstats["blocked_ip_list"] = interface.blocked_ip_list
|
||||||
|
|
||||||
ifstats["name"] = str(interface)
|
ifstats["name"] = str(interface)
|
||||||
ifstats["short_name"] = str(interface.name)
|
ifstats["short_name"] = str(interface.name)
|
||||||
ifstats["hash"] = interface.get_hash()
|
ifstats["hash"] = interface.get_hash()
|
||||||
|
|||||||
+83
-46
@@ -238,16 +238,19 @@ class Transport:
|
|||||||
|
|
||||||
if RNS.Reticulum.local_hops_delta(): Transport.local_hops_delta = (ord(os.urandom(1))%6)+2
|
if RNS.Reticulum.local_hops_delta(): Transport.local_hops_delta = (ord(os.urandom(1))%6)+2
|
||||||
|
|
||||||
packet_hashlist_path = RNS.Reticulum.storagepath+"/packet_hashlist"
|
packet_hashlist_path = RNS.Reticulum.storagepath+"/packet_hashlist.raw"
|
||||||
if not Transport.owner.is_connected_to_shared_instance:
|
if RNS.Reticulum.transport_enabled() and not Transport.owner.is_connected_to_shared_instance:
|
||||||
if os.path.isfile(packet_hashlist_path):
|
if os.path.isfile(packet_hashlist_path):
|
||||||
try:
|
try:
|
||||||
file = open(packet_hashlist_path, "rb")
|
with open(packet_hashlist_path, "rb") as file:
|
||||||
hashlist_data = umsgpack.unpackb(file.read())
|
hashlen = RNS.Identity.HASHLENGTH//8
|
||||||
Transport.packet_hashlist = set(hashlist_data)
|
done = False
|
||||||
file.close()
|
while not done:
|
||||||
except Exception as e:
|
packet_hash = file.read(hashlen)
|
||||||
RNS.log("Could not load packet hashlist from storage, the contained exception was: "+str(e), RNS.LOG_ERROR)
|
if len(packet_hash) == hashlen: Transport.packet_hashlist.add(packet_hash)
|
||||||
|
else: done = True
|
||||||
|
|
||||||
|
except Exception as e: RNS.log("Could not load packet hashlist from storage, the contained exception was: "+str(e), RNS.LOG_ERROR)
|
||||||
|
|
||||||
Transport.reload_blackhole()
|
Transport.reload_blackhole()
|
||||||
|
|
||||||
@@ -966,7 +969,10 @@ class Transport:
|
|||||||
# Clean known destinations
|
# Clean known destinations
|
||||||
if time.time() > Transport.destinations_last_cleaned+Transport.known_destinations_interval:
|
if time.time() > Transport.destinations_last_cleaned+Transport.known_destinations_interval:
|
||||||
Transport.destinations_last_cleaned = time.time()
|
Transport.destinations_last_cleaned = time.time()
|
||||||
def job(): RNS.Identity.clean_known_destinations()
|
def job():
|
||||||
|
try: RNS.Identity.clean_known_destinations(background=True)
|
||||||
|
except Exception as e: RNS.log(f"Error while running scheduled known destinations cleaning: {e}", RNS.LOG_ERROR)
|
||||||
|
finally: Transport.destinations_last_cleaned = time.time()
|
||||||
threading.Thread(target=job, daemon=True).start()
|
threading.Thread(target=job, daemon=True).start()
|
||||||
|
|
||||||
# Send announces for management destinations
|
# Send announces for management destinations
|
||||||
@@ -2226,7 +2232,7 @@ class Transport:
|
|||||||
else: RNS.log("Invalid link request proof in transport for link "+RNS.prettyhexrep(packet.destination_hash)+", dropping proof.", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
|
else: RNS.log("Invalid link request proof in transport for link "+RNS.prettyhexrep(packet.destination_hash)+", dropping proof.", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
|
||||||
except Exception as e: RNS.log("Could not transport link request proof. The contained exception was: "+str(e), RNS.LOG_DEBUG) if RNS.sl(LOG_DEBUG) else None
|
except Exception as e: RNS.log("Could not transport link request proof. The contained exception was: "+str(e), RNS.LOG_DEBUG) if RNS.sl(LOG_DEBUG) else None
|
||||||
else: RNS.log("Link request proof received on wrong interface, not transporting it.", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
|
else: RNS.log("Link request proof received on wrong interface, not transporting it.", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
|
||||||
else: RNS.log(f"Received link request proof with hop mismatch ({packet.hops}/{link_entry[IDX_LT_NH_IF]}), not transporting it", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
|
else: RNS.log(f"Received link request proof with hop mismatch ({packet.hops}/{link_entry[IDX_LT_REM_HOPS]}:{link_entry[IDX_LT_NH_IF]}->{link_entry[IDX_LT_RCVD_IF]}), not transporting it", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
|
||||||
|
|
||||||
else:
|
else:
|
||||||
# Check if we can deliver it to a local
|
# Check if we can deliver it to a local
|
||||||
@@ -2513,6 +2519,10 @@ class Transport:
|
|||||||
|
|
||||||
gc.collect()
|
gc.collect()
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def interface_hashes():
|
||||||
|
return {interface.get_hash() for interface in Transport.interfaces}
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def find_interface_from_hash(interface_hash):
|
def find_interface_from_hash(interface_hash):
|
||||||
for interface in Transport.interfaces:
|
for interface in Transport.interfaces:
|
||||||
@@ -3222,6 +3232,7 @@ class Transport:
|
|||||||
@staticmethod
|
@staticmethod
|
||||||
def save_packet_hashlist(background=False):
|
def save_packet_hashlist(background=False):
|
||||||
if not Transport.owner.is_connected_to_shared_instance:
|
if not Transport.owner.is_connected_to_shared_instance:
|
||||||
|
if not RNS.Reticulum.transport_enabled(): return
|
||||||
if hasattr(Transport, "saving_packet_hashlist"):
|
if hasattr(Transport, "saving_packet_hashlist"):
|
||||||
wait_interval = 0.2
|
wait_interval = 0.2
|
||||||
wait_timeout = 5
|
wait_timeout = 5
|
||||||
@@ -3236,23 +3247,28 @@ class Transport:
|
|||||||
Transport.saving_packet_hashlist = True
|
Transport.saving_packet_hashlist = True
|
||||||
save_start = time.time()
|
save_start = time.time()
|
||||||
|
|
||||||
if not RNS.Reticulum.transport_enabled(): Transport.packet_hashlist = set()
|
if RNS.Reticulum.transport_enabled(): RNS.log("Saving packet hashlist to storage...", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
|
||||||
else: RNS.log("Saving packet hashlist to storage...", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
|
else: return
|
||||||
|
|
||||||
packet_hashlist_path = RNS.Reticulum.storagepath+"/packet_hashlist"
|
round_started_at = save_start
|
||||||
file = open(packet_hashlist_path, "wb")
|
yield_threshold = 0.010
|
||||||
file.write(umsgpack.packb(list(Transport.packet_hashlist.copy())))
|
|
||||||
file.close()
|
|
||||||
|
|
||||||
save_time = time.time() - save_start
|
packet_hashlist_path = RNS.Reticulum.storagepath+"/packet_hashlist.raw"
|
||||||
if save_time < 1: time_str = str(round(save_time*1000,2))+"ms"
|
with open(packet_hashlist_path, "wb") as file:
|
||||||
else: time_str = str(round(save_time,2))+"s"
|
for packet_hash in Transport.packet_hashlist.copy():
|
||||||
RNS.log("Saved packet hashlist in "+time_str, RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
|
if background:
|
||||||
|
if time.time() - round_started_at > yield_threshold:
|
||||||
|
# Low priority, yield thread
|
||||||
|
round_started_at = time.time()
|
||||||
|
time.sleep(0.001)
|
||||||
|
|
||||||
except Exception as e:
|
file.write(packet_hash)
|
||||||
RNS.log("Could not save packet hashlist to storage, the contained exception was: "+str(e), RNS.LOG_ERROR)
|
|
||||||
|
RNS.log(f"Saved packet hashlist in {RNS.prettyshorttime(time.time()-save_start)}", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
|
||||||
|
|
||||||
|
except Exception as e: RNS.log("Could not save packet hashlist to storage, the contained exception was: "+str(e), RNS.LOG_ERROR)
|
||||||
|
finally: Transport.saving_packet_hashlist = False
|
||||||
|
|
||||||
Transport.saving_packet_hashlist = False
|
|
||||||
gc.collect()
|
gc.collect()
|
||||||
|
|
||||||
|
|
||||||
@@ -3274,40 +3290,52 @@ class Transport:
|
|||||||
save_start = time.time()
|
save_start = time.time()
|
||||||
RNS.log("Saving path table to storage...", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
|
RNS.log("Saving path table to storage...", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
|
||||||
|
|
||||||
serialised_destinations = []
|
serialised_destinations = []
|
||||||
path_table = Transport.path_table.copy()
|
path_table = Transport.path_table.copy()
|
||||||
|
interface_hashes_updated_at = 0
|
||||||
|
round_started_at = save_start
|
||||||
|
yield_threshold = 0.010
|
||||||
|
|
||||||
for destination_hash in path_table:
|
for destination_hash in path_table:
|
||||||
|
if background:
|
||||||
|
if time.time() - round_started_at > yield_threshold:
|
||||||
|
# Low priority, yield thread
|
||||||
|
round_started_at = time.time()
|
||||||
|
time.sleep(0.001)
|
||||||
try:
|
try:
|
||||||
|
# Throttle interface hash lookup to 2 seconds
|
||||||
|
if time.time() > interface_hashes_updated_at + 2:
|
||||||
|
interface_hashes = Transport.interface_hashes()
|
||||||
|
|
||||||
# Get the destination entry from the destination table
|
# Get the destination entry from the destination table
|
||||||
de = path_table[destination_hash]
|
de = path_table[destination_hash]
|
||||||
interface_hash = de[IDX_PT_RVCD_IF].get_hash()
|
interface = de[IDX_PT_RVCD_IF]
|
||||||
|
|
||||||
# Only store destination table entry if the associated
|
# Only store destination table entry if the associated
|
||||||
# interface is still active
|
# interface is still active
|
||||||
interface = Transport.find_interface_from_hash(interface_hash)
|
if not interface.get_hash() in interface_hashes: RNS.log(f"Skipping persist for path table entry {RNS.prettyhexrep(destination_hash)}, interface {interface} no longer active", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
|
||||||
if interface != None:
|
else:
|
||||||
# Get the destination entry from the destination table
|
# Get the destination entry from the destination table
|
||||||
if not destination_hash in path_table:
|
if not destination_hash in path_table:
|
||||||
RNS.log(f"Skipping persist for path table entry {RNS.prettyhexrep(destination_hash)}, no longer in table", RNS.LOG_PATHING) if RNS.sl(RNS.LOG_PATHING) else None
|
RNS.log(f"Skipping persist for path table entry {RNS.prettyhexrep(destination_hash)}, no longer in table", RNS.LOG_PATHING) if RNS.sl(RNS.LOG_PATHING) else None
|
||||||
|
|
||||||
de = path_table[destination_hash]
|
de = path_table[destination_hash]
|
||||||
timestamp = de[IDX_PT_TIMESTAMP]
|
timestamp = de[IDX_PT_TIMESTAMP]
|
||||||
received_from = de[IDX_PT_NEXT_HOP]
|
received_from = de[IDX_PT_NEXT_HOP]
|
||||||
hops = de[IDX_PT_HOPS]
|
hops = de[IDX_PT_HOPS]
|
||||||
expires = de[IDX_PT_EXPIRES]
|
expires = de[IDX_PT_EXPIRES]
|
||||||
random_blobs = de[IDX_PT_RANDBLOBS]
|
random_blobs = de[IDX_PT_RANDBLOBS]
|
||||||
packet_hash = de[IDX_PT_PACKET]
|
packet_hash = de[IDX_PT_PACKET]
|
||||||
|
interface_hash = interface.get_hash()
|
||||||
|
|
||||||
serialised_entry = [
|
serialised_entry = [ destination_hash,
|
||||||
destination_hash,
|
timestamp,
|
||||||
timestamp,
|
received_from,
|
||||||
received_from,
|
hops,
|
||||||
hops,
|
expires,
|
||||||
expires,
|
random_blobs,
|
||||||
random_blobs,
|
interface_hash,
|
||||||
interface_hash,
|
packet_hash ]
|
||||||
packet_hash
|
|
||||||
]
|
|
||||||
|
|
||||||
serialised_destinations.append(serialised_entry)
|
serialised_destinations.append(serialised_entry)
|
||||||
|
|
||||||
@@ -3353,7 +3381,16 @@ class Transport:
|
|||||||
RNS.log("Saving tunnel table to storage...", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
|
RNS.log("Saving tunnel table to storage...", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
|
||||||
|
|
||||||
serialised_tunnels = []
|
serialised_tunnels = []
|
||||||
|
round_started_at = save_start
|
||||||
|
yield_threshold = 0.010
|
||||||
|
|
||||||
for tunnel_id in Transport.tunnels.copy():
|
for tunnel_id in Transport.tunnels.copy():
|
||||||
|
if background:
|
||||||
|
if time.time() - round_started_at > yield_threshold:
|
||||||
|
# Low priority, yield thread
|
||||||
|
round_started_at = time.time()
|
||||||
|
time.sleep(0.001)
|
||||||
|
|
||||||
te = Transport.tunnels[tunnel_id]
|
te = Transport.tunnels[tunnel_id]
|
||||||
interface = te[1]
|
interface = te[1]
|
||||||
tunnel_paths = te[2].copy()
|
tunnel_paths = te[2].copy()
|
||||||
|
|||||||
+1
-1
@@ -1 +1 @@
|
|||||||
__version__ = "1.3.9"
|
__version__ = "1.4.0"
|
||||||
|
|||||||
Reference in New Issue
Block a user