Need help: LXMF sync from propagation node for own application

[SebastianObi]

Hi Mark,

i develop some small tools/programs which use the LXMF protocoll.

Through the tutorial, the source code and the code snippets from HarlekinSimplex I basically understood the functionality. Everything works except for retrieving messages on a propagation node.

I am not sure how this works in detail, as it is not apparent in the reticulum manual.

At the moment i'm working on a LXMF distribution group (E-Mail style group but with LXMF).

Here is my whole sourcecode:
https://github.com/SebastianObi/LXMF-Tools/blob/main/lxmf_distribution_group.py

I maked a small LXMF class.

When i call my sync function:

def sync_now(self, limit=None):
        if self.message_router.get_outbound_propagation_node() is not None:
            if self.message_router.propagation_transfer_state == LXMF.LXMRouter.PR_IDLE or self.message_router.propagation_transfer_state == LXMF.LXMRouter.PR_COMPLETE:
                log("LXMF - Message sync requested from propagation node " + RNS.prettyhexrep(self.message_router.get_outbound_propagation_node()) + " for " + str(self.identity))
                self.message_router.request_messages_from_propagation_node(self.identity, max_messages = limit)
                return True
            else:
                return False
        else:
            return False

I get the error:

[2022-10-07 14:30:26] [Debug] Activating link <2191160c94bf60b07a7bc1e853f3d028>
[2022-10-07 14:30:26] [Verbose] Link <2191160c94bf60b07a7bc1e853f3d028> established with <lxmf.propagation.b54e09cae2d82e4964b3da1a28d2b46a/ca2762fe5283873719aececfb9e18835>, RTT is 0.185s
[2022-10-07 14:30:26] [Error] Error while executing response received callback from <RNS.Link.RequestReceipt object at 0x7fa3fc616af0>. The contained exception was: '<' not supported between instances of 'int' and 'str'

The whole class:

