diff --git a/LXMF/LXMF.py b/LXMF/LXMF.py index cfe0d97..3612902 100644 --- a/LXMF/LXMF.py +++ b/LXMF/LXMF.py @@ -7,460 +7,460 @@ import RNS.vendor.umsgpack as msgpack APP_NAME = "lxmf" class LXMessage: - DRAFT = 0x00 - OUTBOUND = 0x01 - SENDING = 0x02 - SENT = 0x04 - DELIVERED = 0x08 - FAILED = 0xFF - states = [DRAFT, OUTBOUND, SENDING, SENT, DELIVERED, FAILED] - - UNKNOWN = 0x00 - PACKET = 0x01 - RESOURCE = 0x02 - representations = [UNKNOWN, PACKET, RESOURCE] - - OPPORTUNISTIC = 0x01 - DIRECT = 0x02 - PROPAGATED = 0x03 - valid_methods = [OPPORTUNISTIC, DIRECT, PROPAGATED] - - SOURCE_UNKNOWN = 0x01 - SIGNATURE_INVALID = 0x02 - unverified_reasons = [SOURCE_UNKNOWN, SIGNATURE_INVALID] - - DESTINATION_LENGTH = RNS.Identity.TRUNCATED_HASHLENGTH//8 - SIGNATURE_LENGTH = RNS.Identity.SIGLENGTH//8 - - # LXMF overhead is 99 bytes per message: - # 10 bytes for destination hash - # 10 bytes for source hash - # 64 bytes for Ed25519 signature - # 8 bytes for timestamp - # 7 bytes for msgpack structure - LXMF_OVERHEAD = 2*DESTINATION_LENGTH + SIGNATURE_LENGTH + 8 + 7 - - # With an MTU of 500, the maximum amount of data - # we can send in a single encrypted packet is - # 383 bytes. - ENCRYPTED_PACKET_MDU = RNS.Packet.ENCRYPTED_MDU - - # The max content length we can fit in LXMF message - # inside a single RNS packet is the encrypted MDU, minus - # the LXMF overhead. We can optimise a bit though, by - # inferring the destination hash from the destination - # field of the packet, therefore we also add the length - # of a destination hash to the calculation. With default - # RNS and LXMF parameters, the largest single-packet - # LXMF message we can send is 294 bytes. If a message - # is larger than that, a Reticulum link will be used. - ENCRYPTED_PACKET_MAX_CONTENT = ENCRYPTED_PACKET_MDU - LXMF_OVERHEAD + DESTINATION_LENGTH - - # Links can carry a larger MDU, due to less overhead per - # packet. The link MDU with default Reticulum parameters - # is 415 bytes. - LINK_PACKET_MDU = RNS.Link.MDU - - # Which means that we can deliver single-packet LXMF - # messages with content of up to 316 bytes over a link. - # If a message is larger than that, LXMF will sequence - # and transfer it as a RNS resource over the link instead. - LINK_PACKET_MAX_CONTENT = LINK_PACKET_MDU - LXMF_OVERHEAD - - # For plain packets without encryption, we can - # fit up to 324 bytes of content. - PLAIN_PACKET_MDU = RNS.Packet.PLAIN_MDU - PLAIN_PACKET_MAX_CONTENT = PLAIN_PACKET_MDU - LXMF_OVERHEAD + DESTINATION_LENGTH - - # Descriptive strings regarding transport encryption - ENCRYPTION_DESCRIPTION_AES = "AES-128" - ENCRYPTION_DESCRIPTION_EC = "Curve25519" - ENCRYPTION_DESCRIPTION_UNENCRYPTED = "Unencrypted" - - def __str__(self): - if self.hash != None: - return "" - else: - return "" - - def __init__(self, destination, source, content = "", title = "", fields = None, desired_method = None, destination_hash = None, source_hash = None): - if isinstance(destination, RNS.Destination) or destination == None: - self.__destination = destination - if destination != None: - self.destination_hash = destination.hash - else: - self.destination_hash = destination_hash - else: - raise ValueError("LXMessage initialised with invalid destination") - - if isinstance(source, RNS.Destination) or source == None: - self.__source = source - if source != None: - self.source_hash = source.hash - else: - self.source_hash = source_hash - else: - raise ValueError("LXMessage initialised with invalid source") - - self.set_title_from_string(title) - self.set_content_from_string(content) - - self.set_fields(fields) - - self.payload = None - self.timestamp = None - self.signature = None - self.hash = None - self.packed = None - self.progress = None - self.state = LXMessage.DRAFT - self.method = LXMessage.UNKNOWN - - self.incoming = False - self.signature_validated = False - self.unverified_reason = None - - self.representation = LXMessage.UNKNOWN - self.desired_method = desired_method - self.delivery_attempts = 0 - self.transport_encrypted = False - self.transport_encryption = None - self.packet_representation = None - self.resource_representation = None - self.__delivery_destination = None - self.__delivery_callback = None - self.failed_callback = None - - def set_title_from_string(self, title_string): - self.title = title_string.encode("utf-8") - - def set_title_from_bytes(self, title_bytes): - self.title = title_bytes - - def title_as_string(self): - return self.title.decode("utf-8") - - def set_content_from_string(self, content_string): - self.content = content_string.encode("utf-8") - - def set_content_from_bytes(self, content_bytes): - self.content = content_bytes - - def content_as_string(self): - return self.content.decode("utf-8") - - def set_fields(self, fields): - if isinstance(fields, dict) or fields == None: - self.fields = fields - else: - raise ValueError("LXMessage property \"fields\" can only be dict or None") - - def get_fields(self): - return self.__fields - - def set_destination(self, destination): - if self.destination == None: - if isinstance(destination, RNS.Destination): - self.__destination = destination - else: - raise ValueError("Invalid destination set on LXMessage") - else: - raise ValueError("Cannot reassign destination on LXMessage") - - def get_destination(self): - return self.__destination - - def set_source(self, source): - if self.source == None: - if isinstance(source, RNS.Destination): - self.__source = source - else: - raise ValueError("Invalid source set on LXMessage") - else: - raise ValueError("Cannot reassign source on LXMessage") - - def get_source(self): - return self.__source - - def set_delivery_destination(self, delivery_destination): - self.__delivery_destination = delivery_destination - - def register_delivery_callback(self, callback): - self.__delivery_callback = callback - - def register_failed_callback(self, callback): - self.failed_callback = callback - - def pack(self): - if not self.packed: - if self.timestamp == None: - self.timestamp = time.time() - - self.payload = [self.timestamp, self.title, self.content, self.fields] - - hashed_part = b"" - hashed_part += self.__destination.hash - hashed_part += self.__source.hash - hashed_part += msgpack.packb(self.payload) - self.hash = RNS.Identity.full_hash(hashed_part) - self.message_id = self.hash - - signed_part = b"" - signed_part += hashed_part - signed_part += self.hash - self.signature = self.__source.sign(signed_part) - self.signature_validated = True - - self.packed = b"" - self.packed += self.__destination.hash - self.packed += self.__source.hash - self.packed += self.signature - packed_payload = msgpack.packb(self.payload) - self.packed += packed_payload - self.packed_size = len(self.packed) - content_size = len(packed_payload) - - # If no desired delivery method has been defined, - # one will be chosen according to these rules: - if self.desired_method == None: - self.desired_method == LXMessage.DIRECT - # TODO: Expand rules to something more intelligent - - if self.desired_method == LXMessage.OPPORTUNISTIC: - if self.__destination.type == RNS.Destination.SINGLE: - single_packet_content_limit = LXMessage.ENCRYPTED_PACKET_MAX_CONTENT - elif self.__destination.type == RNS.Destination.PLAIN: - single_packet_content_limit = LXMessage.PLAIN_PACKET_MAX_CONTENT - - if content_size > single_packet_content_limit: - raise TypeError("LXMessage desired opportunistic delivery method, but content exceeds single-packet size.") - else: - self.method = LXMessage.OPPORTUNISTIC - self.representation = LXMessage.PACKET - self.__delivery_destination = self.__destination - - elif self.desired_method == LXMessage.DIRECT or self.desired_method == LXMessage.PROPAGATED: - single_packet_content_limit = LXMessage.LINK_PACKET_MAX_CONTENT - if content_size <= single_packet_content_limit: - self.method = self.desired_method - self.representation = LXMessage.PACKET - else: - self.method = self.desired_method - self.representation = LXMessage.RESOURCE - else: - raise ValueError("Attempt to re-pack LXMessage "+str(self)+" that was already packed") - - def send(self): - self.determine_transport_encryption() - - if self.method == LXMessage.OPPORTUNISTIC: - self.__as_packet().send().set_delivery_callback(self.__mark_delivered) - self.state = LXMessage.SENT - - elif self.method == LXMessage.DIRECT: - self.state = LXMessage.SENDING - - if self.representation == LXMessage.PACKET: - receipt = self.__as_packet().send() - receipt.set_delivery_callback(self.__mark_delivered) - receipt.set_timeout_callback(self.__link_packet_timed_out) - - elif self.representation == LXMessage.RESOURCE: - self.resource_representation = self.__as_resource() - - elif self.method == LXMessage.PROPAGATED: - # TODO: Implement propagation - pass - - def determine_transport_encryption(self): - if RNS.Reticulum.should_allow_unencrypted(): - self.transport_encrypted = False - self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_UNENCRYPTED - else: - if self.method == LXMessage.OPPORTUNISTIC: - if self.destination.type == RNS.Destination.SINGLE: - self.transport_encrypted = True - self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_EC - elif destination_type == RNS.Destination.GROUP: - self.transport_encrypted = True - self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_AES - else: - self.transport_encrypted = False - self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_UNENCRYPTED - elif self.method == LXMessage.DIRECT: - self.transport_encrypted = True - self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_EC - elif self.method == LXMessage.PROPAGATED: - if self.destination.type == RNS.Destination.SINGLE: - self.transport_encrypted = True - self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_EC - elif destination_type == RNS.Destination.GROUP: - self.transport_encrypted = True - self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_AES - else: - self.transport_encrypted = False - self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_UNENCRYPTED - else: - self.transport_encrypted = False - self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_UNENCRYPTED - - def __mark_delivered(self, receipt = None): - RNS.log("Received delivery notification for "+str(self), RNS.LOG_DEBUG) - self.state = LXMessage.DELIVERED - - if self.__delivery_callback != None: - self.__delivery_callback(self) - - def __resource_concluded(self, resource): - if resource.status == RNS.Resource.COMPLETE: - self.__mark_delivered() - else: - resource.link.teardown() - self.state = LXMessage.OUTBOUND - - def __link_packet_timed_out(self, packet_receipt): - packet_receipt.destination.teardown() - self.state = LXMessage.OUTBOUND - - - def __update_transfer_progress(self, resource): - self.progress = resource.progress() - - def __as_packet(self): - if not self.packed: - self.pack() - - if not self.__delivery_destination: - raise ValueError("Can't synthesize packet for LXMF message before delivery destination is known") - - if self.method == LXMessage.OPPORTUNISTIC: - return RNS.Packet(self.__delivery_destination, self.packed[LXMessage.DESTINATION_LENGTH:]) - elif self.method == LXMessage.DIRECT or self.method == LXMessage.PROPAGATED: - return RNS.Packet(self.__delivery_destination, self.packed) - - def __as_resource(self): - if not self.packed: - self.pack() - - if not self.__delivery_destination: - raise ValueError("Can't synthesize resource for LXMF message before delivery destination is known") - - if not self.__delivery_destination.type == RNS.Destination.LINK: - raise TypeError("Tried to synthesize resource for LXMF message on a delivery destination that was not a link") - - if not self.__delivery_destination.status == RNS.Link.ACTIVE: - raise ConnectionError("Tried to synthesize resource for LXMF message on a link that was not active") - - self.progress = 0.0 - return RNS.Resource(self.packed, self.__delivery_destination, callback = self.__resource_concluded, progress_callback = self.__update_transfer_progress) - - def write_to_directory(self, directory_path): - file_name = RNS.hexrep(self.hash, delimit=False) - file_path = directory_path+"/"+file_name - - try: - if not self.packed: - self.pack() - - container = { - "state": self.state, - "lxmf_bytes": self.packed, - "transport_encrypted": self.transport_encrypted, - "transport_encryption": self.transport_encryption - } - - packed_container = msgpack.packb(container) - - file = open(file_path, "wb") - file.write(packed_container) - file.close() - - return file_path - - except Exception as e: - RNS.log("Error while writing LXMF message to file \""+str(file_path)+"\". The contained exception was: "+str(e), RNS.LOG_ERROR) - return None - - - @staticmethod - def unpack_from_bytes(lxmf_bytes): - destination_hash = lxmf_bytes[:LXMessage.DESTINATION_LENGTH] - source_hash = lxmf_bytes[LXMessage.DESTINATION_LENGTH:2*LXMessage.DESTINATION_LENGTH] - signature = lxmf_bytes[2*LXMessage.DESTINATION_LENGTH:2*LXMessage.DESTINATION_LENGTH+LXMessage.SIGNATURE_LENGTH] - packed_payload = lxmf_bytes[2*LXMessage.DESTINATION_LENGTH+LXMessage.SIGNATURE_LENGTH:] - hashed_part = b"" + destination_hash + source_hash + packed_payload - message_hash = RNS.Identity.full_hash(hashed_part) - signed_part = b"" + hashed_part + message_hash - unpacked_payload = msgpack.unpackb(packed_payload) - timestamp = unpacked_payload[0] - title_bytes = unpacked_payload[1] - content_bytes = unpacked_payload[2] - fields = unpacked_payload[3] - - destination_identity = RNS.Identity.recall(destination_hash) - if destination_identity != None: - destination = RNS.Destination(destination_identity, RNS.Destination.OUT, RNS.Destination.SINGLE, APP_NAME, "delivery") - else: - destination = None - - source_identity = RNS.Identity.recall(source_hash) - if source_identity != None: - source = RNS.Destination(source_identity, RNS.Destination.OUT, RNS.Destination.SINGLE, APP_NAME, "delivery") - else: - source = None - - message = LXMessage( - destination = destination, - source = source, - content = "", - title = "", - fields = fields, - destination_hash = destination_hash, - source_hash = source_hash) - - message.hash = message_hash - message.signature = signature - message.incoming = True - message.timestamp = timestamp - message.packed = lxmf_bytes - message.packed_size = len(lxmf_bytes) - message.set_title_from_bytes(title_bytes) - message.set_content_from_bytes(content_bytes) - - try: - if source: - if source.identity.validate(signature, signed_part): - message.signature_validated = True - else: - message.signature_validated = False - message.unverified_reason = LXMessage.SIGNATURE_INVALID - else: - signature_validated = False - message.unverified_reason = LXMessage.SOURCE_UNKNOWN - RNS.log("Unpacked LXMF message signature could not be validated, since source identity is unknown", RNS.LOG_DEBUG) - except Exception as e: - message.signature_validated = False - RNS.log("Error while validating LXMF message signature. The contained exception was: "+str(e), RNS.LOG_ERROR) - - return message - - @staticmethod - def unpack_from_file(lxmf_file_handle): - try: - container = msgpack.unpackb(lxmf_file_handle.read()) - lxm = LXMessage.unpack_from_bytes(container["lxmf_bytes"]) - - if "state" in container: - lxm.state = container["state"] - if "transport_encrypted" in container: - lxm.transport_encrypted = container["transport_encrypted"] - if "transport_encryption" in container: - lxm.transport_encryption = container["transport_encryption"] - - return lxm - except Exception as e: - RNS.log("Could not unpack LXMessage from file. The contained exception was: "+str(e), RNS.LOG_ERROR) - return None + DRAFT = 0x00 + OUTBOUND = 0x01 + SENDING = 0x02 + SENT = 0x04 + DELIVERED = 0x08 + FAILED = 0xFF + states = [DRAFT, OUTBOUND, SENDING, SENT, DELIVERED, FAILED] + + UNKNOWN = 0x00 + PACKET = 0x01 + RESOURCE = 0x02 + representations = [UNKNOWN, PACKET, RESOURCE] + + OPPORTUNISTIC = 0x01 + DIRECT = 0x02 + PROPAGATED = 0x03 + valid_methods = [OPPORTUNISTIC, DIRECT, PROPAGATED] + + SOURCE_UNKNOWN = 0x01 + SIGNATURE_INVALID = 0x02 + unverified_reasons = [SOURCE_UNKNOWN, SIGNATURE_INVALID] + + DESTINATION_LENGTH = RNS.Identity.TRUNCATED_HASHLENGTH//8 + SIGNATURE_LENGTH = RNS.Identity.SIGLENGTH//8 + + # LXMF overhead is 99 bytes per message: + # 10 bytes for destination hash + # 10 bytes for source hash + # 64 bytes for Ed25519 signature + # 8 bytes for timestamp + # 7 bytes for msgpack structure + LXMF_OVERHEAD = 2*DESTINATION_LENGTH + SIGNATURE_LENGTH + 8 + 7 + + # With an MTU of 500, the maximum amount of data + # we can send in a single encrypted packet is + # 383 bytes. + ENCRYPTED_PACKET_MDU = RNS.Packet.ENCRYPTED_MDU + + # The max content length we can fit in LXMF message + # inside a single RNS packet is the encrypted MDU, minus + # the LXMF overhead. We can optimise a bit though, by + # inferring the destination hash from the destination + # field of the packet, therefore we also add the length + # of a destination hash to the calculation. With default + # RNS and LXMF parameters, the largest single-packet + # LXMF message we can send is 294 bytes. If a message + # is larger than that, a Reticulum link will be used. + ENCRYPTED_PACKET_MAX_CONTENT = ENCRYPTED_PACKET_MDU - LXMF_OVERHEAD + DESTINATION_LENGTH + + # Links can carry a larger MDU, due to less overhead per + # packet. The link MDU with default Reticulum parameters + # is 415 bytes. + LINK_PACKET_MDU = RNS.Link.MDU + + # Which means that we can deliver single-packet LXMF + # messages with content of up to 316 bytes over a link. + # If a message is larger than that, LXMF will sequence + # and transfer it as a RNS resource over the link instead. + LINK_PACKET_MAX_CONTENT = LINK_PACKET_MDU - LXMF_OVERHEAD + + # For plain packets without encryption, we can + # fit up to 324 bytes of content. + PLAIN_PACKET_MDU = RNS.Packet.PLAIN_MDU + PLAIN_PACKET_MAX_CONTENT = PLAIN_PACKET_MDU - LXMF_OVERHEAD + DESTINATION_LENGTH + + # Descriptive strings regarding transport encryption + ENCRYPTION_DESCRIPTION_AES = "AES-128" + ENCRYPTION_DESCRIPTION_EC = "Curve25519" + ENCRYPTION_DESCRIPTION_UNENCRYPTED = "Unencrypted" + + def __str__(self): + if self.hash != None: + return "" + else: + return "" + + def __init__(self, destination, source, content = "", title = "", fields = None, desired_method = None, destination_hash = None, source_hash = None): + if isinstance(destination, RNS.Destination) or destination == None: + self.__destination = destination + if destination != None: + self.destination_hash = destination.hash + else: + self.destination_hash = destination_hash + else: + raise ValueError("LXMessage initialised with invalid destination") + + if isinstance(source, RNS.Destination) or source == None: + self.__source = source + if source != None: + self.source_hash = source.hash + else: + self.source_hash = source_hash + else: + raise ValueError("LXMessage initialised with invalid source") + + self.set_title_from_string(title) + self.set_content_from_string(content) + + self.set_fields(fields) + + self.payload = None + self.timestamp = None + self.signature = None + self.hash = None + self.packed = None + self.progress = None + self.state = LXMessage.DRAFT + self.method = LXMessage.UNKNOWN + + self.incoming = False + self.signature_validated = False + self.unverified_reason = None + + self.representation = LXMessage.UNKNOWN + self.desired_method = desired_method + self.delivery_attempts = 0 + self.transport_encrypted = False + self.transport_encryption = None + self.packet_representation = None + self.resource_representation = None + self.__delivery_destination = None + self.__delivery_callback = None + self.failed_callback = None + + def set_title_from_string(self, title_string): + self.title = title_string.encode("utf-8") + + def set_title_from_bytes(self, title_bytes): + self.title = title_bytes + + def title_as_string(self): + return self.title.decode("utf-8") + + def set_content_from_string(self, content_string): + self.content = content_string.encode("utf-8") + + def set_content_from_bytes(self, content_bytes): + self.content = content_bytes + + def content_as_string(self): + return self.content.decode("utf-8") + + def set_fields(self, fields): + if isinstance(fields, dict) or fields == None: + self.fields = fields + else: + raise ValueError("LXMessage property \"fields\" can only be dict or None") + + def get_fields(self): + return self.__fields + + def set_destination(self, destination): + if self.destination == None: + if isinstance(destination, RNS.Destination): + self.__destination = destination + else: + raise ValueError("Invalid destination set on LXMessage") + else: + raise ValueError("Cannot reassign destination on LXMessage") + + def get_destination(self): + return self.__destination + + def set_source(self, source): + if self.source == None: + if isinstance(source, RNS.Destination): + self.__source = source + else: + raise ValueError("Invalid source set on LXMessage") + else: + raise ValueError("Cannot reassign source on LXMessage") + + def get_source(self): + return self.__source + + def set_delivery_destination(self, delivery_destination): + self.__delivery_destination = delivery_destination + + def register_delivery_callback(self, callback): + self.__delivery_callback = callback + + def register_failed_callback(self, callback): + self.failed_callback = callback + + def pack(self): + if not self.packed: + if self.timestamp == None: + self.timestamp = time.time() + + self.payload = [self.timestamp, self.title, self.content, self.fields] + + hashed_part = b"" + hashed_part += self.__destination.hash + hashed_part += self.__source.hash + hashed_part += msgpack.packb(self.payload) + self.hash = RNS.Identity.full_hash(hashed_part) + self.message_id = self.hash + + signed_part = b"" + signed_part += hashed_part + signed_part += self.hash + self.signature = self.__source.sign(signed_part) + self.signature_validated = True + + self.packed = b"" + self.packed += self.__destination.hash + self.packed += self.__source.hash + self.packed += self.signature + packed_payload = msgpack.packb(self.payload) + self.packed += packed_payload + self.packed_size = len(self.packed) + content_size = len(packed_payload) + + # If no desired delivery method has been defined, + # one will be chosen according to these rules: + if self.desired_method == None: + self.desired_method == LXMessage.DIRECT + # TODO: Expand rules to something more intelligent + + if self.desired_method == LXMessage.OPPORTUNISTIC: + if self.__destination.type == RNS.Destination.SINGLE: + single_packet_content_limit = LXMessage.ENCRYPTED_PACKET_MAX_CONTENT + elif self.__destination.type == RNS.Destination.PLAIN: + single_packet_content_limit = LXMessage.PLAIN_PACKET_MAX_CONTENT + + if content_size > single_packet_content_limit: + raise TypeError("LXMessage desired opportunistic delivery method, but content exceeds single-packet size.") + else: + self.method = LXMessage.OPPORTUNISTIC + self.representation = LXMessage.PACKET + self.__delivery_destination = self.__destination + + elif self.desired_method == LXMessage.DIRECT or self.desired_method == LXMessage.PROPAGATED: + single_packet_content_limit = LXMessage.LINK_PACKET_MAX_CONTENT + if content_size <= single_packet_content_limit: + self.method = self.desired_method + self.representation = LXMessage.PACKET + else: + self.method = self.desired_method + self.representation = LXMessage.RESOURCE + else: + raise ValueError("Attempt to re-pack LXMessage "+str(self)+" that was already packed") + + def send(self): + self.determine_transport_encryption() + + if self.method == LXMessage.OPPORTUNISTIC: + self.__as_packet().send().set_delivery_callback(self.__mark_delivered) + self.state = LXMessage.SENT + + elif self.method == LXMessage.DIRECT: + self.state = LXMessage.SENDING + + if self.representation == LXMessage.PACKET: + receipt = self.__as_packet().send() + receipt.set_delivery_callback(self.__mark_delivered) + receipt.set_timeout_callback(self.__link_packet_timed_out) + + elif self.representation == LXMessage.RESOURCE: + self.resource_representation = self.__as_resource() + + elif self.method == LXMessage.PROPAGATED: + # TODO: Implement propagation + pass + + def determine_transport_encryption(self): + if RNS.Reticulum.should_allow_unencrypted(): + self.transport_encrypted = False + self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_UNENCRYPTED + else: + if self.method == LXMessage.OPPORTUNISTIC: + if self.destination.type == RNS.Destination.SINGLE: + self.transport_encrypted = True + self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_EC + elif destination_type == RNS.Destination.GROUP: + self.transport_encrypted = True + self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_AES + else: + self.transport_encrypted = False + self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_UNENCRYPTED + elif self.method == LXMessage.DIRECT: + self.transport_encrypted = True + self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_EC + elif self.method == LXMessage.PROPAGATED: + if self.destination.type == RNS.Destination.SINGLE: + self.transport_encrypted = True + self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_EC + elif destination_type == RNS.Destination.GROUP: + self.transport_encrypted = True + self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_AES + else: + self.transport_encrypted = False + self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_UNENCRYPTED + else: + self.transport_encrypted = False + self.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_UNENCRYPTED + + def __mark_delivered(self, receipt = None): + RNS.log("Received delivery notification for "+str(self), RNS.LOG_DEBUG) + self.state = LXMessage.DELIVERED + + if self.__delivery_callback != None: + self.__delivery_callback(self) + + def __resource_concluded(self, resource): + if resource.status == RNS.Resource.COMPLETE: + self.__mark_delivered() + else: + resource.link.teardown() + self.state = LXMessage.OUTBOUND + + def __link_packet_timed_out(self, packet_receipt): + packet_receipt.destination.teardown() + self.state = LXMessage.OUTBOUND + + + def __update_transfer_progress(self, resource): + self.progress = resource.progress() + + def __as_packet(self): + if not self.packed: + self.pack() + + if not self.__delivery_destination: + raise ValueError("Can't synthesize packet for LXMF message before delivery destination is known") + + if self.method == LXMessage.OPPORTUNISTIC: + return RNS.Packet(self.__delivery_destination, self.packed[LXMessage.DESTINATION_LENGTH:]) + elif self.method == LXMessage.DIRECT or self.method == LXMessage.PROPAGATED: + return RNS.Packet(self.__delivery_destination, self.packed) + + def __as_resource(self): + if not self.packed: + self.pack() + + if not self.__delivery_destination: + raise ValueError("Can't synthesize resource for LXMF message before delivery destination is known") + + if not self.__delivery_destination.type == RNS.Destination.LINK: + raise TypeError("Tried to synthesize resource for LXMF message on a delivery destination that was not a link") + + if not self.__delivery_destination.status == RNS.Link.ACTIVE: + raise ConnectionError("Tried to synthesize resource for LXMF message on a link that was not active") + + self.progress = 0.0 + return RNS.Resource(self.packed, self.__delivery_destination, callback = self.__resource_concluded, progress_callback = self.__update_transfer_progress) + + def write_to_directory(self, directory_path): + file_name = RNS.hexrep(self.hash, delimit=False) + file_path = directory_path+"/"+file_name + + try: + if not self.packed: + self.pack() + + container = { + "state": self.state, + "lxmf_bytes": self.packed, + "transport_encrypted": self.transport_encrypted, + "transport_encryption": self.transport_encryption + } + + packed_container = msgpack.packb(container) + + file = open(file_path, "wb") + file.write(packed_container) + file.close() + + return file_path + + except Exception as e: + RNS.log("Error while writing LXMF message to file \""+str(file_path)+"\". The contained exception was: "+str(e), RNS.LOG_ERROR) + return None + + + @staticmethod + def unpack_from_bytes(lxmf_bytes): + destination_hash = lxmf_bytes[:LXMessage.DESTINATION_LENGTH] + source_hash = lxmf_bytes[LXMessage.DESTINATION_LENGTH:2*LXMessage.DESTINATION_LENGTH] + signature = lxmf_bytes[2*LXMessage.DESTINATION_LENGTH:2*LXMessage.DESTINATION_LENGTH+LXMessage.SIGNATURE_LENGTH] + packed_payload = lxmf_bytes[2*LXMessage.DESTINATION_LENGTH+LXMessage.SIGNATURE_LENGTH:] + hashed_part = b"" + destination_hash + source_hash + packed_payload + message_hash = RNS.Identity.full_hash(hashed_part) + signed_part = b"" + hashed_part + message_hash + unpacked_payload = msgpack.unpackb(packed_payload) + timestamp = unpacked_payload[0] + title_bytes = unpacked_payload[1] + content_bytes = unpacked_payload[2] + fields = unpacked_payload[3] + + destination_identity = RNS.Identity.recall(destination_hash) + if destination_identity != None: + destination = RNS.Destination(destination_identity, RNS.Destination.OUT, RNS.Destination.SINGLE, APP_NAME, "delivery") + else: + destination = None + + source_identity = RNS.Identity.recall(source_hash) + if source_identity != None: + source = RNS.Destination(source_identity, RNS.Destination.OUT, RNS.Destination.SINGLE, APP_NAME, "delivery") + else: + source = None + + message = LXMessage( + destination = destination, + source = source, + content = "", + title = "", + fields = fields, + destination_hash = destination_hash, + source_hash = source_hash) + + message.hash = message_hash + message.signature = signature + message.incoming = True + message.timestamp = timestamp + message.packed = lxmf_bytes + message.packed_size = len(lxmf_bytes) + message.set_title_from_bytes(title_bytes) + message.set_content_from_bytes(content_bytes) + + try: + if source: + if source.identity.validate(signature, signed_part): + message.signature_validated = True + else: + message.signature_validated = False + message.unverified_reason = LXMessage.SIGNATURE_INVALID + else: + signature_validated = False + message.unverified_reason = LXMessage.SOURCE_UNKNOWN + RNS.log("Unpacked LXMF message signature could not be validated, since source identity is unknown", RNS.LOG_DEBUG) + except Exception as e: + message.signature_validated = False + RNS.log("Error while validating LXMF message signature. The contained exception was: "+str(e), RNS.LOG_ERROR) + + return message + + @staticmethod + def unpack_from_file(lxmf_file_handle): + try: + container = msgpack.unpackb(lxmf_file_handle.read()) + lxm = LXMessage.unpack_from_bytes(container["lxmf_bytes"]) + + if "state" in container: + lxm.state = container["state"] + if "transport_encrypted" in container: + lxm.transport_encrypted = container["transport_encrypted"] + if "transport_encryption" in container: + lxm.transport_encryption = container["transport_encryption"] + + return lxm + except Exception as e: + RNS.log("Could not unpack LXMessage from file. The contained exception was: "+str(e), RNS.LOG_ERROR) + return None class LXMFDeliveryAnnounceHandler: @@ -470,253 +470,253 @@ class LXMFDeliveryAnnounceHandler: def received_announce(self, destination_hash, announced_identity, app_data): for lxmessage in self.lxmrouter.pending_outbound: - if destination_hash == lxmessage.destination_hash: - if lxmessage.method == LXMessage.DIRECT: - lxmessage.next_delivery_attempt = time.time() + if destination_hash == lxmessage.destination_hash: + if lxmessage.method == LXMessage.DIRECT: + lxmessage.next_delivery_attempt = time.time() - while self.lxmrouter.processing_outbound: - time.sleep(0.1) + while self.lxmrouter.processing_outbound: + time.sleep(0.1) - self.lxmrouter.process_outbound() + self.lxmrouter.process_outbound() class LXMRouter: - MAX_DELIVERY_ATTEMPTS = 3 - PROCESSING_INTERVAL = 5 - DELIVERY_RETRY_WAIT = 15 - PATH_REQUEST_WAIT = 5 - LINK_MAX_INACTIVITY = 10*60 + MAX_DELIVERY_ATTEMPTS = 3 + PROCESSING_INTERVAL = 5 + DELIVERY_RETRY_WAIT = 15 + PATH_REQUEST_WAIT = 5 + LINK_MAX_INACTIVITY = 10*60 - def __init__(self): - self.pending_inbound = [] - self.pending_outbound = [] - self.failed_outbound = [] - self.direct_links = {} - self.delivery_destinations = {} + def __init__(self): + self.pending_inbound = [] + self.pending_outbound = [] + self.failed_outbound = [] + self.direct_links = {} + self.delivery_destinations = {} - self.processing_outbound = False - self.processing_inbound = False + self.processing_outbound = False + self.processing_inbound = False - self.identity = RNS.Identity() - self.lxmf_query_destination = RNS.Destination(None, RNS.Destination.IN, RNS.Destination.PLAIN, APP_NAME, "query") - self.propagation_destination = RNS.Destination(self.identity, RNS.Destination.IN, RNS.Destination.SINGLE, APP_NAME, "propagation") + self.identity = RNS.Identity() + self.lxmf_query_destination = RNS.Destination(None, RNS.Destination.IN, RNS.Destination.PLAIN, APP_NAME, "query") + self.propagation_destination = RNS.Destination(self.identity, RNS.Destination.IN, RNS.Destination.SINGLE, APP_NAME, "propagation") - RNS.Transport.register_announce_handler(LXMFDeliveryAnnounceHandler(self)) + RNS.Transport.register_announce_handler(LXMFDeliveryAnnounceHandler(self)) - self.__delivery_callback = None + self.__delivery_callback = None - job_thread = threading.Thread(target=self.jobloop) - job_thread.setDaemon(True) - job_thread.start() + job_thread = threading.Thread(target=self.jobloop) + job_thread.setDaemon(True) + job_thread.start() - def register_delivery_identity(self, identity, display_name = None): - delivery_destination = RNS.Destination(identity, RNS.Destination.IN, RNS.Destination.SINGLE, "lxmf", "delivery") - delivery_destination.set_packet_callback(self.delivery_packet) - delivery_destination.set_link_established_callback(self.delivery_link_established) - delivery_destination.display_name = display_name + def register_delivery_identity(self, identity, display_name = None): + delivery_destination = RNS.Destination(identity, RNS.Destination.IN, RNS.Destination.SINGLE, "lxmf", "delivery") + delivery_destination.set_packet_callback(self.delivery_packet) + delivery_destination.set_link_established_callback(self.delivery_link_established) + delivery_destination.display_name = display_name - if display_name != None: - delivery_destination.set_default_app_data(display_name.encode("utf-8")) + if display_name != None: + delivery_destination.set_default_app_data(display_name.encode("utf-8")) - self.delivery_destinations[delivery_destination.hash] = delivery_destination - return delivery_destination + self.delivery_destinations[delivery_destination.hash] = delivery_destination + return delivery_destination - def register_delivery_callback(self, callback): - self.__delivery_callback = callback + def register_delivery_callback(self, callback): + self.__delivery_callback = callback - def announce(self, destination_hash): - if destination_hash in self.delivery_destinations: - delivery_destination = self.delivery_destinations[destination_hash] - delivery_destination.announce(delivery_destination.display_name.encode("utf-8")) + def announce(self, destination_hash): + if destination_hash in self.delivery_destinations: + delivery_destination = self.delivery_destinations[destination_hash] + delivery_destination.announce(delivery_destination.display_name.encode("utf-8")) - def handle_outbound(self, lxmessage): - lxmessage.state = LXMessage.OUTBOUND - if not lxmessage.packed: - lxmessage.pack() + def handle_outbound(self, lxmessage): + lxmessage.state = LXMessage.OUTBOUND + if not lxmessage.packed: + lxmessage.pack() - lxmessage.determine_transport_encryption() + lxmessage.determine_transport_encryption() - while self.processing_outbound: - time.sleep(0.1) + while self.processing_outbound: + time.sleep(0.1) - self.pending_outbound.append(lxmessage) - self.process_outbound() + self.pending_outbound.append(lxmessage) + self.process_outbound() - def lxmf_delivery(self, lxmf_data, destination_type = None): - try: - message = LXMessage.unpack_from_bytes(lxmf_data) + def lxmf_delivery(self, lxmf_data, destination_type = None): + try: + message = LXMessage.unpack_from_bytes(lxmf_data) - if RNS.Reticulum.should_allow_unencrypted(): - message.transport_encrypted = False - message.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_UNENCRYPTED - else: - if destination_type == RNS.Destination.SINGLE: - message.transport_encrypted = True - message.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_RSA - elif destination_type == RNS.Destination.GROUP: - message.transport_encrypted = True - message.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_AES - elif destination_type == RNS.Destination.LINK: - message.transport_encrypted = True - message.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_EC - else: - message.transport_encrypted = False - message.transport_encryption = None + if RNS.Reticulum.should_allow_unencrypted(): + message.transport_encrypted = False + message.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_UNENCRYPTED + else: + if destination_type == RNS.Destination.SINGLE: + message.transport_encrypted = True + message.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_RSA + elif destination_type == RNS.Destination.GROUP: + message.transport_encrypted = True + message.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_AES + elif destination_type == RNS.Destination.LINK: + message.transport_encrypted = True + message.transport_encryption = LXMessage.ENCRYPTION_DESCRIPTION_EC + else: + message.transport_encrypted = False + message.transport_encryption = None - if self.__delivery_callback != None: - self.__delivery_callback(message) + if self.__delivery_callback != None: + self.__delivery_callback(message) - return True + return True - except Exception as e: - RNS.log("Could not assemble LXMF message from received data", RNS.LOG_NOTICE) - RNS.log("The contained exception was: "+str(e), RNS.LOG_DEBUG) - return False + except Exception as e: + RNS.log("Could not assemble LXMF message from received data", RNS.LOG_NOTICE) + RNS.log("The contained exception was: "+str(e), RNS.LOG_DEBUG) + return False - def delivery_packet(self, data, packet): - try: - if packet.destination_type != RNS.Destination.LINK: - lxmf_data = b"" - lxmf_data += packet.destination.hash - lxmf_data += data - else: - lxmf_data = data + def delivery_packet(self, data, packet): + try: + if packet.destination_type != RNS.Destination.LINK: + lxmf_data = b"" + lxmf_data += packet.destination.hash + lxmf_data += data + else: + lxmf_data = data - if self.lxmf_delivery(lxmf_data, packet.destination_type): - packet.prove() + if self.lxmf_delivery(lxmf_data, packet.destination_type): + packet.prove() - except Exception as e: - RNS.log("Exception occurred while parsing incoming LXMF data.", RNS.LOG_ERROR) - RNS.log("The contained exception was: "+str(e), RNS.LOG_ERROR) + except Exception as e: + RNS.log("Exception occurred while parsing incoming LXMF data.", RNS.LOG_ERROR) + RNS.log("The contained exception was: "+str(e), RNS.LOG_ERROR) - def delivery_link_established(self, link): - link.set_packet_callback(self.delivery_packet) - link.set_resource_strategy(RNS.Link.ACCEPT_ALL) - link.set_resource_started_callback(self.resource_transfer_began) - link.set_resource_concluded_callback(self.resource_transfer_concluded) + def delivery_link_established(self, link): + link.set_packet_callback(self.delivery_packet) + link.set_resource_strategy(RNS.Link.ACCEPT_ALL) + link.set_resource_started_callback(self.resource_transfer_began) + link.set_resource_concluded_callback(self.resource_transfer_concluded) - def delivery_link_closed(self, link): - pass + def delivery_link_closed(self, link): + pass - def resource_transfer_began(self, resource): - RNS.log("Transfer began for resource "+str(resource), RNS.LOG_DEBUG) + def resource_transfer_began(self, resource): + RNS.log("Transfer began for resource "+str(resource), RNS.LOG_DEBUG) - def resource_transfer_concluded(self, resource): - RNS.log("Transfer concluded for resource "+str(resource), RNS.LOG_DEBUG) - if resource.status == RNS.Resource.COMPLETE: - self.lxmf_delivery(resource.data.read(), resource.link.type) + def resource_transfer_concluded(self, resource): + RNS.log("Transfer concluded for resource "+str(resource), RNS.LOG_DEBUG) + if resource.status == RNS.Resource.COMPLETE: + self.lxmf_delivery(resource.data.read(), resource.link.type) - def jobloop(self): - while (True): - # TODO: Improve this to scheduling, so manual - # triggers can delay next run - self.jobs() - time.sleep(LXMRouter.PROCESSING_INTERVAL) + def jobloop(self): + while (True): + # TODO: Improve this to scheduling, so manual + # triggers can delay next run + self.jobs() + time.sleep(LXMRouter.PROCESSING_INTERVAL) - def jobs(self): - self.process_outbound() - self.clean_links() + def jobs(self): + self.process_outbound() + self.clean_links() - def clean_links(self): - closed_links = [] - for link_hash in self.direct_links: - link = self.direct_links[link_hash] - inactive_time = link.inactive_for() + def clean_links(self): + closed_links = [] + for link_hash in self.direct_links: + link = self.direct_links[link_hash] + inactive_time = link.inactive_for() - if inactive_time > LXMRouter.LINK_MAX_INACTIVITY: - link.teardown() - closed_links.append(link_hash) + if inactive_time > LXMRouter.LINK_MAX_INACTIVITY: + link.teardown() + closed_links.append(link_hash) - for link_hash in closed_links: - cleaned_link = self.direct_links.pop(link_hash) - RNS.log("Cleaned link "+str(cleaned_link), RNS.LOG_DEBUG) + for link_hash in closed_links: + cleaned_link = self.direct_links.pop(link_hash) + RNS.log("Cleaned link "+str(cleaned_link), RNS.LOG_DEBUG) - def fail_message(self, lxmessage): - RNS.log(str(lxmessage)+" failed to send", RNS.LOG_DEBUG) + def fail_message(self, lxmessage): + RNS.log(str(lxmessage)+" failed to send", RNS.LOG_DEBUG) - self.pending_outbound.remove(lxmessage) - self.failed_outbound.append(lxmessage) + self.pending_outbound.remove(lxmessage) + self.failed_outbound.append(lxmessage) - lxmessage.state = LXMessage.FAILED - if lxmessage.failed_callback != None: - lxmessage.failed_callback(lxmessage) + lxmessage.state = LXMessage.FAILED + if lxmessage.failed_callback != None: + lxmessage.failed_callback(lxmessage) - def process_outbound(self, sender = None): - if self.processing_outbound: - return + def process_outbound(self, sender = None): + if self.processing_outbound: + return - for lxmessage in self.pending_outbound: - if lxmessage.state == LXMessage.DELIVERED: - RNS.log("Delivery has occurred for "+str(lxmessage)+", removing from outbound queue", RNS.LOG_DEBUG) - self.pending_outbound.remove(lxmessage) - else: - RNS.log("Starting outbound processing for "+str(lxmessage)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash), RNS.LOG_DEBUG) - # Outbound handling for opportunistic messages - if lxmessage.method == LXMessage.OPPORTUNISTIC: - if lxmessage.delivery_attempts <= LXMRouter.MAX_DELIVERY_ATTEMPTS: - if not hasattr(lxmessage, "next_delivery_attempt") or time.time() > lxmessage.next_delivery_attempt: - lxmessage.delivery_attempts += 1 - lxmessage.next_delivery_attempt = time.time() + LXMRouter.DELIVERY_RETRY_WAIT - RNS.log("Opportunistic delivery attempt "+str(lxmessage.delivery_attempts)+" for "+str(lxmessage)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash), RNS.LOG_DEBUG) - lxmessage.send() - else: - RNS.log("Max delivery attempts reached for oppertunistic "+str(lxmessage)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash), RNS.LOG_DEBUG) - self.fail_message(lxmessage) + for lxmessage in self.pending_outbound: + if lxmessage.state == LXMessage.DELIVERED: + RNS.log("Delivery has occurred for "+str(lxmessage)+", removing from outbound queue", RNS.LOG_DEBUG) + self.pending_outbound.remove(lxmessage) + else: + RNS.log("Starting outbound processing for "+str(lxmessage)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash), RNS.LOG_DEBUG) + # Outbound handling for opportunistic messages + if lxmessage.method == LXMessage.OPPORTUNISTIC: + if lxmessage.delivery_attempts <= LXMRouter.MAX_DELIVERY_ATTEMPTS: + if not hasattr(lxmessage, "next_delivery_attempt") or time.time() > lxmessage.next_delivery_attempt: + lxmessage.delivery_attempts += 1 + lxmessage.next_delivery_attempt = time.time() + LXMRouter.DELIVERY_RETRY_WAIT + RNS.log("Opportunistic delivery attempt "+str(lxmessage.delivery_attempts)+" for "+str(lxmessage)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash), RNS.LOG_DEBUG) + lxmessage.send() + else: + RNS.log("Max delivery attempts reached for oppertunistic "+str(lxmessage)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash), RNS.LOG_DEBUG) + self.fail_message(lxmessage) - # Outbound handling for messages transferred - # over a direct link to the final recipient - elif lxmessage.method == LXMessage.DIRECT: - if lxmessage.delivery_attempts <= LXMRouter.MAX_DELIVERY_ATTEMPTS: - delivery_destination_hash = lxmessage.get_destination().hash + # Outbound handling for messages transferred + # over a direct link to the final recipient + elif lxmessage.method == LXMessage.DIRECT: + if lxmessage.delivery_attempts <= LXMRouter.MAX_DELIVERY_ATTEMPTS: + delivery_destination_hash = lxmessage.get_destination().hash - if delivery_destination_hash in self.direct_links: - # A link already exists, so we'll try to use it - # to deliver the message - direct_link = self.direct_links[delivery_destination_hash] - if direct_link.status == RNS.Link.ACTIVE: - if lxmessage.state != LXMessage.SENDING: - RNS.log("Starting transfer of "+str(lxmessage)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash), RNS.LOG_DEBUG) - lxmessage.set_delivery_destination(direct_link) - lxmessage.send() - else: - if lxmessage.representation == LXMessage.RESOURCE: - RNS.log("The transfer of "+str(lxmessage)+" is in progress ("+str(round(lxmessage.progress*100, 1))+"%)", RNS.LOG_DEBUG) - else: - RNS.log("Waiting for proof for "+str(lxmessage)+" sent as link packet", RNS.LOG_DEBUG) - elif direct_link.status == RNS.Link.CLOSED: - RNS.log("The link to "+RNS.prettyhexrep(lxmessage.get_destination().hash)+" was closed", RNS.LOG_DEBUG) - lxmessage.set_delivery_destination(None) - self.direct_links.pop(delivery_destination_hash) - lxmessage.next_delivery_attempt = time.time() + LXMRouter.DELIVERY_RETRY_WAIT - else: - # Simply wait for the link to become - # active or close - RNS.log("The link to "+RNS.prettyhexrep(lxmessage.get_destination().hash)+" is pending, waiting for link to become active", RNS.LOG_DEBUG) - else: - # No link exists, so we'll try to establish one, but - # only if we've never tried before, or the retry wait - # period has elapsed. - if not hasattr(lxmessage, "next_delivery_attempt") or time.time() > lxmessage.next_delivery_attempt: - lxmessage.delivery_attempts += 1 - lxmessage.next_delivery_attempt = time.time() + LXMRouter.DELIVERY_RETRY_WAIT + if delivery_destination_hash in self.direct_links: + # A link already exists, so we'll try to use it + # to deliver the message + direct_link = self.direct_links[delivery_destination_hash] + if direct_link.status == RNS.Link.ACTIVE: + if lxmessage.state != LXMessage.SENDING: + RNS.log("Starting transfer of "+str(lxmessage)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash), RNS.LOG_DEBUG) + lxmessage.set_delivery_destination(direct_link) + lxmessage.send() + else: + if lxmessage.representation == LXMessage.RESOURCE: + RNS.log("The transfer of "+str(lxmessage)+" is in progress ("+str(round(lxmessage.progress*100, 1))+"%)", RNS.LOG_DEBUG) + else: + RNS.log("Waiting for proof for "+str(lxmessage)+" sent as link packet", RNS.LOG_DEBUG) + elif direct_link.status == RNS.Link.CLOSED: + RNS.log("The link to "+RNS.prettyhexrep(lxmessage.get_destination().hash)+" was closed", RNS.LOG_DEBUG) + lxmessage.set_delivery_destination(None) + self.direct_links.pop(delivery_destination_hash) + lxmessage.next_delivery_attempt = time.time() + LXMRouter.DELIVERY_RETRY_WAIT + else: + # Simply wait for the link to become + # active or close + RNS.log("The link to "+RNS.prettyhexrep(lxmessage.get_destination().hash)+" is pending, waiting for link to become active", RNS.LOG_DEBUG) + else: + # No link exists, so we'll try to establish one, but + # only if we've never tried before, or the retry wait + # period has elapsed. + if not hasattr(lxmessage, "next_delivery_attempt") or time.time() > lxmessage.next_delivery_attempt: + lxmessage.delivery_attempts += 1 + lxmessage.next_delivery_attempt = time.time() + LXMRouter.DELIVERY_RETRY_WAIT - if lxmessage.delivery_attempts < LXMRouter.MAX_DELIVERY_ATTEMPTS: - if RNS.Transport.has_path(lxmessage.get_destination().hash): - RNS.log("Establishing link to "+RNS.prettyhexrep(lxmessage.get_destination().hash)+" for delivery attempt "+str(lxmessage.delivery_attempts)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash), RNS.LOG_DEBUG) - delivery_link = RNS.Link(lxmessage.get_destination()) - delivery_link.set_link_established_callback(self.process_outbound) - self.direct_links[delivery_destination_hash] = delivery_link - else: - RNS.log("No path known for delivery attempt "+str(lxmessage.delivery_attempts)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash)+". Requesting path...", RNS.LOG_DEBUG) - RNS.Transport.request_path(lxmessage.get_destination().hash) - lxmessage.next_delivery_attempt = time.time() + LXMRouter.PATH_REQUEST_WAIT - else: - RNS.log("Max delivery attempts reached for direct "+str(lxmessage)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash), RNS.LOG_DEBUG) - self.fail_message(lxmessage) + if lxmessage.delivery_attempts < LXMRouter.MAX_DELIVERY_ATTEMPTS: + if RNS.Transport.has_path(lxmessage.get_destination().hash): + RNS.log("Establishing link to "+RNS.prettyhexrep(lxmessage.get_destination().hash)+" for delivery attempt "+str(lxmessage.delivery_attempts)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash), RNS.LOG_DEBUG) + delivery_link = RNS.Link(lxmessage.get_destination()) + delivery_link.set_link_established_callback(self.process_outbound) + self.direct_links[delivery_destination_hash] = delivery_link + else: + RNS.log("No path known for delivery attempt "+str(lxmessage.delivery_attempts)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash)+". Requesting path...", RNS.LOG_DEBUG) + RNS.Transport.request_path(lxmessage.get_destination().hash) + lxmessage.next_delivery_attempt = time.time() + LXMRouter.PATH_REQUEST_WAIT + else: + RNS.log("Max delivery attempts reached for direct "+str(lxmessage)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash), RNS.LOG_DEBUG) + self.fail_message(lxmessage) - # Outbound handling for messages transported via - # propagation to a LXMF router network. - elif lxmessage.method == LXMessage.PROPAGATED: - RNS.log("Attempting propagated delivery for "+str(lxmessage)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash), RNS.LOG_DEBUG) - raise NotImplementedError("LXMF propagation is not implemented yet") \ No newline at end of file + # Outbound handling for messages transported via + # propagation to a LXMF router network. + elif lxmessage.method == LXMessage.PROPAGATED: + RNS.log("Attempting propagated delivery for "+str(lxmessage)+" to "+RNS.prettyhexrep(lxmessage.get_destination().hash), RNS.LOG_DEBUG) + raise NotImplementedError("LXMF propagation is not implemented yet") \ No newline at end of file