mirror of
https://github.com/markqvist/Reticulum.git
synced 2026-07-21 23:48:12 -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
|
||||
from threading import Lock
|
||||
from .vendor import umsgpack as msgpack
|
||||
from collections import deque
|
||||
|
||||
NAME = 0xFF
|
||||
TRANSPORT_ID = 0xFE
|
||||
@@ -31,7 +32,7 @@ APP_NAME = "rnstransport"
|
||||
|
||||
class InterfaceAnnouncer():
|
||||
JOB_INTERVAL = 60
|
||||
DEFAULT_STAMP_VALUE = 14
|
||||
DEFAULT_STAMP_VALUE = 16
|
||||
WORKBLOCK_EXPAND_ROUNDS = 20
|
||||
|
||||
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.panic()
|
||||
|
||||
self.aspect_filter = APP_NAME+".discovery.interface"
|
||||
self.required_value = required_value
|
||||
self.callback = callback
|
||||
self.stamper = LXStamper
|
||||
self.aspect_filter = APP_NAME+".discovery.interface"
|
||||
self.required_value = required_value
|
||||
self.callback = callback
|
||||
self.stamper = LXStamper
|
||||
self.valid_cache = {}
|
||||
self.valid_cache_max = 2048
|
||||
self.invalid_cache = deque(maxlen=2048)
|
||||
self.validation_lock = threading.Lock()
|
||||
|
||||
@staticmethod
|
||||
def sanitize_name(name):
|
||||
@@ -250,21 +255,43 @@ class InterfaceAnnounceHandler:
|
||||
app_data = app_data[1:]
|
||||
signed = flags & self.FLAG_SIGNED
|
||||
encrypted = flags & self.FLAG_ENCRYPTED
|
||||
fullhash = RNS.Identity.full_hash(app_data)
|
||||
|
||||
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 fullhash in self.invalid_cache:
|
||||
RNS.log(f"Ignored previously discovered interface with insufficient stamp value", RNS.LOG_PATHING) if RNS.sl(RNS.LOG_PATHING) else None
|
||||
return
|
||||
|
||||
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)
|
||||
if fullhash in self.valid_cache:
|
||||
RNS.log(f"Discovery announce cache hit", RNS.LOG_PATHING) if RNS.sl(RNS.LOG_PATHING) else None
|
||||
cache_entry = self.valid_cache[fullhash]
|
||||
stamp = cache_entry["stamp"]
|
||||
packed = cache_entry["packed"]
|
||||
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:
|
||||
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
|
||||
|
||||
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
|
||||
|
||||
@staticmethod
|
||||
def save_known_destinations(background=False, recombine=True):
|
||||
# TODO: Improve the storage method so we don't have to
|
||||
# deserialize and serialize the entire table on every
|
||||
# save, but the only changes. It might be possible to
|
||||
# simply overwrite on exit now that every local client
|
||||
# disconnect triggers a data persist.
|
||||
def save_known_destinations(background=False, recombine=False):
|
||||
if recombine: RNS.log(f"Recombining known destinations from disk cache on persist is deprecated, argument ignored", RNS.LOG_WARNING)
|
||||
if RNS.Transport.owner.is_connected_to_shared_instance: return
|
||||
|
||||
try:
|
||||
if hasattr(Identity, "saving_known_destinations"):
|
||||
@@ -194,28 +191,9 @@ class Identity:
|
||||
|
||||
Identity.saving_known_destinations = True
|
||||
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
|
||||
temp_file = RNS.Reticulum.storagepath+f"/known_destinations.tmp.{time.time()}"
|
||||
|
||||
temp_file = RNS.Reticulum.storagepath+f"/known_destinations.tmp.{time.time()}"
|
||||
try:
|
||||
with open(temp_file,"wb") as file: umsgpack.dump(Identity.known_destinations.copy(), file)
|
||||
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)
|
||||
raise e
|
||||
|
||||
save_time = time.time() - save_start
|
||||
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
|
||||
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
|
||||
|
||||
except Exception as e:
|
||||
RNS.log("Error while saving known destinations to disk, the contained exception was: "+str(e), RNS.LOG_ERROR)
|
||||
RNS.trace_exception(e)
|
||||
|
||||
Identity.saving_known_destinations = False
|
||||
finally: Identity.saving_known_destinations = False
|
||||
|
||||
@staticmethod
|
||||
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)
|
||||
|
||||
@staticmethod
|
||||
def clean_known_destinations():
|
||||
def clean_known_destinations(background=False):
|
||||
now = time.time()
|
||||
st = now
|
||||
total = len(Identity.known_destinations)
|
||||
@@ -318,9 +292,13 @@ class Identity:
|
||||
never_used = 0
|
||||
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())
|
||||
for destination_hash in destination_hashes:
|
||||
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
|
||||
else:
|
||||
has_path = False
|
||||
@@ -366,7 +344,8 @@ class Identity:
|
||||
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)
|
||||
|
||||
# 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)
|
||||
|
||||
@staticmethod
|
||||
|
||||
@@ -276,7 +276,9 @@ class BackboneInterface(Interface):
|
||||
|
||||
try: BackboneInterface.epoll.unregister(fileno)
|
||||
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
|
||||
def deregister_listeners():
|
||||
@@ -296,7 +298,7 @@ class BackboneInterface(Interface):
|
||||
try: BackboneInterface.epoll.modify(fileno, select.EPOLLOUT)
|
||||
except Exception as e:
|
||||
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)
|
||||
raise e
|
||||
|
||||
@@ -342,8 +344,10 @@ class BackboneInterface(Interface):
|
||||
written = 0
|
||||
if not spawned_interface.detached:
|
||||
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("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)
|
||||
BackboneInterface.deregister_fileno(fileno)
|
||||
|
||||
@@ -398,7 +402,7 @@ class BackboneInterface(Interface):
|
||||
try: client_socket.close()
|
||||
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)
|
||||
try: client_socket.close()
|
||||
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
|
||||
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
|
||||
def blocked_ip_count(self):
|
||||
if not self.block_fast_flapping: return 0
|
||||
@@ -659,7 +668,8 @@ class BackboneClientInterface(Interface):
|
||||
try:
|
||||
if self.socket != None: self.socket.shutdown(socket.SHUT_RDWR)
|
||||
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)
|
||||
|
||||
try:
|
||||
@@ -824,7 +834,7 @@ class BackboneClientInterface(Interface):
|
||||
|
||||
else:
|
||||
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
|
||||
if connected_time < self.parent_interface.fast_flap_threshold:
|
||||
try:
|
||||
|
||||
@@ -102,6 +102,7 @@ class Interface:
|
||||
self.online = False
|
||||
self.bitrate = 62500
|
||||
self.HW_MTU = None
|
||||
self.__hash = None
|
||||
|
||||
self.supports_discovery = False
|
||||
self.discoverable = False
|
||||
@@ -139,7 +140,8 @@ class Interface:
|
||||
self.op_freq_deque = deque(maxlen=Interface.OA_FREQ_SAMPLES)
|
||||
|
||||
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
|
||||
# should activate ingress limiting. Since this can vary for
|
||||
@@ -364,11 +366,9 @@ class Interface:
|
||||
|
||||
@staticmethod
|
||||
def get_config_obj(config_in):
|
||||
if type(config_in) == ConfigObj:
|
||||
return config_in
|
||||
if type(config_in) == ConfigObj: return config_in
|
||||
else:
|
||||
try:
|
||||
return ConfigObj(config_in)
|
||||
try: return ConfigObj(config_in)
|
||||
except Exception as e:
|
||||
RNS.log(f"Could not parse supplied configuration data. The contained exception was: {e}", RNS.LOG_ERROR)
|
||||
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)
|
||||
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:
|
||||
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
|
||||
self.status = Link.STALE
|
||||
|
||||
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:
|
||||
sleep_time = 0.001
|
||||
@@ -766,9 +766,9 @@ class Link:
|
||||
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:
|
||||
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()
|
||||
sleep_time = 0.1
|
||||
|
||||
@@ -1099,9 +1099,10 @@ class Link:
|
||||
|
||||
elif packet.context == RNS.Packet.KEEPALIVE:
|
||||
if not self.initiator and packet.data == bytes([0xFF]):
|
||||
keepalive_packet = RNS.Packet(self, bytes([0xFE]), context=RNS.Packet.KEEPALIVE)
|
||||
keepalive_packet.send()
|
||||
self.had_outbound(is_keepalive = True)
|
||||
if time.time() >= self.last_outbound + self.keepalive:
|
||||
keepalive_packet = RNS.Packet(self, bytes([0xFE]), context=RNS.Packet.KEEPALIVE)
|
||||
keepalive_packet.send()
|
||||
self.had_outbound(is_keepalive = True)
|
||||
|
||||
|
||||
# TODO: find the most efficient way to allow multiple
|
||||
|
||||
@@ -1418,6 +1418,9 @@ class Reticulum:
|
||||
if hasattr(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["short_name"] = str(interface.name)
|
||||
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
|
||||
|
||||
packet_hashlist_path = RNS.Reticulum.storagepath+"/packet_hashlist"
|
||||
if not Transport.owner.is_connected_to_shared_instance:
|
||||
packet_hashlist_path = RNS.Reticulum.storagepath+"/packet_hashlist.raw"
|
||||
if RNS.Reticulum.transport_enabled() and not Transport.owner.is_connected_to_shared_instance:
|
||||
if os.path.isfile(packet_hashlist_path):
|
||||
try:
|
||||
file = open(packet_hashlist_path, "rb")
|
||||
hashlist_data = umsgpack.unpackb(file.read())
|
||||
Transport.packet_hashlist = set(hashlist_data)
|
||||
file.close()
|
||||
except Exception as e:
|
||||
RNS.log("Could not load packet hashlist from storage, the contained exception was: "+str(e), RNS.LOG_ERROR)
|
||||
with open(packet_hashlist_path, "rb") as file:
|
||||
hashlen = RNS.Identity.HASHLENGTH//8
|
||||
done = False
|
||||
while not done:
|
||||
packet_hash = file.read(hashlen)
|
||||
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()
|
||||
|
||||
@@ -966,7 +969,10 @@ class Transport:
|
||||
# Clean known destinations
|
||||
if time.time() > Transport.destinations_last_cleaned+Transport.known_destinations_interval:
|
||||
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()
|
||||
|
||||
# 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
|
||||
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(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:
|
||||
# Check if we can deliver it to a local
|
||||
@@ -2513,6 +2519,10 @@ class Transport:
|
||||
|
||||
gc.collect()
|
||||
|
||||
@staticmethod
|
||||
def interface_hashes():
|
||||
return {interface.get_hash() for interface in Transport.interfaces}
|
||||
|
||||
@staticmethod
|
||||
def find_interface_from_hash(interface_hash):
|
||||
for interface in Transport.interfaces:
|
||||
@@ -3222,6 +3232,7 @@ class Transport:
|
||||
@staticmethod
|
||||
def save_packet_hashlist(background=False):
|
||||
if not Transport.owner.is_connected_to_shared_instance:
|
||||
if not RNS.Reticulum.transport_enabled(): return
|
||||
if hasattr(Transport, "saving_packet_hashlist"):
|
||||
wait_interval = 0.2
|
||||
wait_timeout = 5
|
||||
@@ -3236,23 +3247,28 @@ class Transport:
|
||||
Transport.saving_packet_hashlist = True
|
||||
save_start = time.time()
|
||||
|
||||
if not RNS.Reticulum.transport_enabled(): Transport.packet_hashlist = set()
|
||||
else: RNS.log("Saving packet hashlist to storage...", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
|
||||
if RNS.Reticulum.transport_enabled(): 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"
|
||||
file = open(packet_hashlist_path, "wb")
|
||||
file.write(umsgpack.packb(list(Transport.packet_hashlist.copy())))
|
||||
file.close()
|
||||
round_started_at = save_start
|
||||
yield_threshold = 0.010
|
||||
|
||||
save_time = time.time() - save_start
|
||||
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 packet hashlist in "+time_str, RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
|
||||
packet_hashlist_path = RNS.Reticulum.storagepath+"/packet_hashlist.raw"
|
||||
with open(packet_hashlist_path, "wb") as file:
|
||||
for packet_hash in Transport.packet_hashlist.copy():
|
||||
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:
|
||||
RNS.log("Could not save packet hashlist to storage, the contained exception was: "+str(e), RNS.LOG_ERROR)
|
||||
file.write(packet_hash)
|
||||
|
||||
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()
|
||||
|
||||
|
||||
@@ -3274,40 +3290,52 @@ class Transport:
|
||||
save_start = time.time()
|
||||
RNS.log("Saving path table to storage...", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
|
||||
|
||||
serialised_destinations = []
|
||||
path_table = Transport.path_table.copy()
|
||||
serialised_destinations = []
|
||||
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:
|
||||
if background:
|
||||
if time.time() - round_started_at > yield_threshold:
|
||||
# Low priority, yield thread
|
||||
round_started_at = time.time()
|
||||
time.sleep(0.001)
|
||||
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
|
||||
de = path_table[destination_hash]
|
||||
interface_hash = de[IDX_PT_RVCD_IF].get_hash()
|
||||
de = path_table[destination_hash]
|
||||
interface = de[IDX_PT_RVCD_IF]
|
||||
|
||||
# Only store destination table entry if the associated
|
||||
# interface is still active
|
||||
interface = Transport.find_interface_from_hash(interface_hash)
|
||||
if interface != None:
|
||||
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
|
||||
else:
|
||||
# Get the destination entry from the destination 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
|
||||
|
||||
de = path_table[destination_hash]
|
||||
timestamp = de[IDX_PT_TIMESTAMP]
|
||||
received_from = de[IDX_PT_NEXT_HOP]
|
||||
hops = de[IDX_PT_HOPS]
|
||||
expires = de[IDX_PT_EXPIRES]
|
||||
random_blobs = de[IDX_PT_RANDBLOBS]
|
||||
packet_hash = de[IDX_PT_PACKET]
|
||||
de = path_table[destination_hash]
|
||||
timestamp = de[IDX_PT_TIMESTAMP]
|
||||
received_from = de[IDX_PT_NEXT_HOP]
|
||||
hops = de[IDX_PT_HOPS]
|
||||
expires = de[IDX_PT_EXPIRES]
|
||||
random_blobs = de[IDX_PT_RANDBLOBS]
|
||||
packet_hash = de[IDX_PT_PACKET]
|
||||
interface_hash = interface.get_hash()
|
||||
|
||||
serialised_entry = [
|
||||
destination_hash,
|
||||
timestamp,
|
||||
received_from,
|
||||
hops,
|
||||
expires,
|
||||
random_blobs,
|
||||
interface_hash,
|
||||
packet_hash
|
||||
]
|
||||
serialised_entry = [ destination_hash,
|
||||
timestamp,
|
||||
received_from,
|
||||
hops,
|
||||
expires,
|
||||
random_blobs,
|
||||
interface_hash,
|
||||
packet_hash ]
|
||||
|
||||
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
|
||||
|
||||
serialised_tunnels = []
|
||||
round_started_at = save_start
|
||||
yield_threshold = 0.010
|
||||
|
||||
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]
|
||||
interface = te[1]
|
||||
tunnel_paths = te[2].copy()
|
||||
|
||||
+1
-1
@@ -1 +1 @@
|
||||
__version__ = "1.3.9"
|
||||
__version__ = "1.4.0"
|
||||
|
||||
Reference in New Issue
Block a user