class lxmf_connection:
    message_received_callback = None
    message_notification_callback = None
    message_notification_success_callback = None
    message_notification_failed_callback = None


    def __init__(self, storage_path=None, destination_name="lxmf", destination_type="delivery", display_name="", send_delay=0, desired_method="direct", propagation_node=None, try_propagation_on_fail=False, announce_startup=False, announce_startup_delay=0, announce_periodic=False, announce_periodic_interval=360, sync_startup=False, sync_limit=8, sync_periodic=False, sync_interval=360):
        self.destination_name = destination_name
        self.destination_type = destination_type
        self.aspect_filter = self.destination_name + "." + self.destination_type

        self.display_name = display_name

        self.send_delay = int(send_delay)

        if desired_method == "propagated" or desired_method == "PROPAGATED":
            self.desired_method_direct = False
        else:
            self.desired_method_direct = True
        self.propagation_node = propagation_node
        self.try_propagation_on_fail = try_propagation_on_fail

        self.storage_path = storage_path

        self.announce_startup = announce_startup
        self.announce_startup_delay = int(announce_startup_delay)

        self.announce_periodic = announce_periodic
        self.announce_periodic_interval = int(announce_periodic_interval)

        self.sync_startup = sync_startup
        self.sync_limit = sync_limit
        self.sync_periodic = sync_periodic
        self.sync_interval = int(sync_interval)

        if not os.path.isdir(self.storage_path):
            os.makedirs(self.storage_path)
            log("LXMF - Storage path was created", LOG_NOTICE)
        log("LXMF - Storage path: " + self.storage_path, LOG_INFO)

        self.identity_path = self.storage_path + "/identity"

        if os.path.isfile(self.identity_path):
            try:
                self.identity = RNS.Identity.from_file(self.identity_path)
                if self.identity != None:
                    log("LXMF - Loaded Primary Identity %s from %s" % (str(self.identity), self.identity_path))
                else:
                    log("LXMF - Could not load the Primary Identity from "+self.identity_path, LOG_ERROR)
            except Exception as e:
                log("LXMF - Could not load the Primary Identity from "+self.identity_path, LOG_ERROR)
                log("LXMF - The contained exception was: %s" % (str(e)), LOG_ERROR)
        else:
            try:
                log("LXMF - No Primary Identity file found, creating new...")
                self.identity = RNS.Identity()
                self.identity.to_file(self.identity_path)
                log("LXMF - Created new Primary Identity %s" % (str(self.identity)))
            except Exception as e:
                log("LXMF - Could not create and save a new Primary Identity", LOG_ERROR)
                log("LXMF - The contained exception was: %s" % (str(e)), LOG_ERROR)

        self.message_router = LXMF.LXMRouter(identity=self.identity, storagepath=self.storage_path)

        #self.destination = RNS.Destination(self.identity, RNS.Destination.IN, RNS.Destination.SINGLE, self.destination_name, self.destination_type)
        self.destination = self.message_router.register_delivery_identity(self.identity, display_name=self.display_name)

        if self.display_name == "":
            self.display_name = RNS.prettyhexrep(self.destination_hash())

        self.destination.set_default_app_data(self.display_name.encode("utf-8"))

        self.destination.set_proof_strategy(RNS.Destination.PROVE_ALL)

        RNS.Identity.remember(packet_hash=None, destination_hash=self.destination.hash, public_key=self.identity.get_public_key(), app_data=None)

        log("LXMF - Identity: " + str(self.identity), LOG_INFO)
        log("LXMF - Destination: " + str(self.destination), LOG_INFO)
        log("LXMF - Hash: " + RNS.prettyhexrep(self.destination_hash()), LOG_INFO)
        log("LXMF - My own address: " + RNS.prettyhexrep(self.destination_hash()) + " (to communicate with)", LOG_NOTICE)

        self.destination.set_link_established_callback(lxmf_connection.client_connected)

        self.autoselect_propagation_node()

        if self.announce_startup:
            time.sleep(self.announce_startup_delay)
            self.announce_now()

        if self.announce_periodic:
            self.announce(True)
 
        if self.sync_startup:
            self.sync_now(limit=self.sync_limit)

        if self.sync_periodic:
            self.sync(True)


    def register_announce_callback(self, handler_function):
        self.announce_callback = handler_function(self.aspect_filter)
        RNS.Transport.register_announce_handler(self.announce_callback)


    def register_message_received_callback(self, handler_function):
        self.message_received_callback = handler_function


    def register_message_notification_callback(self, handler_function):
        self.message_notification_callback = handler_function


    def register_message_notification_success_callback(self, handler_function):
        self.message_notification_success_callback = handler_function


    def register_message_notification_failed_callback(self, handler_function):
        self.message_notification_failed_callback = handler_function


    def destination_hash(self):
        return self.destination.hash


    def destination_check(self, destination):
        if type(destination) is not bytes:
            if len(destination) == ((RNS.Reticulum.TRUNCATED_HASHLENGTH//8)*2)+2:
                destination = destination[1:-1]

            if len(destination) != ((RNS.Reticulum.TRUNCATED_HASHLENGTH//8)*2):
                log("LXMF - Destination length is invalid", LOG_ERROR)
                return False

            try:    
                destination = bytes.fromhex(destination)
            except Exception as e:
                log("LXMF - Destination is invalid", LOG_ERROR)
                return False

        return True


    def destination_correct(self, destination):
        if type(destination) is not bytes:
            if len(destination) == ((RNS.Reticulum.TRUNCATED_HASHLENGTH//8)*2)+2:
                destination = destination[1:-1]

            if len(destination) != ((RNS.Reticulum.TRUNCATED_HASHLENGTH//8)*2):
                return ""

            try:    
                destination_bytes = bytes.fromhex(destination)
                return destination
            except Exception as e:
                return ""

        return ""


    def send(self, destination, content="", title=None):
        if type(destination) is not bytes:
            if len(destination) == ((RNS.Reticulum.TRUNCATED_HASHLENGTH//8)*2)+2:
                destination = destination[1:-1]

            if len(destination) != ((RNS.Reticulum.TRUNCATED_HASHLENGTH//8)*2):
                log("LXMF - Destination length is invalid", LOG_ERROR)
                return

            try:    
                destination = bytes.fromhex(destination)
            except Exception as e:
                log("LXMF - Destination is invalid", LOG_ERROR)
                return

        destination_identity = RNS.Identity.recall(destination)
        destination = RNS.Destination(destination_identity, RNS.Destination.OUT, RNS.Destination.SINGLE, self.destination_name, self.destination_type)
        self.send_message(destination, self.destination, content, title)


    def send_message(self, destination, source, content="", title=None):
        if self.desired_method_direct:
            desired_method = LXMF.LXMessage.DIRECT
        else:
            desired_method = LXMF.LXMessage.PROPAGATED

        message = LXMF.LXMessage(destination, source, content, title=title, desired_method=desired_method)

        self.log_message(message, "LXMF - Message send")

        message.register_delivery_callback(self.message_notification)
        message.register_failed_callback(self.message_notification)

        if self.message_router.get_outbound_propagation_node() != None:
            message.try_propagation_on_fail = self.try_propagation_on_fail

        try:
            self.message_router.handle_outbound(message)
            time.sleep(self.send_delay)
        except Exception as e:
            log("LXMF - Could not send message " + str(message), LOG_ERROR)
            log("LXMF - The contained exception was: " + str(e), LOG_ERROR)
            return


    def message_notification(self, message):
        if self.message_notification_callback is not None:
            self.message_notification_callback(message)

        if message.state == LXMF.LXMessage.FAILED and hasattr(message, "try_propagation_on_fail") and message.try_propagation_on_fail:
            self.log_message(message, "LXMF - Delivery receipt (failed) Retrying as propagated message")
            message.try_propagation_on_fail = None
            message.delivery_attempts = 0
            del message.next_delivery_attempt
            message.packed = None
            message.desired_method = LXMF.LXMessage.PROPAGATED
            self.message_router.handle_outbound(message)
        elif message.state == LXMF.LXMessage.FAILED:
            self.log_message(message, "LXMF - Delivery receipt (failed)")
            if self.message_notification_failed_callback is not None:
                self.message_notification_failed_callback(message)
        else:
            self.log_message(message, "LXMF - Delivery receipt (success)")
            if self.message_notification_success_callback is not None:
                self.message_notification_success_callback(message)


    def announce(self, initial=False):
        announce_timer = threading.Timer(self.announce_periodic_interval*60, self.announce)
        announce_timer.daemon = True
        announce_timer.start()
        
        if initial:
            return

        self.announce_now()


    def announce_now(self):
        self.destination.announce()
        log("LXMF - Announced: " + RNS.prettyhexrep(self.destination_hash()) + ": " + self.display_name, LOG_DEBUG)


    def sync(self, initial=False):
        sync_timer = threading.Timer(self.sync_interval*60, self.sync)
        sync_timer.daemon = True
        sync_timer.start()

        if initial:
            return

        self.sync_now(self.sync_limit)


    def sync_now(self, limit=None):
        if self.message_router.get_outbound_propagation_node() is not None:
            if self.message_router.propagation_transfer_state == LXMF.LXMRouter.PR_IDLE or self.message_router.propagation_transfer_state == LXMF.LXMRouter.PR_COMPLETE:
                log("LXMF - Message sync requested from propagation node " + RNS.prettyhexrep(self.message_router.get_outbound_propagation_node()) + " for " + str(self.identity))
                self.message_router.request_messages_from_propagation_node(self.identity, max_messages = limit)
                return True
            else:
                return False
        else:
            return False


    def autoselect_propagation_node(self):
        if self.propagation_node is not None:
            if len(self.propagation_node) != ((RNS.Reticulum.TRUNCATED_HASHLENGTH//8)*2):
                    log("LXMF - Propagation node length is invalid", LOG_ERROR)
            else:
                try:    
                    propagation_hash = bytes.fromhex(self.propagation_node)
                except Exception as e:
                    log("LXMF - Propagation node is invalid", LOG_ERROR)
                    return

            node_identity = RNS.Identity.recall(propagation_hash)
            if node_identity != None:
                log("LXMF - Propagation node: " + RNS.prettyhexrep(propagation_hash), LOG_INFO)
                propagation_hash = RNS.Destination.hash_from_name_and_identity("lxmf.propagation", node_identity)
                self.message_router.set_outbound_propagation_node(propagation_hash)


    @staticmethod
    def client_connected(link):
        log("LXMF - Client connected " + str(link), LOG_EXTREME)
        link.set_resource_strategy(RNS.Link.ACCEPT_ALL)
        link.set_resource_concluded_callback(lxmf_connection.resource_concluded)
        link.set_packet_callback(lxmf_connection.packet_received)


    @staticmethod
    def client_disconnect(link):
        log("LXMF - Client disconnected " + str(link), LOG_EXTREME)
        if link.teardown_reason == RNS.Link.TIMEOUT:
            log("LXMF - The link timed out, exiting now", LOG_EXTREME)
        elif link.teardown_reason == RNS.Link.DESTINATION_CLOSED:
            log("LXMF - The link was closed by the server, exiting now", LOG_EXTREME)
        else:
            log("LXMF - Link closed", LOG_EXTREME)


    @staticmethod
    def packet_received(lxmf_bytes, packet):
        log("LXMF - Single packet delivered " + str(packet), LOG_EXTREME)
        lxmf_connection.process_lxmf_message_bytes(lxmf_bytes)


    @staticmethod
    def resource_concluded(resource):
        log("LXMF - Resource data transfer (multi packet) delivered " + str(resource.file), LOG_EXTREME)
        if resource.status == RNS.Resource.COMPLETE:
            lxmf_bytes = resource.data.read()
            lxmf_connection.process_lxmf_message_bytes(lxmf_bytes)
        else:
            log("LXMF - Received resource message is not complete", LOG_EXTREME)


    @staticmethod
    def process_lxmf_message_bytes(lxmf_bytes):
        try:
            message = LXMF.LXMessage.unpack_from_bytes(lxmf_bytes)
        except Exception as e:
            log("LXMF - Could not assemble LXMF message from received data", LOG_ERROR)
            log("LXMF - The contained exception was: " + str(e), LOG_ERROR)
            return

        lxmf_connection.log_message(message, "LXMF - Message received")

        if LXMF_CONNECTION.message_received_callback is not None:
            log("LXMF - Call to registered message received callback", LOG_DEBUG)
            LXMF_CONNECTION.message_received_callback(message)
        else:
            log("LXMF - No message received callback registered", LOG_DEBUG)


    @staticmethod
    def log_message(message, message_tag="LXMF - Message log"):
        if message.signature_validated:
            signature_string = "Validated"
        else:
            if message.unverified_reason == LXMF.LXMessage.SIGNATURE_INVALID:
                signature_string = "Invalid signature"
            elif message.unverified_reason == LXMF.LXMessage.SOURCE_UNKNOWN:
                signature_string = "Cannot verify, source is unknown"
            else:
                signature_string = "Signature is invalid, reason undetermined"
        title = message.title.decode('utf-8')
        content = message.content.decode('utf-8')
        fields = message.fields
        log(message_tag + ":", LOG_DEBUG)
        log("-   Date/Time: " + time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(message.timestamp)), LOG_DEBUG)
        log("-       Title: " + title, LOG_DEBUG)
        log("-     Content: " + content, LOG_DEBUG)
        log("-      Fields: " + str(fields), LOG_DEBUG)
        log("-        Size: " + str(len(title) + len(content) + len(title) + len(pickle.dumps(fields))) + " bytes", LOG_DEBUG)
        log("-      Source: " + RNS.prettyhexrep(message.source_hash), LOG_DEBUG)
        log("- Destination: " + RNS.prettyhexrep(message.destination_hash), LOG_DEBUG)
        log("-   Signature: " + signature_string, LOG_DEBUG)

I don't know why i get this error and how to implement the propagation sync.
When I look at the source code of sideband it is completely different.

Maybe you can help me?

Many thanks.

Sebastian

[markqvist]

I took the liberty to format your post so the code was more readable :)

I only looked at this for like 20 seconds, so please correct me if I am wrong, but I think that maybe your problem is here:

https://github.com/SebastianObi/LXMF-Tools/blob/91f1a3be28d0d0dd377cdbc6dfc5a4bb0b5ae341/lxmf_distribution_group.py#L1531

sync_limit=CONFIG["lxmf"]["sync_limit"],

That way, you are reading in that config value as a string, not an integer, hence leading to the error '<' not supported between instances of 'int' and 'str', when you are telling LXMF that it should sync max "8" messages, instead of 8 ;)

So parse it to an int, and I think it will work :)

[markqvist]

Also, I will hopefully have the full LXMF documentation made soon, but everything takes time, and I have to prioritise it all. I am amazed at how much you have made here without any sort of docs on the LXMF part.

[SebastianObi]

Thanks, this was my stupid failure :)

Now i get no error messages.

But i have to figure out how the messages are received. How the callback work.

I have registered the following callback "register_delivery_callback" but it don't get used?

I start the sync with the function call "request_messages_from_propagation_node(self.identity, max_messages = limit)"
Inside this function of the LXMRouter.py are some debug messages but i don't get any message.

Maybe i do something wrong or something is missing.

This have no priority for helping me. At the moment this function is not important.
I know you have many other work :)

[SebastianObi]

I have analyzed the lxmf code and found my error. I forgotten: message_router.register_delivery_callback()

[SebastianObi]

message_router.request_messages_from_propagation_node(identity, max_messages = limit)

message_router.register_delivery_callback()