markqvist/Reticulum
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Refactoring work for free-threaded transport I/O

Browse source
This commit is contained in:
Mark Qvist 2026-04-12 14:55:42 +02:00
parent fa353fb0b3
commit 23c0a493b1
4 changed files with 208 additions and 228 deletions
Download patch file Download diff file Expand all files Collapse all files

3
RNS/Destination.py
View file

@ -421,8 +421,7 @@ class Destination:
else: else:
if packet.packet_type == RNS.Packet.DATA: if packet.packet_type == RNS.Packet.DATA:
if self.callbacks.packet != None: if self.callbacks.packet != None:
try: try: self.callbacks.packet(plaintext, packet)
self.callbacks.packet(plaintext, packet)
except Exception as e: except Exception as e:
RNS.log("Error while executing receive callback from "+str(self)+". The contained exception was: "+str(e), RNS.LOG_ERROR) RNS.log("Error while executing receive callback from "+str(self)+". The contained exception was: "+str(e), RNS.LOG_ERROR)

425
RNS/Transport.py
View file

@ -119,8 +119,11 @@ class Transport:
discovery_pr_tags = [] # A table for keeping track of tagged path requests discovery_pr_tags = [] # A table for keeping track of tagged path requests
max_pr_tags = 32000 # Maximum amount of unique path request tags to remember max_pr_tags = 32000 # Maximum amount of unique path request tags to remember
destinations_lock = Lock()
inbound_announce_lock = Lock() inbound_announce_lock = Lock()
announce_table_lock = Lock() announce_table_lock = Lock()
announce_rate_table_lock = Lock()
announce_handler_lock = Lock()
path_table_lock = Lock() path_table_lock = Lock()
reverse_table_lock = Lock() reverse_table_lock = Lock()
link_table_lock = Lock() link_table_lock = Lock()
@ -156,8 +159,6 @@ class Transport:
pending_local_path_requests = {} pending_local_path_requests = {}
start_time = None start_time = None
jobs_locked = False
jobs_running = False
hashlist_maxsize = 1000000 hashlist_maxsize = 1000000
job_interval = 0.250 job_interval = 0.250
links_last_checked = 0.0 links_last_checked = 0.0
@ -193,7 +194,6 @@ class Transport:
@staticmethod @staticmethod
def start(reticulum_instance): def start(reticulum_instance):
Transport.jobs_running = True
Transport.owner = reticulum_instance Transport.owner = reticulum_instance
if Transport.identity == None: if Transport.identity == None:
@ -262,7 +262,6 @@ class Transport:
Transport.last_mgmt_announce = time.time() - Transport.mgmt_announce_interval + 15 Transport.last_mgmt_announce = time.time() - Transport.mgmt_announce_interval + 15
# Start job loops # Start job loops
Transport.jobs_running = False
threading.Thread(target=Transport.jobloop, daemon=True).start() threading.Thread(target=Transport.jobloop, daemon=True).start()
threading.Thread(target=Transport.count_traffic_loop, daemon=True).start() threading.Thread(target=Transport.count_traffic_loop, daemon=True).start()
@ -473,7 +472,6 @@ class Transport:
outgoing = [] outgoing = []
path_requests = {} path_requests = {}
blocked_if = None blocked_if = None
Transport.jobs_running = True
try: try:
with Transport.jobs_lock: with Transport.jobs_lock:
@ -912,8 +910,6 @@ class Transport:
RNS.log("An exception occurred while running Transport jobs.", RNS.LOG_ERROR) RNS.log("An exception occurred while running Transport jobs.", RNS.LOG_ERROR)
RNS.log("The contained exception was: "+str(e), RNS.LOG_ERROR) RNS.log("The contained exception was: "+str(e), RNS.LOG_ERROR)
Transport.jobs_running = False
for packet in outgoing: packet.send() for packet in outgoing: packet.send()
for destination_hash in path_requests: for destination_hash in path_requests:
@ -966,10 +962,6 @@ class Transport:
@staticmethod @staticmethod
def outbound(packet): def outbound(packet):
while (Transport.jobs_running): sleep(0.0005)
Transport.jobs_locked = True
sent = False sent = False
outbound_time = time.time() outbound_time = time.time()
@ -992,31 +984,37 @@ class Transport:
if generate_receipt: if generate_receipt:
packet.receipt = RNS.PacketReceipt(packet) packet.receipt = RNS.PacketReceipt(packet)
Transport.receipts.append(packet.receipt) with Transport.receipts_lock: Transport.receipts.append(packet.receipt)
# TODO: Enable when caching has been redesigned # TODO: Enable when caching has been redesigned
# Transport.cache(packet) # Transport.cache(packet)
# Check if we have a known path for the destination in the path table # Check if we have a known path for the destination in the path table
if packet.packet_type != RNS.Packet.ANNOUNCE and packet.destination.type != RNS.Destination.PLAIN and packet.destination.type != RNS.Destination.GROUP and packet.destination_hash in Transport.path_table: if packet.packet_type != RNS.Packet.ANNOUNCE and packet.destination.type != RNS.Destination.PLAIN and packet.destination.type != RNS.Destination.GROUP and packet.destination_hash in Transport.path_table:
outbound_interface = Transport.path_table[packet.destination_hash][IDX_PT_RVCD_IF] with Transport.path_table_lock:
if not packet.destination_hash in Transport.path_table:
RNS.log(f"Dropped packet since path table entry disappeared during outbound processing", RNS.LOG_WARNING)
return False
else: path_entry = Transport.path_table[packet.destination_hash]
outbound_interface = path_entry[IDX_PT_RVCD_IF]
# If there's more than one hop to the destination, and we know # If there's more than one hop to the destination, and we know
# a path, we insert the packet into transport by adding the next # a path, we insert the packet into transport by adding the next
# transport nodes address to the header, and modifying the flags. # transport nodes address to the header, and modifying the flags.
# This rule applies both for "normal" transport, and when connected # This rule applies both for "normal" transport, and when connected
# to a local shared Reticulum instance. # to a local shared Reticulum instance.
if Transport.path_table[packet.destination_hash][IDX_PT_HOPS] > 1: if path_entry[IDX_PT_HOPS] > 1:
if packet.header_type == RNS.Packet.HEADER_1: if packet.header_type == RNS.Packet.HEADER_1:
# Insert packet into transport # Insert packet into transport
new_flags = (RNS.Packet.HEADER_2) << 6 | (Transport.TRANSPORT) << 4 | (packet.flags & 0b00001111) new_flags = (RNS.Packet.HEADER_2) << 6 | (Transport.TRANSPORT) << 4 | (packet.flags & 0b00001111)
new_raw = struct.pack("!B", new_flags) new_raw = struct.pack("!B", new_flags)
new_raw += packet.raw[1:2] new_raw += packet.raw[1:2]
new_raw += Transport.path_table[packet.destination_hash][IDX_PT_NEXT_HOP] new_raw += path_entry[IDX_PT_NEXT_HOP]
new_raw += packet.raw[2:] new_raw += packet.raw[2:]
packet_sent(packet) packet_sent(packet)
Transport.transmit(outbound_interface, new_raw) Transport.transmit(outbound_interface, new_raw)
Transport.path_table[packet.destination_hash][IDX_PT_TIMESTAMP] = time.time() path_entry[IDX_PT_TIMESTAMP] = time.time()
sent = True sent = True
# In the special case where we are connected to a local shared # In the special case where we are connected to a local shared
@ -1026,17 +1024,17 @@ class Transport:
# one hop away would just be broadcast directly, but since we # one hop away would just be broadcast directly, but since we
# are "behind" a shared instance, we need to get that instance # are "behind" a shared instance, we need to get that instance
# to transport it onto the network. # to transport it onto the network.
elif Transport.path_table[packet.destination_hash][IDX_PT_HOPS] == 1 and Transport.owner.is_connected_to_shared_instance: elif path_entry[IDX_PT_HOPS] == 1 and Transport.owner.is_connected_to_shared_instance:
if packet.header_type == RNS.Packet.HEADER_1: if packet.header_type == RNS.Packet.HEADER_1:
# Insert packet into transport # Insert packet into transport
new_flags = (RNS.Packet.HEADER_2) << 6 | (Transport.TRANSPORT) << 4 | (packet.flags & 0b00001111) new_flags = (RNS.Packet.HEADER_2) << 6 | (Transport.TRANSPORT) << 4 | (packet.flags & 0b00001111)
new_raw = struct.pack("!B", new_flags) new_raw = struct.pack("!B", new_flags)
new_raw += packet.raw[1:2] new_raw += packet.raw[1:2]
new_raw += Transport.path_table[packet.destination_hash][IDX_PT_NEXT_HOP] new_raw += path_entry[IDX_PT_NEXT_HOP]
new_raw += packet.raw[2:] new_raw += packet.raw[2:]
packet_sent(packet) packet_sent(packet)
Transport.transmit(outbound_interface, new_raw) Transport.transmit(outbound_interface, new_raw)
Transport.path_table[packet.destination_hash][IDX_PT_TIMESTAMP] = time.time() path_entry[IDX_PT_TIMESTAMP] = time.time()
sent = True sent = True
# If none of the above applies, we know the destination is # If none of the above applies, we know the destination is
@ -1058,10 +1056,8 @@ class Transport:
should_transmit = True should_transmit = True
if packet.destination.type == RNS.Destination.LINK: if packet.destination.type == RNS.Destination.LINK:
if packet.destination.status == RNS.Link.CLOSED: if packet.destination.status == RNS.Link.CLOSED: should_transmit = False
should_transmit = False if interface != packet.destination.attached_interface: should_transmit = False
if interface != packet.destination.attached_interface:
should_transmit = False
if packet.attached_interface != None and interface != packet.attached_interface: if packet.attached_interface != None and interface != packet.attached_interface:
should_transmit = False should_transmit = False
@ -1073,7 +1069,9 @@ class Transport:
should_transmit = False should_transmit = False
elif interface.mode == RNS.Interfaces.Interface.Interface.MODE_ROAMING: elif interface.mode == RNS.Interfaces.Interface.Interface.MODE_ROAMING:
local_destination = next((d for d in Transport.destinations if d.hash == packet.destination_hash), None) with Transport.destinations_lock:
local_destination = next((d for d in Transport.destinations if d.hash == packet.destination_hash), None)
if local_destination != None: if local_destination != None:
# RNS.log("Allowing announce broadcast on roaming-mode interface from instance-local destination", RNS.LOG_EXTREME) # RNS.log("Allowing announce broadcast on roaming-mode interface from instance-local destination", RNS.LOG_EXTREME)
pass pass
@ -1094,7 +1092,8 @@ class Transport:
should_transmit = False should_transmit = False
elif interface.mode == RNS.Interfaces.Interface.Interface.MODE_BOUNDARY: elif interface.mode == RNS.Interfaces.Interface.Interface.MODE_BOUNDARY:
local_destination = next((d for d in Transport.destinations if d.hash == packet.destination_hash), None) with Transport.destinations_lock:
local_destination = next((d for d in Transport.destinations if d.hash == packet.destination_hash), None)
if local_destination != None: if local_destination != None:
# RNS.log("Allowing announce broadcast on boundary-mode interface from instance-local destination", RNS.LOG_EXTREME) # RNS.log("Allowing announce broadcast on boundary-mode interface from instance-local destination", RNS.LOG_EXTREME)
pass pass
@ -1117,14 +1116,9 @@ class Transport:
# TODO: Rethink whether this is actually optimal. # TODO: Rethink whether this is actually optimal.
if packet.hops > 0: if packet.hops > 0:
if not hasattr(interface, "announce_cap"): if not hasattr(interface, "announce_cap"): interface.announce_cap = RNS.Reticulum.ANNOUNCE_CAP
interface.announce_cap = RNS.Reticulum.ANNOUNCE_CAP if not hasattr(interface, "announce_allowed_at"): interface.announce_allowed_at = 0
if not hasattr(interface, "announce_queue"): interface.announce_queue = []
if not hasattr(interface, "announce_allowed_at"):
interface.announce_allowed_at = 0
if not hasattr(interface, "announce_queue"):
interface.announce_queue = []
queued_announces = True if len(interface.announce_queue) > 0 else False queued_announces = True if len(interface.announce_queue) > 0 else False
if not queued_announces and outbound_time > interface.announce_allowed_at and interface.bitrate != None and interface.bitrate != 0: if not queued_announces and outbound_time > interface.announce_allowed_at and interface.bitrate != None and interface.bitrate != 0:
@ -1154,13 +1148,11 @@ class Transport:
e["raw"] = packet.raw e["raw"] = packet.raw
if should_queue: if should_queue:
entry = { entry = { "destination": packet.destination_hash,
"destination": packet.destination_hash, "time": outbound_time,
"time": outbound_time, "hops": packet.hops,
"hops": packet.hops, "emitted": Transport.announce_emitted(packet),
"emitted": Transport.announce_emitted(packet), "raw": packet.raw }
"raw": packet.raw
}
queued_announces = True if len(interface.announce_queue) > 0 else False queued_announces = True if len(interface.announce_queue) > 0 else False
interface.announce_queue.append(entry) interface.announce_queue.append(entry)
@ -1170,10 +1162,8 @@ class Transport:
timer = threading.Timer(wait_time, interface.process_announce_queue) timer = threading.Timer(wait_time, interface.process_announce_queue)
timer.start() timer.start()
if wait_time < 1: if wait_time < 1: wait_time_str = str(round(wait_time*1000,2))+"ms"
wait_time_str = str(round(wait_time*1000,2))+"ms" else: wait_time_str = str(round(wait_time*1,2))+"s"
else:
wait_time_str = str(round(wait_time*1,2))+"s"
ql_str = str(len(interface.announce_queue)) ql_str = str(len(interface.announce_queue))
RNS.log("Added announce to queue (height "+ql_str+") on "+str(interface)+" for processing in "+wait_time_str, RNS.LOG_EXTREME) RNS.log("Added announce to queue (height "+ql_str+") on "+str(interface)+" for processing in "+wait_time_str, RNS.LOG_EXTREME)
@ -1181,10 +1171,8 @@ class Transport:
else: else:
wait_time = max(interface.announce_allowed_at - time.time(), 0) wait_time = max(interface.announce_allowed_at - time.time(), 0)
if wait_time < 1: if wait_time < 1: wait_time_str = str(round(wait_time*1000,2))+"ms"
wait_time_str = str(round(wait_time*1000,2))+"ms" else: wait_time_str = str(round(wait_time*1,2))+"s"
else:
wait_time_str = str(round(wait_time*1,2))+"s"
ql_str = str(len(interface.announce_queue)) ql_str = str(len(interface.announce_queue))
RNS.log("Added announce to queue (height "+ql_str+") on "+str(interface)+" for processing in "+wait_time_str, RNS.LOG_EXTREME) RNS.log("Added announce to queue (height "+ql_str+") on "+str(interface)+" for processing in "+wait_time_str, RNS.LOG_EXTREME)
@ -1199,12 +1187,10 @@ class Transport:
stored_hash = True stored_hash = True
Transport.transmit(interface, packet.raw) Transport.transmit(interface, packet.raw)
if packet.packet_type == RNS.Packet.ANNOUNCE: if packet.packet_type == RNS.Packet.ANNOUNCE: interface.sent_announce()
interface.sent_announce()
packet_sent(packet) packet_sent(packet)
sent = True sent = True
Transport.jobs_locked = False
return sent return sent
@staticmethod @staticmethod
@ -1253,8 +1239,7 @@ class Transport:
RNS.log("Dropped invalid GROUP announce packet", RNS.LOG_DEBUG) RNS.log("Dropped invalid GROUP announce packet", RNS.LOG_DEBUG)
return False return False
if not packet.packet_hash in Transport.packet_hashlist and not packet.packet_hash in Transport.packet_hashlist_prev: if not packet.packet_hash in Transport.packet_hashlist and not packet.packet_hash in Transport.packet_hashlist_prev: return True
return True
else: else:
if packet.packet_type == RNS.Packet.ANNOUNCE: if packet.packet_type == RNS.Packet.ANNOUNCE:
if packet.destination_type == RNS.Destination.SINGLE: if packet.destination_type == RNS.Destination.SINGLE:
@ -1321,16 +1306,10 @@ class Transport:
else: return else: return
while (Transport.jobs_running): sleep(0.0005)
if Transport.identity == None: return if Transport.identity == None: return
Transport.jobs_locked = True
packet = RNS.Packet(None, raw) packet = RNS.Packet(None, raw)
if not packet.unpack(): if not packet.unpack(): return
Transport.jobs_locked = False
return
packet.receiving_interface = interface packet.receiving_interface = interface
packet.hops += 1 packet.hops += 1
@ -1430,9 +1409,7 @@ class Transport:
# it, do so and stop processing. Otherwise resume # it, do so and stop processing. Otherwise resume
# normal processing. # normal processing.
if packet.context == RNS.Packet.CACHE_REQUEST: if packet.context == RNS.Packet.CACHE_REQUEST:
if Transport.cache_request_packet(packet): if Transport.cache_request_packet(packet): return
Transport.jobs_locked = False
return
# If the packet is in transport, check whether we # If the packet is in transport, check whether we
# are the designated next hop, and process it # are the designated next hop, and process it
@ -1503,7 +1480,7 @@ class Transport:
False, # 7: Validated False, # 7: Validated
proof_timeout ] # 8: Proof timeout timestamp proof_timeout ] # 8: Proof timeout timestamp
Transport.link_table[RNS.Link.link_id_from_lr_packet(packet)] = link_entry with Transport.link_table_lock: Transport.link_table[RNS.Link.link_id_from_lr_packet(packet)] = link_entry
else: else:
# Entry format is # Entry format is
@ -1511,10 +1488,10 @@ class Transport:
outbound_interface, # 1: Outbound interface outbound_interface, # 1: Outbound interface
time.time() ] # 2: Timestamp time.time() ] # 2: Timestamp
Transport.reverse_table[packet.getTruncatedHash()] = reverse_entry with Transport.reverse_table_lock: Transport.reverse_table[packet.getTruncatedHash()] = reverse_entry
Transport.transmit(outbound_interface, new_raw) Transport.transmit(outbound_interface, new_raw)
Transport.path_table[packet.destination_hash][IDX_PT_TIMESTAMP] = time.time() with Transport.path_table_lock: Transport.path_table[packet.destination_hash][IDX_PT_TIMESTAMP] = time.time()
else: else:
# TODO: There should probably be some kind of REJECT # TODO: There should probably be some kind of REJECT
@ -1561,8 +1538,7 @@ class Transport:
Transport.transmit(outbound_interface, new_raw) Transport.transmit(outbound_interface, new_raw)
Transport.link_table[packet.destination_hash][IDX_LT_TIMESTAMP] = time.time() Transport.link_table[packet.destination_hash][IDX_LT_TIMESTAMP] = time.time()
# TODO: Test and possibly enable this at some point # TODO: Can we return safely here? Test and possibly enable this at some point.
# Transport.jobs_locked = False
# return # return
@ -1580,10 +1556,10 @@ class Transport:
# by normal announce rate limiting. # by normal announce rate limiting.
if interface.should_ingress_limit(): if interface.should_ingress_limit():
interface.hold_announce(packet) interface.hold_announce(packet)
Transport.jobs_locked = False
return return
local_destination = next((d for d in Transport.destinations if d.hash == packet.destination_hash), None) with Transport.destinations_lock:
local_destination = next((d for d in Transport.destinations if d.hash == packet.destination_hash), None)
if local_destination == None and RNS.Identity.validate_announce(packet): if local_destination == None and RNS.Identity.validate_announce(packet):
if packet.transport_id != None: if packet.transport_id != None:
received_from = packet.transport_id received_from = packet.transport_id
@ -1618,7 +1594,8 @@ class Transport:
# First, check that the announce is not for a destination # First, check that the announce is not for a destination
# local to this system, and that hops are less than the max # local to this system, and that hops are less than the max
if (not any(packet.destination_hash == d.hash for d in Transport.destinations) and packet.hops < Transport.PATHFINDER_M+1): with Transport.destinations_lock: local_and_hops_condition = (not any(packet.destination_hash == d.hash for d in Transport.destinations) and packet.hops < Transport.PATHFINDER_M+1)
if local_and_hops_condition:
announce_emitted = Transport.announce_emitted(packet) announce_emitted = Transport.announce_emitted(packet)
random_blob = packet.data[RNS.Identity.KEYSIZE//8+RNS.Identity.NAME_HASH_LENGTH//8:RNS.Identity.KEYSIZE//8+RNS.Identity.NAME_HASH_LENGTH//8+10] random_blob = packet.data[RNS.Identity.KEYSIZE//8+RNS.Identity.NAME_HASH_LENGTH//8:RNS.Identity.KEYSIZE//8+RNS.Identity.NAME_HASH_LENGTH//8+10]
@ -1700,36 +1677,38 @@ class Transport:
if should_add: if should_add:
now = time.time() now = time.time()
is_from_local_client = Transport.from_local_client(packet)
rate_blocked = False rate_blocked = False
if packet.context != RNS.Packet.PATH_RESPONSE and packet.receiving_interface.announce_rate_target != None: if packet.context != RNS.Packet.PATH_RESPONSE and packet.receiving_interface.announce_rate_target != None:
if not packet.destination_hash in Transport.announce_rate_table: with Transport.announce_rate_table_lock:
rate_entry = { "last": now, "rate_violations": 0, "blocked_until": 0, "timestamps": [now]} if not packet.destination_hash in Transport.announce_rate_table:
Transport.announce_rate_table[packet.destination_hash] = rate_entry rate_entry = { "last": now, "rate_violations": 0, "blocked_until": 0, "timestamps": [now]}
Transport.announce_rate_table[packet.destination_hash] = rate_entry
else:
rate_entry = Transport.announce_rate_table[packet.destination_hash]
rate_entry["timestamps"].append(now)
while len(rate_entry["timestamps"]) > Transport.MAX_RATE_TIMESTAMPS:
rate_entry["timestamps"].pop(0)
current_rate = now - rate_entry["last"]
if now > rate_entry["blocked_until"]:
if current_rate < packet.receiving_interface.announce_rate_target: rate_entry["rate_violations"] += 1
else: rate_entry["rate_violations"] = max(0, rate_entry["rate_violations"]-1)
if rate_entry["rate_violations"] > packet.receiving_interface.announce_rate_grace:
rate_target = packet.receiving_interface.announce_rate_target
rate_penalty = packet.receiving_interface.announce_rate_penalty
rate_entry["blocked_until"] = rate_entry["last"] + rate_target + rate_penalty
rate_blocked = True
else:
rate_entry["last"] = now
else: else:
rate_blocked = True rate_entry = Transport.announce_rate_table[packet.destination_hash]
rate_entry["timestamps"].append(now)
while len(rate_entry["timestamps"]) > Transport.MAX_RATE_TIMESTAMPS:
rate_entry["timestamps"].pop(0)
current_rate = now - rate_entry["last"]
if now > rate_entry["blocked_until"]:
if current_rate < packet.receiving_interface.announce_rate_target: rate_entry["rate_violations"] += 1
else: rate_entry["rate_violations"] = max(0, rate_entry["rate_violations"]-1)
if rate_entry["rate_violations"] > packet.receiving_interface.announce_rate_grace:
rate_target = packet.receiving_interface.announce_rate_target
rate_penalty = packet.receiving_interface.announce_rate_penalty
rate_entry["blocked_until"] = rate_entry["last"] + rate_target + rate_penalty
rate_blocked = True
else:
rate_entry["last"] = now
else:
rate_blocked = True
retries = 0 retries = 0
@ -1741,43 +1720,41 @@ class Transport:
retransmit_timeout = now + (RNS.rand() * Transport.PATHFINDER_RW) retransmit_timeout = now + (RNS.rand() * Transport.PATHFINDER_RW)
if hasattr(packet.receiving_interface, "mode") and packet.receiving_interface.mode == RNS.Interfaces.Interface.Interface.MODE_ACCESS_POINT: if hasattr(packet.receiving_interface, "mode") and packet.receiving_interface.mode == RNS.Interfaces.Interface.Interface.MODE_ACCESS_POINT:
expires = now + Transport.AP_PATH_TIME expires = now + Transport.AP_PATH_TIME
elif hasattr(packet.receiving_interface, "mode") and packet.receiving_interface.mode == RNS.Interfaces.Interface.Interface.MODE_ROAMING: elif hasattr(packet.receiving_interface, "mode") and packet.receiving_interface.mode == RNS.Interfaces.Interface.Interface.MODE_ROAMING:
expires = now + Transport.ROAMING_PATH_TIME expires = now + Transport.ROAMING_PATH_TIME
else: else:
expires = now + Transport.PATHFINDER_E expires = now + Transport.PATHFINDER_E
if not random_blob in random_blobs: if not random_blob in random_blobs:
random_blobs.append(random_blob) random_blobs.append(random_blob)
random_blobs = random_blobs[-Transport.MAX_RANDOM_BLOBS:] random_blobs = random_blobs[-Transport.MAX_RANDOM_BLOBS:]
if (RNS.Reticulum.transport_enabled() or Transport.from_local_client(packet)) and packet.context != RNS.Packet.PATH_RESPONSE: if (RNS.Reticulum.transport_enabled() or is_from_local_client) and packet.context != RNS.Packet.PATH_RESPONSE:
# Insert announce into announce table for retransmission # Insert announce into announce table for retransmission
if rate_blocked: if rate_blocked: RNS.log("Blocking rebroadcast of announce from "+RNS.prettyhexrep(packet.destination_hash)+" due to excessive announce rate", RNS.LOG_DEBUG)
RNS.log("Blocking rebroadcast of announce from "+RNS.prettyhexrep(packet.destination_hash)+" due to excessive announce rate", RNS.LOG_DEBUG)
else: else:
if Transport.from_local_client(packet): if is_from_local_client:
# If the announce is from a local client, # If the announce is from a local client,
# it is announced immediately, but only one time. # it is announced immediately, but only one time.
retransmit_timeout = now retransmit_timeout = now
retries = Transport.PATHFINDER_R retries = Transport.PATHFINDER_R
Transport.announce_table[packet.destination_hash] = [ with Transport.announce_table_lock:
now, # 0: IDX_AT_TIMESTAMP Transport.announce_table[packet.destination_hash] = [
retransmit_timeout, # 1: IDX_AT_RTRNS_TMO now, # 0: IDX_AT_TIMESTAMP
retries, # 2: IDX_AT_RETRIES retransmit_timeout, # 1: IDX_AT_RTRNS_TMO
received_from, # 3: IDX_AT_RCVD_IF retries, # 2: IDX_AT_RETRIES
announce_hops, # 4: IDX_AT_HOPS received_from, # 3: IDX_AT_RCVD_IF
packet, # 5: IDX_AT_PACKET announce_hops, # 4: IDX_AT_HOPS
local_rebroadcasts, # 6: IDX_AT_LCL_RBRD packet, # 5: IDX_AT_PACKET
block_rebroadcasts, # 7: IDX_AT_BLCK_RBRD local_rebroadcasts, # 6: IDX_AT_LCL_RBRD
attached_interface, # 8: IDX_AT_ATTCHD_IF block_rebroadcasts, # 7: IDX_AT_BLCK_RBRD
] attached_interface, # 8: IDX_AT_ATTCHD_IF
]
# TODO: Check from_local_client once and store result elif is_from_local_client and packet.context == RNS.Packet.PATH_RESPONSE:
elif Transport.from_local_client(packet) and packet.context == RNS.Packet.PATH_RESPONSE:
# If this is a path response from a local client, # If this is a path response from a local client,
# check if any external interfaces have pending # check if any external interfaces have pending
# path requests. # path requests.
@ -1787,17 +1764,18 @@ class Transport:
retransmit_timeout = now retransmit_timeout = now
retries = Transport.PATHFINDER_R retries = Transport.PATHFINDER_R
Transport.announce_table[packet.destination_hash] = [ with Transport.announce_table_lock:
now, Transport.announce_table[packet.destination_hash] = [
retransmit_timeout, now,
retries, retransmit_timeout,
received_from, retries,
announce_hops, received_from,
packet, announce_hops,
local_rebroadcasts, packet,
block_rebroadcasts, local_rebroadcasts,
attached_interface block_rebroadcasts,
] attached_interface
]
# If we have any local clients connected, we re- # If we have any local clients connected, we re-
# transmit the announce to them immediately # transmit the announce to them immediately
@ -1810,13 +1788,13 @@ class Transport:
announce_data = packet.data announce_data = packet.data
# TODO: Shouldn't the context be PATH_RESPONSE in the first case here? # TODO: Shouldn't the context be PATH_RESPONSE in the first case here?
if Transport.from_local_client(packet) and packet.context == RNS.Packet.PATH_RESPONSE: if is_from_local_client and packet.context == RNS.Packet.PATH_RESPONSE:
for local_interface in Transport.local_client_interfaces: for local_interface in Transport.local_client_interfaces:
if packet.receiving_interface != local_interface: if packet.receiving_interface != local_interface:
new_announce = RNS.Packet( new_announce = RNS.Packet(
announce_destination, announce_destination,
announce_data, announce_data,
RNS.Packet.ANNOUNCE, RNS.Packet.ANNOUNCE, # <-- This one?
context = announce_context, context = announce_context,
header_type = RNS.Packet.HEADER_2, header_type = RNS.Packet.HEADER_2,
transport_type = Transport.TRANSPORT, transport_type = Transport.TRANSPORT,
@ -1880,73 +1858,75 @@ class Transport:
if not Transport.owner.is_connected_to_shared_instance: Transport.cache(packet, force_cache=True, packet_type="announce") if not Transport.owner.is_connected_to_shared_instance: Transport.cache(packet, force_cache=True, packet_type="announce")
path_table_entry = [now, received_from, announce_hops, expires, random_blobs, packet.receiving_interface, packet.packet_hash] path_table_entry = [now, received_from, announce_hops, expires, random_blobs, packet.receiving_interface, packet.packet_hash]
Transport.path_table[packet.destination_hash] = path_table_entry with Transport.path_table_lock: Transport.path_table[packet.destination_hash] = path_table_entry
RNS.log("Destination "+RNS.prettyhexrep(packet.destination_hash)+" is now "+str(announce_hops)+" hops away via "+RNS.prettyhexrep(received_from)+" on "+str(packet.receiving_interface), RNS.LOG_DEBUG) RNS.log("Destination "+RNS.prettyhexrep(packet.destination_hash)+" is now "+str(announce_hops)+" hops away via "+RNS.prettyhexrep(received_from)+" on "+str(packet.receiving_interface), RNS.LOG_DEBUG)
# If the receiving interface is a tunnel, we add the # If the receiving interface is a tunnel, we add the
# announce to the tunnels table # announce to the tunnels table
if hasattr(packet.receiving_interface, "tunnel_id") and packet.receiving_interface.tunnel_id != None: if hasattr(packet.receiving_interface, "tunnel_id") and packet.receiving_interface.tunnel_id != None:
tunnel_entry = Transport.tunnels[packet.receiving_interface.tunnel_id] with Transport.tunnels_lock:
paths = tunnel_entry[IDX_TT_PATHS] tunnel_entry = Transport.tunnels[packet.receiving_interface.tunnel_id]
paths[packet.destination_hash] = [now, received_from, announce_hops, expires, random_blobs, None, packet.packet_hash] paths = tunnel_entry[IDX_TT_PATHS]
expires = time.time() + Transport.DESTINATION_TIMEOUT paths[packet.destination_hash] = [now, received_from, announce_hops, expires, random_blobs, None, packet.packet_hash]
tunnel_entry[IDX_TT_EXPIRES] = expires expires = time.time() + Transport.DESTINATION_TIMEOUT
RNS.log("Path to "+RNS.prettyhexrep(packet.destination_hash)+" associated with tunnel "+RNS.prettyhexrep(packet.receiving_interface.tunnel_id), RNS.LOG_DEBUG) tunnel_entry[IDX_TT_EXPIRES] = expires
RNS.log("Path to "+RNS.prettyhexrep(packet.destination_hash)+" associated with tunnel "+RNS.prettyhexrep(packet.receiving_interface.tunnel_id), RNS.LOG_DEBUG)
# Call externally registered callbacks from apps # Call externally registered callbacks from apps
# wanting to know when an announce arrives # wanting to know when an announce arrives
for handler in Transport.announce_handlers: with Transport.announce_handler_lock:
try: for handler in Transport.announce_handlers:
# Check that the announced destination matches try:
# the handlers aspect filter # Check that the announced destination matches
execute_callback = False # the handlers aspect filter
announce_identity = RNS.Identity.recall(packet.destination_hash) execute_callback = False
if handler.aspect_filter == None: announce_identity = RNS.Identity.recall(packet.destination_hash)
# If the handlers aspect filter is set to if handler.aspect_filter == None:
# None, we execute the callback in all cases # If the handlers aspect filter is set to
execute_callback = True # None, we execute the callback in all cases
else: execute_callback = True
handler_expected_hash = RNS.Destination.hash_from_name_and_identity(handler.aspect_filter, announce_identity)
if packet.destination_hash == handler_expected_hash: execute_callback = True
# If this is a path response, check whether the
# handler wants to receive it.
if packet.context == RNS.Packet.PATH_RESPONSE:
if hasattr(handler, "receive_path_responses") and handler.receive_path_responses == True: pass
else: execute_callback = False
if execute_callback:
if len(inspect.signature(handler.received_announce).parameters) == 3:
def job():
handler.received_announce(destination_hash=packet.destination_hash,
announced_identity=announce_identity,
app_data=RNS.Identity.recall_app_data(packet.destination_hash))
threading.Thread(target=job, daemon=True).start()
elif len(inspect.signature(handler.received_announce).parameters) == 4:
def job():
handler.received_announce(destination_hash=packet.destination_hash,
announced_identity=announce_identity,
app_data=RNS.Identity.recall_app_data(packet.destination_hash),
announce_packet_hash = packet.packet_hash)
threading.Thread(target=job, daemon=True).start()
elif len(inspect.signature(handler.received_announce).parameters) == 5:
def job():
handler.received_announce(destination_hash=packet.destination_hash,
announced_identity=announce_identity,
app_data=RNS.Identity.recall_app_data(packet.destination_hash),
announce_packet_hash = packet.packet_hash,
is_path_response = packet.context == RNS.Packet.PATH_RESPONSE)
threading.Thread(target=job, daemon=True).start()
else: else:
raise TypeError("Invalid signature for announce handler callback") handler_expected_hash = RNS.Destination.hash_from_name_and_identity(handler.aspect_filter, announce_identity)
if packet.destination_hash == handler_expected_hash: execute_callback = True
except Exception as e: # If this is a path response, check whether the
RNS.log("Error while processing external announce callback.", RNS.LOG_ERROR) # handler wants to receive it.
RNS.log("The contained exception was: "+str(e), RNS.LOG_ERROR) if packet.context == RNS.Packet.PATH_RESPONSE:
RNS.trace_exception(e) if hasattr(handler, "receive_path_responses") and handler.receive_path_responses == True: pass
else: execute_callback = False
if execute_callback:
if len(inspect.signature(handler.received_announce).parameters) == 3:
def job():
handler.received_announce(destination_hash=packet.destination_hash,
announced_identity=announce_identity,
app_data=RNS.Identity.recall_app_data(packet.destination_hash))
threading.Thread(target=job, daemon=True).start()
elif len(inspect.signature(handler.received_announce).parameters) == 4:
def job():
handler.received_announce(destination_hash=packet.destination_hash,
announced_identity=announce_identity,
app_data=RNS.Identity.recall_app_data(packet.destination_hash),
announce_packet_hash = packet.packet_hash)
threading.Thread(target=job, daemon=True).start()
elif len(inspect.signature(handler.received_announce).parameters) == 5:
def job():
handler.received_announce(destination_hash=packet.destination_hash,
announced_identity=announce_identity,
app_data=RNS.Identity.recall_app_data(packet.destination_hash),
announce_packet_hash = packet.packet_hash,
is_path_response = packet.context == RNS.Packet.PATH_RESPONSE)
threading.Thread(target=job, daemon=True).start()
else:
raise TypeError("Invalid signature for announce handler callback")
except Exception as e:
RNS.log("Error while processing external announce callback.", RNS.LOG_ERROR)
RNS.log("The contained exception was: "+str(e), RNS.LOG_ERROR)
RNS.trace_exception(e)
# Handling for link requests to local destinations # Handling for link requests to local destinations
elif packet.packet_type == RNS.Packet.LINKREQUEST: elif packet.packet_type == RNS.Packet.LINKREQUEST:
@ -1993,9 +1973,8 @@ class Transport:
RNS.Packet(destination=link, data=cached_packet.data, RNS.Packet(destination=link, data=cached_packet.data,
packet_type=cached_packet.packet_type, context=cached_packet.context).send() packet_type=cached_packet.packet_type, context=cached_packet.context).send()
Transport.jobs_locked = False else: link.receive(packet)
else:
link.receive(packet)
else: else:
# In the strange and rare case that an interface # In the strange and rare case that an interface
# is partly malfunctioning, and a link-associated # is partly malfunctioning, and a link-associated
@ -2011,14 +1990,12 @@ class Transport:
if destination.hash == packet.destination_hash and destination.type == packet.destination_type: if destination.hash == packet.destination_hash and destination.type == packet.destination_type:
packet.destination = destination packet.destination = destination
if destination.receive(packet): if destination.receive(packet):
if destination.proof_strategy == RNS.Destination.PROVE_ALL: if destination.proof_strategy == RNS.Destination.PROVE_ALL: packet.prove()
packet.prove()
elif destination.proof_strategy == RNS.Destination.PROVE_APP: elif destination.proof_strategy == RNS.Destination.PROVE_APP:
if destination.callbacks.proof_requested: if destination.callbacks.proof_requested:
try: try:
if destination.callbacks.proof_requested(packet): if destination.callbacks.proof_requested(packet): packet.prove()
packet.prove()
except Exception as e: except Exception as e:
RNS.log("Error while executing proof request callback. The contained exception was: "+str(e), RNS.LOG_ERROR) RNS.log("Error while executing proof request callback. The contained exception was: "+str(e), RNS.LOG_ERROR)
@ -2128,8 +2105,6 @@ class Transport:
if receipt in Transport.receipts: if receipt in Transport.receipts:
Transport.receipts.remove(receipt) Transport.receipts.remove(receipt)
Transport.jobs_locked = False
@staticmethod @staticmethod
def synthesize_tunnel(interface): def synthesize_tunnel(interface):
interface_hash = interface.get_hash() interface_hash = interface.get_hash()
@ -2234,11 +2209,12 @@ class Transport:
def register_destination(destination): def register_destination(destination):
destination.MTU = RNS.Reticulum.MTU destination.MTU = RNS.Reticulum.MTU
if destination.direction == RNS.Destination.IN: if destination.direction == RNS.Destination.IN:
for registered_destination in Transport.destinations: with Transport.destinations_lock:
if destination.hash == registered_destination.hash: for registered_destination in Transport.destinations:
raise KeyError("Attempt to register an already registered destination.") if destination.hash == registered_destination.hash:
raise KeyError("Attempt to register an already registered destination.")
Transport.destinations.append(destination) Transport.destinations.append(destination)
if Transport.owner.is_connected_to_shared_instance: if Transport.owner.is_connected_to_shared_instance:
if destination.type == RNS.Destination.SINGLE: if destination.type == RNS.Destination.SINGLE:
@ -2249,28 +2225,28 @@ class Transport:
@staticmethod @staticmethod
def deregister_destination(destination): def deregister_destination(destination):
if destination in Transport.destinations: with Transport.destinations_lock:
Transport.destinations.remove(destination) if destination in Transport.destinations: Transport.destinations.remove(destination)
@staticmethod @staticmethod
def register_link(link): def register_link(link):
RNS.log("Registering link "+str(link), RNS.LOG_EXTREME) RNS.log("Registering link "+str(link), RNS.LOG_EXTREME)
if link.initiator: if link.initiator:
Transport.pending_links.append(link) with Transport.pending_links_lock: Transport.pending_links.append(link)
else: else:
Transport.active_links.append(link) with Transport.active_links_lock: Transport.active_links.append(link)
@staticmethod @staticmethod
def activate_link(link): def activate_link(link):
RNS.log("Activating link "+str(link), RNS.LOG_EXTREME) RNS.log("Activating link "+str(link), RNS.LOG_EXTREME)
if link in Transport.pending_links: with Transport.pending_links_lock:
if link.status != RNS.Link.ACTIVE: if link in Transport.pending_links:
raise IOError("Invalid link state for link activation: "+str(link.status)) if link.status != RNS.Link.ACTIVE: raise IOError("Invalid link state for link activation: "+str(link.status))
Transport.pending_links.remove(link) Transport.pending_links.remove(link)
Transport.active_links.append(link) with Transport.active_links_lock: Transport.active_links.append(link)
link.status = RNS.Link.ACTIVE link.status = RNS.Link.ACTIVE
else: else:
RNS.log("Attempted to activate a link that was not in the pending table", RNS.LOG_ERROR) RNS.log("Attempted to activate a link that was not in the pending table", RNS.LOG_ERROR)
@staticmethod @staticmethod
def register_announce_handler(handler): def register_announce_handler(handler):
@ -2283,9 +2259,9 @@ class Transport:
optionally have a *receive_path_responses* attribute set to ``True``, to also receive all path responses, in addition to live optionally have a *receive_path_responses* attribute set to ``True``, to also receive all path responses, in addition to live
announces. See the :ref:`Announce Example<example-announce>` for more info. announces. See the :ref:`Announce Example<example-announce>` for more info.
""" """
if hasattr(handler, "received_announce") and callable(handler.received_announce): with Transport.announce_handler_lock:
if hasattr(handler, "aspect_filter"): if hasattr(handler, "received_announce") and callable(handler.received_announce):
Transport.announce_handlers.append(handler) if hasattr(handler, "aspect_filter"): Transport.announce_handlers.append(handler)
@staticmethod @staticmethod
def deregister_announce_handler(handler): def deregister_announce_handler(handler):
@ -2294,7 +2270,9 @@ class Transport:
:param handler: The announce handler to be deregistered. :param handler: The announce handler to be deregistered.
""" """
while handler in Transport.announce_handlers: Transport.announce_handlers.remove(handler) with Transport.announce_handler_lock:
while handler in Transport.announce_handlers: Transport.announce_handlers.remove(handler)
gc.collect() gc.collect()
@staticmethod @staticmethod
@ -2434,8 +2412,9 @@ class Transport:
:param destination_hash: A destination hash as *bytes*. :param destination_hash: A destination hash as *bytes*.
:returns: The number of hops to the specified destination, or ``RNS.Transport.PATHFINDER_M`` if the number of hops is unknown. :returns: The number of hops to the specified destination, or ``RNS.Transport.PATHFINDER_M`` if the number of hops is unknown.
""" """
if destination_hash in Transport.path_table: return Transport.path_table[destination_hash][IDX_PT_HOPS] with Transport.path_table_lock:
else: return Transport.PATHFINDER_M if destination_hash in Transport.path_table: return Transport.path_table[destination_hash][IDX_PT_HOPS]
else: return Transport.PATHFINDER_M
@staticmethod @staticmethod
def next_hop(destination_hash): def next_hop(destination_hash):
@ -2443,8 +2422,9 @@ class Transport:
:param destination_hash: A destination hash as *bytes*. :param destination_hash: A destination hash as *bytes*.
:returns: The destination hash as *bytes* for the next hop to the specified destination, or *None* if the next hop is unknown. :returns: The destination hash as *bytes* for the next hop to the specified destination, or *None* if the next hop is unknown.
""" """
if destination_hash in Transport.path_table: return Transport.path_table[destination_hash][IDX_PT_NEXT_HOP] with Transport.path_table_lock:
else: return None if destination_hash in Transport.path_table: return Transport.path_table[destination_hash][IDX_PT_NEXT_HOP]
else: return None
@staticmethod @staticmethod
def next_hop_interface(destination_hash): def next_hop_interface(destination_hash):
@ -2452,8 +2432,9 @@ class Transport:
:param destination_hash: A destination hash as *bytes*. :param destination_hash: A destination hash as *bytes*.
:returns: The interface for the next hop to the specified destination, or *None* if the interface is unknown. :returns: The interface for the next hop to the specified destination, or *None* if the interface is unknown.
""" """
if destination_hash in Transport.path_table: return Transport.path_table[destination_hash][IDX_PT_RVCD_IF] with Transport.path_table_lock:
else: return None if destination_hash in Transport.path_table: return Transport.path_table[destination_hash][IDX_PT_RVCD_IF]
else: return None
@staticmethod @staticmethod
def next_hop_interface_bitrate(destination_hash): def next_hop_interface_bitrate(destination_hash):

2
RNS/_version.py
View file

@ -1 +1 @@
__version__ = "1.1.4" __version__ = "1.1.5"

2
tests/link.py
View file

@ -768,7 +768,7 @@ class TestLink(unittest.TestCase):
data = bytearray() data = bytearray()
for rx in received: for rx in received:
data.extend(rx) if rx: data.extend(rx)
rx_message = data rx_message = data
print(f"Received {len(received)} chunks, totalling {len(rx_message)} bytes") print(f"Received {len(received)} chunks, totalling {len(rx_message)} bytes")