fixes: time out network commands and report failures instead of hanging, support OpenBSD better

README.md

@@ -6,6 +6,9 @@ To install use:
 
     uv tool install cmsend
 
+On platforms without a `deltachat-rpc-server` wheel, such as the BSDs,
+please build the server yourself from https://github.com/chatmail/core and put it on PATH.
+
 To send and receive from a single chatmail relay:
 
     cmsend --init nine.testrun.org   # <-- substitute with the domain you want to set as origin
@@ -15,7 +18,7 @@ so recipients see which host a message came from.
 
 To setup a tagged chat using an invite link:
 
-    cmsend -t LOG --join "INVITELINK"       # <-- quotes are neccessary because links contain "&"
+    cmsend -t LOG --join "INVITELINK"       # <-- quotes are necessary because links contain "&"
 
 To send a message to a tagged chat:
 
@@ -29,6 +32,9 @@ To send a message to a tagged chat with an attachment:
 
     cmsend -t LOG -m "here is the file" -a README.md
 
+Commands that use the network give up after 60 seconds
+and exit with an error, `--timeout` changes that limit.
+
 To show help:
 
     cmsend -h

cmsend.py

@@ -3,11 +3,14 @@
 """
 
 import argparse
+import shutil
 import socket
 import sys
+import sysconfig
 import time
+from queue import Empty
 
-from deltachat_rpc_client import DeltaChat, EventType, Rpc
+from deltachat_rpc_client import AttrDict, DeltaChat, EventType, JsonRpcError, Rpc
 from xdg_base_dirs import xdg_config_home
 
 
@@ -50,46 +53,60 @@ def main(argv=None):
     parser.add_argument(
         "-a", dest="filename", type=str, default=None, help="add file attachment"
     )
+    parser.add_argument(
+        "--timeout",
+        type=float,
+        default=60,
+        help="give up on network operations after this many seconds (default: 60)",
+    )
     args = parser.parse_args(argv)
 
     try:
         return perform_main(args)
     except KeyboardInterrupt:
         raise SystemExit(2)
+    except TimeoutError:
+        raise SystemExit(f"cmsend timed out after {args.timeout:g}s")
 
 
 def perform_main(args):
     accounts_dir = xdg_config_home().joinpath("cmsend")
     if args.verbose >= 1:
         print(f"# using accounts_dir at: {accounts_dir}")
-    with Rpc(accounts_dir=accounts_dir) as rpc:
-        dc = DeltaChat(rpc)
-        profile = Profile(dc, verbosity=args.verbose)
+    # "uv tool install" puts only cmsend on PATH, not the server installed with it
+    server = shutil.which("deltachat-rpc-server", path=sysconfig.get_path("scripts"))
+    server = server or shutil.which("deltachat-rpc-server")
+    if not server:
+        raise SystemExit("deltachat-rpc-server not found, install it or put it on PATH")
+    if args.msg is None and not (args.relay or args.invitelink or args.listtags):
+        args.msg = sys.stdin.read()
+
+    with Rpc(accounts_dir=accounts_dir, rpc_server_path=server) as rpc:
+        profile = Profile(DeltaChat(rpc), verbosity=args.verbose, timeout=args.timeout)
 
         if args.relay:
             profile.perform_init(domain=args.relay)
             return
         if not profile._account:
-            print("profile is not configured, run --init")
+            print("profile is not configured, run --init", file=sys.stderr)
             raise SystemExit(2)
 
         if args.invitelink:
             profile.perform_join(tag=args.tag, invitelink=args.invitelink)
         elif args.listtags:
             profile.perform_listtags()
         else:
-            if args.msg is None:
-                args.msg = sys.stdin.read()
-            profile.perform_send(text=args.msg, filename=args.filename, tag=args.tag)
+            profile.perform_send(args.tag, args.msg, args.filename)
 
 
 class Profile:
     _account = None
     UI_CONFIG_TAGGED_CHATS = "ui.cmsend.tagged_chats"
 
-    def __init__(self, dc, verbosity=0):
+    def __init__(self, dc, verbosity=0, timeout=60):
         self.dc = dc
         self.verbosity = verbosity
+        self.deadline = time.monotonic() + timeout
         for account in self.dc.get_all_accounts():
             if account.is_configured():
                 self._account = account
@@ -121,16 +138,30 @@ def perform_init(self, domain):
             raise SystemExit(3)
         print(f"# creating profile on {domain}")
         self._account = account = self.dc.add_account()
-        account.set_config_from_qr(f"dcaccount:{domain}")
-        self.start_io()
-        account.wait_for_event(EventType.IMAP_INBOX_IDLE)
+        try:
+            account.set_config_from_qr(f"dcaccount:{domain}")
+            self.start_io()
+            self.wait_for_event(lambda event: event.kind == EventType.IMAP_INBOX_IDLE)
+        except JsonRpcError as e:
+            account.remove()
+            raise SystemExit(f"could not create profile: {e.args[0]['message']}")
+        except TimeoutError:
+            account.remove()
+            raise
         self.verbose1(f"profile {self!r} is configured and active now")
 
     def perform_join(self, tag, invitelink):
+        kind = self._account.check_qr(invitelink)["kind"]
+        if kind not in ("askVerifyContact", "askVerifyGroup"):
+            raise SystemExit(f"not a contact or group invite link: {kind}")
+
         self.start_io()
         chat = self._account.secure_join(invitelink)
 
         def check_joined(event):
+            if event.kind == EventType.MSG_FAILED:
+                msg = self._account.get_message_by_id(event.msg_id)
+                raise SystemExit(f"joining failed: {msg.get_snapshot().error}")
             if (
                 event.kind == EventType.SECUREJOIN_JOINER_PROGRESS
                 and event["progress"] == 1000
@@ -152,7 +183,7 @@ def check_can_send(event):
         print(f"joining completed with chat_id == {chat.id} tag={tag}")
         self._account.set_config(f"{self.UI_CONFIG_TAGGED_CHATS}.{tag}", str(chat.id))
         list_tags = self._account.get_config(self.UI_CONFIG_TAGGED_CHATS) or ""
-        tags = set(list_tags.split(","))
+        tags = set(filter(None, list_tags.split(",")))
         tags.add(tag)
         self._account.set_config(self.UI_CONFIG_TAGGED_CHATS, ",".join(tags))
 
@@ -182,33 +213,41 @@ def check_sent(event):
             if self.wait_for_event(check_sent).kind == EventType.MSG_FAILED:
                 raise SystemExit(f"message {msg.id} failed: {msg.get_snapshot().error}")
             return 0
+        print(f"chat_id={chat.id} tag={tag} is not sendable", file=sys.stderr)
         raise SystemExit(5)
 
     def get_tagged_chat(self, tag):
         chat_id = self._account.get_config(f"{self.UI_CONFIG_TAGGED_CHATS}.{tag}")
         if not chat_id:
             print(
                 f"No chat tagged with tag={tag} found for sending on {self!r}, "
-                f"use -t {tag} --join 'https://i.delta.chat/...'"
+                f"use -t {tag} --join 'https://i.delta.chat/...'",
+                file=sys.stderr,
             )
             raise SystemExit(5)
 
         return self._account.get_chat_by_id(int(chat_id))
 
-    def wait_for_event(self, check_event=lambda ev: None):
+    def wait_for_event(self, check_event):
         account = self._account
+        events = self.dc.rpc.get_queue(account.id)
         start_clock = time.time()
 
         log = self.verbose2
 
-        while event := account.wait_for_event():
+        while True:
+            remaining = max(0, self.deadline - time.monotonic())
+            try:
+                event = AttrDict(events.get(timeout=remaining))
+            except Empty:
+                raise TimeoutError
             if event.kind == EventType.INCOMING_MSG:
                 msg = account.get_message_by_id(event.msg_id)
                 text = msg.get_snapshot().text
                 log(f"!received historic message: {text}")
-            if event.kind == EventType.ERROR:
+            elif event.kind == EventType.ERROR:
                 log(f"ERROR: {event.msg}")
-            if event.kind == EventType.MSG_FAILED:
+            elif event.kind == EventType.MSG_FAILED:
                 msg = account.get_message_by_id(event.msg_id)
                 text = msg.get_snapshot().text
                 log(f"Message failed: {text}")

pyproject.toml

@@ -8,10 +8,11 @@ readme = "README.md"
 description = "send end-to-end encrypted messages to chats on chatmail relays"
 requires-python = ">=3.11"
 dependencies = [
-  # 2.62.0 renamed the "acfactory" test fixture to "acf"
-  "deltachat-rpc-server>=2.62.0",
+  # 2.62.0 renamed the "acfactory" test fixture to "acf";
+  # BSDs and others have no server wheel, cmsend then uses one from PATH
+  "deltachat-rpc-server>=2.62.0; sys_platform == 'linux' or sys_platform == 'darwin' or sys_platform == 'win32'",
   "deltachat-rpc-client>=2.62.0",
-  "xdg-base-dirs>=6.0.2", 
+  "xdg-base-dirs>=6.0.2",
 ]
 
 dynamic = ["version"]

test_cmsend.py

@@ -22,8 +22,12 @@ def run(*args):
     return run
 
 
-def test_join_and_send(acf, run_cmsend):
+def test_join_and_send(acf, run_cmsend, capfd):
     (ac,) = acf.get_online_accounts(1)
+    # a timed-out --init must not leave a profile behind
+    with pytest.raises(subprocess.CalledProcessError):
+        run_cmsend("--timeout", "0", "--init", ci_chatmail_domain)
+    assert "timed out after 0s" in capfd.readouterr().err
     run_cmsend("--init", ci_chatmail_domain)
 
     run_cmsend("-t", "GROUP", "--join", ac.create_group("log").get_qr_code())
@@ -39,3 +43,9 @@ def test_join_and_send(acf, run_cmsend):
         assert snapshot.text == f"hello {tag}"
     sender = ac.get_contact_by_id(snapshot.from_id).get_snapshot()
     assert sender.display_name == f"cmsend[{socket.gethostname()}]"
+
+    ac.stop_io()
+    invitelink = ac.create_group("offline").get_qr_code()
+    with pytest.raises(subprocess.CalledProcessError):
+        run_cmsend("--timeout", "1", "-t", "OFFLINE", "--join", invitelink)
+    assert "timed out after 1s" in capfd.readouterr().err