Queue outbound LXMF when the inbound path has gone idle

crosstalk.py

@@ -34,6 +34,11 @@
 from src.backend.colour_utils import ColourUtils
 from src.backend.interface_config_parser import InterfaceConfigParser
 from src.backend.interface_editor import InterfaceEditor
+from src.backend.outbound_identity import (
+    parse_path_timeout,
+    recall_send_identity,
+    remember_destination_identity,
+)
 from src.backend.reticulum_startup import start_reticulum
 from src.backend.lxmf_message_fields import LxmfImageField, LxmfFileAttachmentsField, LxmfFileAttachment, LxmfAudioField
 from src.backend.audio_call_manager import AudioCall, AudioCallManager
@@ -2225,6 +2230,10 @@ async def index(request):
             if "delivery_method" in data:
                 delivery_method = data["delivery_method"]
 
+            path_timeout_seconds = parse_path_timeout(
+                data.get("path_timeout", request.query.get("path_timeout")),
+            )
+
             # get data from json
             destination_hash = data["lxmf_message"]["destination_hash"]
             content = data["lxmf_message"]["content"]
@@ -2267,7 +2276,8 @@ async def index(request):
                     image_field=image_field,
                     audio_field=audio_field,
                     file_attachments_field=file_attachments_field,
-                    delivery_method=delivery_method
+                    delivery_method=delivery_method,
+                    path_timeout_seconds=path_timeout_seconds,
                 )
 
                 return web.json_response({
@@ -3176,6 +3186,24 @@ def on_lxmf_delivery(self, lxmf_message: LXMF.LXMessage):
             # upsert lxmf message to database
             self.db_upsert_lxmf_message(lxmf_message)
 
+            # Keep the sender identity so a later reply can be queued even if
+            # the inbound path has expired.
+            try:
+                source_hash = lxmf_message.source_hash
+                source_identity = RNS.Identity.recall(source_hash)
+                if source_identity is not None:
+                    remember_destination_identity(source_hash, source_identity)
+                    self.db_upsert_announce(
+                        source_identity,
+                        source_hash,
+                        "lxmf.delivery",
+                        None,
+                        lxmf_message.hash,
+                    )
+            except Exception as e:
+                print("failed to persist inbound LXMF sender identity")
+                print(e)
+
             # update lxmf user icon if icon appearance field is available
             try:
                 message_fields = lxmf_message.get_fields()
@@ -3305,6 +3333,14 @@ def db_upsert_announce(self, identity: RNS.Identity, destination_hash: bytes, as
         query = query.on_conflict(conflict_target=[database.Announce.destination_hash], update=data)
         query.execute()
 
+    def _announce_public_key_b64(self, destination_hash: bytes):
+        announce = database.Announce.get_or_none(
+            database.Announce.destination_hash == destination_hash.hex()
+        )
+        if announce is None:
+            return None
+        return announce.identity_public_key
+
     # upserts a custom destination display name to the database
     def db_upsert_custom_destination_display_name(self, destination_hash: str, display_name: str):
 
@@ -3356,7 +3392,8 @@ async def send_message(self, destination_hash: str, content: str,
                            image_field: LxmfImageField = None,
                            audio_field: LxmfAudioField = None,
                            file_attachments_field: LxmfFileAttachmentsField = None,
-                           delivery_method: str = None) -> LXMF.LXMessage:
+                           delivery_method: str = None,
+                           path_timeout_seconds: float = 0) -> LXMF.LXMessage:
 
         # convert destination hash to bytes
         destination_hash = bytes.fromhex(destination_hash)
@@ -3376,26 +3413,30 @@ async def send_message(self, destination_hash: str, content: str,
                 has_path=RNS.Transport.has_path(destination_hash),
             )
         else:
-
-            # determine when to timeout finding path
-            timeout_after_seconds = time.time() + 10
-
-            # check if we have a path to the destination
+            # A reply is a new outbound delivery. Request a path if needed,
+            # but do not fail the send because the inbound link went idle.
             if not RNS.Transport.has_path(destination_hash):
-
-                # we don't have a path, so we need to request it
                 RNS.Transport.request_path(destination_hash)
 
-                # wait until we have a path, or give up after the configured timeout
-                while not RNS.Transport.has_path(destination_hash) and time.time() < timeout_after_seconds:
-                    await asyncio.sleep(0.1)
+            timeout_after_seconds = time.time() + max(float(path_timeout_seconds), 0)
+            while (
+                path_timeout_seconds > 0
+                and not RNS.Transport.has_path(destination_hash)
+                and self._announce_public_key_b64(destination_hash) is None
+                and RNS.Identity.recall(destination_hash) is None
+                and time.time() < timeout_after_seconds
+            ):
+                await asyncio.sleep(0.1)
 
-        # find destination identity from hash
-        destination_identity = RNS.Identity.recall(destination_hash)
+        destination_identity = recall_send_identity(
+            destination_hash,
+            self._announce_public_key_b64(destination_hash),
+        )
         if destination_identity is None:
-
-            # we have to bail out of sending, since we don't have the identity/path yet
-            raise Exception("Could not find path to destination. Try again later.")
+            raise Exception(
+                "Unknown destination identity. Crosstalk has not received "
+                "from this peer and has no saved announce for them."
+            )
 
         # create destination for recipients lxmf delivery address
         lxmf_destination = RNS.Destination(destination_identity, RNS.Destination.OUT, RNS.Destination.SINGLE, "lxmf", "delivery")

src/backend/outbound_identity.py

@@ -0,0 +1,65 @@
+"""Outbound LXMF send helpers.
+
+A reply is a new delivery. Crosstalk must not require the inbound Reticulum
+path or link to still exist after a long pause. Identity can be restored from
+a saved announce when RNS has culled ``known_destinations``.
+"""
+
+import base64
+
+import RNS
+
+
+# Do not block the send HTTP call waiting for a path. LXMF retries after
+# handle_outbound. Callers that want the old 10s wait can pass path_timeout.
+DEFAULT_SEND_PATH_TIMEOUT_SECONDS = 0
+
+
+def parse_path_timeout(raw_value, default=DEFAULT_SEND_PATH_TIMEOUT_SECONDS):
+    """Parse a send path-wait in seconds. Negative values are treated as 0."""
+    if raw_value is None or raw_value == "":
+        return default
+    timeout = float(raw_value)
+    if timeout < 0:
+        return 0
+    return timeout
+
+
+def identity_from_public_key_bytes(public_key):
+    if not public_key:
+        return None
+    identity = RNS.Identity(create_keys=False)
+    identity.load_public_key(public_key)
+    return identity
+
+
+def identity_from_public_key_b64(public_key_b64):
+    if not public_key_b64:
+        return None
+    return identity_from_public_key_bytes(base64.b64decode(public_key_b64))
+
+
+def remember_destination_identity(destination_hash, identity, packet_hash=None):
+    """Re-seed RNS known destinations so a later recall succeeds."""
+    if identity is None or destination_hash is None:
+        return
+    RNS.Identity.remember(
+        packet_hash or destination_hash,
+        destination_hash,
+        identity.get_public_key(),
+        getattr(identity, "app_data", None),
+    )
+
+
+def recall_send_identity(destination_hash, announce_public_key_b64=None):
+    """Return an identity for sending, from RNS recall or a saved announce key."""
+    identity = RNS.Identity.recall(destination_hash)
+    if identity is not None:
+        return identity
+
+    identity = identity_from_public_key_b64(announce_public_key_b64)
+    if identity is not None:
+        remember_destination_identity(destination_hash, identity)
+        return identity
+
+    return None

tests/test_outbound_identity.py

@@ -0,0 +1,48 @@
+import base64
+import unittest
+
+import RNS
+
+from src.backend.outbound_identity import (
+    DEFAULT_SEND_PATH_TIMEOUT_SECONDS,
+    identity_from_public_key_b64,
+    parse_path_timeout,
+    recall_send_identity,
+    remember_destination_identity,
+)
+
+
+class OutboundIdentityTest(unittest.TestCase):
+    def test_parse_path_timeout_defaults_to_zero(self):
+        self.assertEqual(parse_path_timeout(None), 0)
+        self.assertEqual(parse_path_timeout(""), DEFAULT_SEND_PATH_TIMEOUT_SECONDS)
+        self.assertEqual(parse_path_timeout("15"), 15.0)
+        self.assertEqual(parse_path_timeout(-3), 0)
+
+    def test_restores_identity_from_saved_public_key(self):
+        original = RNS.Identity()
+        public_key_b64 = base64.b64encode(original.get_public_key()).decode("utf-8")
+        restored = identity_from_public_key_b64(public_key_b64)
+        self.assertEqual(restored.get_public_key(), original.get_public_key())
+
+    def test_recall_send_identity_uses_announce_key_when_rns_unknown(self):
+        original = RNS.Identity()
+        destination_hash = bytes(range(16))
+        public_key_b64 = base64.b64encode(original.get_public_key()).decode("utf-8")
+
+        recalled = recall_send_identity(destination_hash, public_key_b64)
+        self.assertIsNotNone(recalled)
+        self.assertEqual(recalled.get_public_key(), original.get_public_key())
+        self.assertIsNotNone(RNS.Identity.recall(destination_hash, _no_use=True))
+
+    def test_remember_destination_identity_seeds_recall(self):
+        original = RNS.Identity()
+        destination_hash = bytes(range(16, 32))
+        remember_destination_identity(destination_hash, original)
+        recalled = RNS.Identity.recall(destination_hash, _no_use=True)
+        self.assertIsNotNone(recalled)
+        self.assertEqual(recalled.get_public_key(), original.get_public_key())
+
+
+if __name__ == "__main__":
+    unittest.main()