Store raw tokens instead of dictionaries in metadata

This is recommended to prevent SPF failure
from rejecting the message early in case messages
are remailed without breaking DKIM.

chatmaild/src/chatmaild/chatmail-metadata.service.f

@@ -2,7 +2,7 @@
 Description=Chatmail dict proxy for IMAP METADATA
 
 [Service]
-ExecStart={execpath} /run/dovecot/metadata.socket vmail {config_path}
+ExecStart={execpath} /run/dovecot/metadata.socket vmail {config_path} /home/vmail/metadata
 Restart=always
 RestartSec=30
 

chatmaild/src/chatmaild/metadata.py

@@ -1,75 +1,129 @@
 import pwd
+
+from pathlib import Path
+from queue import Queue
+from threading import Thread
 from socketserver import (
     UnixStreamServer,
     StreamRequestHandler,
     ThreadingMixIn,
 )
-from .config import read_config, Config
+from .config import read_config
 import sys
 import logging
 import os
 import requests
 
 
-def handle_dovecot_protocol(rfile, wfile, tokens, requests_session, config: Config):
+DICTPROXY_LOOKUP_CHAR = "L"
+DICTPROXY_ITERATE_CHAR = "I"
+DICTPROXY_SET_CHAR = "S"
+DICTPROXY_BEGIN_TRANSACTION_CHAR = "B"
+DICTPROXY_COMMIT_TRANSACTION_CHAR = "C"
+DICTPROXY_TRANSACTION_CHARS = "SBC"
+
+
+class Notifier:
+    def __init__(self, metadata_dir):
+        self.metadata_dir = metadata_dir
+        self.to_notify_queue = Queue()
+
+    def set_token(self, guid, token):
+        guid_path = self.metadata_dir.joinpath(guid)
+        write_path = guid_path.with_suffix(".tmp")
+        write_path.write_text(token)
+        write_path.rename(guid_path)
+
+    def del_token(self, guid):
+        self.metadata_dir.joinpath(guid).unlink(missing_ok=True)
+
+    def get_token(self, guid):
+        guid_path = self.metadata_dir / guid
+        if guid_path.exists():
+            return guid_path.read_text()
+
+    def new_message_for_guid(self, guid):
+        self.to_notify_queue.put(guid)
+
+    def thread_run_loop(self):
+        requests_session = requests.Session()
+        while 1:
+            self.thread_run_one(requests_session)
+
+    def thread_run_one(self, requests_session):
+        guid = self.to_notify_queue.get()
+        token = self.get_token(guid)
+        if token:
+            response = requests_session.post(
+                "https://notifications.delta.chat/notify",
+                data=token,
+                timeout=60,
+            )
+            if response.status_code == 410:
+                # 410 Gone status code
+                # means the token is no longer valid.
+                self.del_token(guid)
+
+
+def handle_dovecot_protocol(rfile, wfile, notifier):
     # HELLO message, ignored.
     msg = rfile.readline().strip().decode()
 
     transactions = {}
-
     while True:
         msg = rfile.readline().strip().decode()
         if not msg:
             break
 
-        short_command = msg[0]
-        if short_command == "L":
-            wfile.write(b"N\n")
-        elif short_command == "S":
-            # See header of
-            # <https://github.com/dovecot/core/blob/5e7965632395793d9355eb906b173bf28d2a10ca/src/lib-storage/mailbox-attribute.h>
-            # for the documentation on the structure of the key.
+        res = handle_dovecot_request(msg, transactions, notifier)
+        if res:
+            wfile.write(res.encode("ascii"))
+            wfile.flush()
 
-            # Request GETMETADATA "INBOX" /private/chatmail
-            # results in a query for
-            # priv/dd72550f05eadc65542a1200cac67ad7/chatmail
-            #
-            # Request GETMETADATA "" /private/chatmail
-            # results in
-            # priv/dd72550f05eadc65542a1200cac67ad7/vendor/vendor.dovecot/pvt/server/chatmail
 
-            parts = msg[1:].split("\t")
-            transaction_id = parts[0]
-            keyname = parts[1].split("/")
-            value = parts[2] if len(parts) > 2 else ""
-            if keyname[0] == "priv" and keyname[2] == "devicetoken":
-                tokens[keyname[1]] = value
-            elif keyname[0] == "priv" and keyname[2] == "messagenew":
-                guid = keyname[1]
-                token = tokens.get(guid)
-                if token:
-                    response = requests_session.post(
-                        "https://notifications.delta.chat/notify",
-                        data=token,
-                        timeout=60,
-                    )
-                    if response.status_code == 410:
-                        # 410 Gone status code
-                        # means the token is no longer valid.
-                        del tokens[guid]
-            else:
-                # Transaction failed.
-                transactions[transaction_id] = b"F\n"
-        elif short_command == "B":
-            # Begin transaction.
-            transaction_id = msg[1:].split("\t")[0]
-            transactions[transaction_id] = b"O\n"
-        elif short_command == "C":
-            # Commit transaction.
-            transaction_id = msg[1:].split("\t")[0]
-            wfile.write(transactions.pop(transaction_id, b"N\n"))
+def handle_dovecot_request(msg, transactions, notifier):
+    # see https://doc.dovecot.org/3.0/developer_manual/design/dict_protocol/
+    short_command = msg[0]
+    parts = msg[1:].split("\t")
+    if short_command == DICTPROXY_LOOKUP_CHAR:
+        return "N\n"
+    elif short_command == DICTPROXY_ITERATE_CHAR:
+        # Empty line means ITER_FINISHED.
+        # If we don't return empty line Dovecot will timeout.
+        return "\n"
 
-        wfile.flush()
+    if short_command not in (DICTPROXY_TRANSACTION_CHARS):
+        return
+
+    transaction_id = parts[0]
+
+    if short_command == DICTPROXY_BEGIN_TRANSACTION_CHAR:
+        transactions[transaction_id] = "O\n"
+    elif short_command == DICTPROXY_COMMIT_TRANSACTION_CHAR:
+        # returns whether it failed or succeeded.
+        return transactions.pop(transaction_id, "N\n")
+    elif short_command == DICTPROXY_SET_CHAR:
+        # See header of
+        # <https://github.com/dovecot/core/blob/5e7965632395793d9355eb906b173bf28d2a10ca/src/lib-storage/mailbox-attribute.h>
+        # for the documentation on the structure of the key.
+
+        # Request GETMETADATA "INBOX" /private/chatmail
+        # results in a query for
+        # priv/dd72550f05eadc65542a1200cac67ad7/chatmail
+        #
+        # Request GETMETADATA "" /private/chatmail
+        # results in
+        # priv/dd72550f05eadc65542a1200cac67ad7/vendor/vendor.dovecot/pvt/server/chatmail
+
+        keyname = parts[1].split("/")
+        value = parts[2] if len(parts) > 2 else ""
+        if keyname[0] == "priv" and keyname[2] == "devicetoken":
+            notifier.set_token(keyname[1], value)
+        elif keyname[0] == "priv" and keyname[2] == "messagenew":
+            notifier.new_message_for_guid(keyname[1])
+        else:
+            # Transaction failed.
+            transactions[transaction_id] = "F\n"
 
 
 class ThreadedUnixStreamServer(ThreadingMixIn, UnixStreamServer):
@@ -77,18 +131,20 @@ class ThreadedUnixStreamServer(ThreadingMixIn, UnixStreamServer):
 
 
 def main():
-    socket, username, config = sys.argv[1:]
+    socket, username, config, metadata_dir = sys.argv[1:]
     passwd_entry = pwd.getpwnam(username)
+
+    # XXX config is not currently used
     config = read_config(config)
-    tokens = {}
-    requests_session = requests.Session()
+    metadata_dir = Path(metadata_dir)
+    if not metadata_dir.exists():
+        metadata_dir.mkdir()
+    notifier = Notifier(metadata_dir)
 
     class Handler(StreamRequestHandler):
         def handle(self):
             try:
-                handle_dovecot_protocol(
-                    self.rfile, self.wfile, tokens, requests_session, config
-                )
+                handle_dovecot_protocol(self.rfile, self.wfile, notifier)
             except Exception:
                 logging.exception("Exception in the handler")
                 raise
@@ -98,6 +154,13 @@ def main():
     except FileNotFoundError:
         pass
 
+    # start notifier thread for signalling new messages to
+    # Delta Chat notification server
+
+    t = Thread(target=notifier.thread_run_loop)
+    t.setDaemon(True)
+    t.start()
+
     with ThreadedUnixStreamServer(socket, Handler) as server:
         os.chown(socket, uid=passwd_entry.pw_uid, gid=passwd_entry.pw_gid)
         try:

chatmaild/src/chatmaild/tests/test_metadata.py

@@ -0,0 +1,163 @@
+import io
+import pytest
+
+from chatmaild.metadata import (
+    handle_dovecot_request,
+    handle_dovecot_protocol,
+    Notifier,
+)
+
+
+@pytest.fixture
+def notifier(tmp_path):
+    metadata_dir = tmp_path.joinpath("metadata")
+    metadata_dir.mkdir()
+    return Notifier(metadata_dir)
+
+
+def test_notifier_persistence(tmp_path):
+    metadata_dir = tmp_path.joinpath("metadata")
+    metadata_dir.mkdir()
+    notifier1 = Notifier(metadata_dir)
+    notifier2 = Notifier(metadata_dir)
+    assert notifier1.get_token(guid="guid00") is None
+    assert notifier2.get_token(guid="guid00") is None
+
+    notifier1.set_token("guid00", "01234")
+    notifier1.set_token("guid03", "456")
+    assert notifier2.get_token("guid00") == "01234"
+    assert notifier2.get_token("guid03") == "456"
+    notifier2.del_token("guid00")
+    assert notifier1.get_token("guid00") is None
+
+
+def test_handle_dovecot_request_lookup_fails(notifier):
+    res = handle_dovecot_request("Lpriv/123/chatmail", {}, notifier)
+    assert res == "N\n"
+
+
+def test_handle_dovecot_request_happy_path(notifier):
+    transactions = {}
+
+    # lookups return the same NOTFOUND result
+    res = handle_dovecot_request("Lpriv/123/chatmail", transactions, notifier)
+    assert res == "N\n"
+    assert notifier.get_token("guid00") is None and not transactions
+
+    # set device token in a transaction
+    tx = "1111"
+    msg = f"B{tx}\tuser"
+    res = handle_dovecot_request(msg, transactions, notifier)
+    assert not res and notifier.get_token("guid00") is None
+    assert transactions == {tx: "O\n"}
+
+    msg = f"S{tx}\tpriv/guid00/devicetoken\t01234"
+    res = handle_dovecot_request(msg, transactions, notifier)
+    assert not res
+    assert len(transactions) == 1
+    assert notifier.get_token("guid00") == "01234"
+
+    msg = f"C{tx}"
+    res = handle_dovecot_request(msg, transactions, notifier)
+    assert res == "O\n"
+    assert len(transactions) == 0
+    assert notifier.get_token("guid00") == "01234"
+
+    # trigger notification for incoming message
+    assert handle_dovecot_request(f"B{tx}\tuser", transactions, notifier) is None
+    msg = f"S{tx}\tpriv/guid00/messagenew"
+    assert handle_dovecot_request(msg, transactions, notifier) is None
+    assert notifier.to_notify_queue.get() == "guid00"
+    assert notifier.to_notify_queue.qsize() == 0
+    assert handle_dovecot_request(f"C{tx}\tuser", transactions, notifier) == "O\n"
+    assert not transactions
+
+
+def test_handle_dovecot_protocol_set_devicetoken(notifier):
+    rfile = io.BytesIO(
+        b"\n".join(
+            [
+                b"HELLO",
+                b"Btx00\tuser",
+                b"Stx00\tpriv/guid00/devicetoken\t01234",
+                b"Ctx00",
+            ]
+        )
+    )
+    wfile = io.BytesIO()
+    handle_dovecot_protocol(rfile, wfile, notifier)
+    assert notifier.get_token("guid00") == "01234"
+    assert wfile.getvalue() == b"O\n"
+
+
+def test_handle_dovecot_protocol_iterate(notifier):
+    rfile = io.BytesIO(
+        b"\n".join(
+            [
+                b"H",
+                b"I9\t0\tpriv/5cbe730f146fea6535be0d003dd4fc98/\tci-2dzsrs@nine.testrun.org",
+            ]
+        )
+    )
+    wfile = io.BytesIO()
+    handle_dovecot_protocol(rfile, wfile, notifier)
+    assert wfile.getvalue() == b"\n"
+
+
+def test_handle_dovecot_protocol_messagenew(notifier):
+    rfile = io.BytesIO(
+        b"\n".join(
+            [
+                b"HELLO",
+                b"Btx01\tuser",
+                b"Stx01\tpriv/guid00/messagenew",
+                b"Ctx01",
+            ]
+        )
+    )
+    wfile = io.BytesIO()
+    handle_dovecot_protocol(rfile, wfile, notifier)
+    assert wfile.getvalue() == b"O\n"
+    assert notifier.to_notify_queue.get() == "guid00"
+    assert notifier.to_notify_queue.qsize() == 0
+
+
+def test_notifier_thread_run(notifier):
+    requests = []
+
+    class ReqMock:
+        def post(self, url, data, timeout):
+            requests.append((url, data, timeout))
+
+            class Result:
+                status_code = 200
+
+            return Result()
+
+    notifier.set_token("guid00", "01234")
+    notifier.new_message_for_guid("guid00")
+    notifier.thread_run_one(ReqMock())
+    url, data, timeout = requests[0]
+    assert data == "01234"
+    assert notifier.get_token("guid00") == "01234"
+
+
+def test_notifier_thread_run_gone_removes_token(notifier):
+    requests = []
+
+    class ReqMock:
+        def post(self, url, data, timeout):
+            requests.append((url, data, timeout))
+
+            class Result:
+                status_code = 410
+
+            return Result()
+
+    notifier.set_token("guid00", "01234")
+    notifier.new_message_for_guid("guid00")
+    assert notifier.get_token("guid00") == "01234"
+    notifier.thread_run_one(ReqMock())
+    url, data, timeout = requests[0]
+    assert data == "01234"
+    assert notifier.get_token("guid00") is None

cmdeploy/src/cmdeploy/__init__.py

@@ -350,15 +350,18 @@ def _configure_dovecot(config: Config, debug: bool = False) -> bool:
     need_restart |= lua_push_notification_script.changed
 
     sieve_script = files.put(
-        src=importlib.resources.files(__package__).joinpath(
-            "dovecot/default.sieve"
-        ),
+        src=importlib.resources.files(__package__).joinpath("dovecot/default.sieve"),
         dest="/etc/dovecot/default.sieve",
         user="root",
         group="root",
         mode="644",
     )
     need_restart |= sieve_script.changed
+    if sieve_script.changed:
+        server.shell(
+            name="compile sieve script",
+            commands=["/usr/bin/sievec /etc/dovecot/default.sieve"],
+        )
 
     files.template(
         src=importlib.resources.files(__package__).joinpath("dovecot/expunge.cron.j2"),
@@ -450,7 +453,9 @@ def check_config(config):
         blocked_words = "merlinux schmieder testrun.org".split()
         for key in config.__dict__:
             value = config.__dict__[key]
-            if key.startswith("privacy") and any(x in str(value) for x in blocked_words):
+            if key.startswith("privacy") and any(
+                x in str(value) for x in blocked_words
+            ):
                 raise ValueError(
                     f"please set your own privacy contacts/addresses in {config._inipath}"
                 )
@@ -471,6 +476,10 @@ def deploy_chatmail(config_path: Path) -> None:
     apt.update(name="apt update", cache_time=24 * 3600)
     server.group(name="Create vmail group", group="vmail", system=True)
     server.user(name="Create vmail user", user="vmail", group="vmail", system=True)
+    apt.packages(
+        name="Install rsync",
+        packages=["rsync"],
+    )
 
     # Run local DNS resolver `unbound`.
     # `resolvconf` takes care of setting up /etc/resolv.conf

cmdeploy/src/cmdeploy/chatmail.zone.f

@@ -6,7 +6,7 @@ _submissions._tcp.{chatmail_domain}. SRV 0 1 465 {chatmail_domain}.
 _imap._tcp.{chatmail_domain}.        SRV 0 1 143 {chatmail_domain}.
 _imaps._tcp.{chatmail_domain}.       SRV 0 1 993 {chatmail_domain}.
 {chatmail_domain}.                   CAA 128 issue "letsencrypt.org;accounturi={acme_account_url}"
-{chatmail_domain}.                   TXT "v=spf1 a:{chatmail_domain} -all"
+{chatmail_domain}.                   TXT "v=spf1 a:{chatmail_domain} ~all"
 _dmarc.{chatmail_domain}.            TXT "v=DMARC1;p=reject;adkim=s;aspf=s"
 _mta-sts.{chatmail_domain}.          TXT "v=STSv1; id={sts_id}"
 mta-sts.{chatmail_domain}.           CNAME {chatmail_domain}.

cmdeploy/src/cmdeploy/dns.py

@@ -5,8 +5,6 @@ import importlib
 import subprocess
 import datetime
 
-from typing import Optional
-
 
 class DNS:
     def __init__(self, out, mail_domain):

cmdeploy/src/cmdeploy/dovecot/default.sieve

@@ -1,5 +1,7 @@
 require ["imap4flags"];
 
+# flag the message so it doesn't cause a push notification
+
 if header :is ["Auto-Submitted"] ["auto-replied", "auto-generated"] {
 	addflag "$Auto";
 }

cmdeploy/src/cmdeploy/dovecot/dovecot.conf.j2

@@ -13,6 +13,12 @@ auth_cache_size = 100M
 mail_debug = yes
 {% endif %}
 
+# Prevent warnings similar to:
+#   config: Warning: service auth { client_limit=1000 } is lower than required under max. load (10200). Counted for protocol services with service_count != 1: service lmtp { process_limit=100 } + service imap-urlauth-login { process_limit=100 } + service imap-login { process_limit=10000 }
+#   config: Warning: service anvil { client_limit=1000 } is lower than required under max. load (10103). Counted with: service imap-urlauth-login { process_limit=100 } + service imap-login { process_limit=10000 } + service auth { process_limit=1 }
+#   master: Warning: service(stats): client_limit (1000) reached, client connections are being dropped
+default_client_limit = 20000
+
 mail_server_admin = mailto:root@{{ config.mail_domain }}
 mail_server_comment = Chatmail server
 

cmdeploy/src/cmdeploy/dovecot/expunge.cron.j2

@@ -8,3 +8,4 @@
 # or only temporary (but then they shouldn't be around after {{ config.delete_mails_after }} days anyway).
 2 0 * * * vmail find /home/vmail/mail/{{ config.mail_domain }} -path '*/tmp/*' -mtime +{{ config.delete_mails_after }} -type f -delete
 2 0 * * * vmail find /home/vmail/mail/{{ config.mail_domain }} -path '*/.*/tmp/*' -mtime +{{ config.delete_mails_after }} -type f -delete
+3 0 * * * vmail find /home/vmail/mail/{{ config.mail_domain }} -name 'maildirsize' -type f -delete

cmdeploy/src/cmdeploy/tests/online/test_2_deltachat.py

@@ -63,7 +63,7 @@ class TestEndToEndDeltaChat:
 
         addr = ac2.get_config("addr").lower()
         saved_ok = 0
-        for line in remote.iter_output("journalctl -f -u dovecot"):
+        for line in remote.iter_output("journalctl -n0 -f -u dovecot"):
             if addr not in line:
                 # print(line)
                 continue
@@ -112,7 +112,7 @@ class TestEndToEndDeltaChat:
         lp.sec("ac1 sends a message and ac2 marks it as seen")
         chat = ac1.create_chat(ac2)
         msg = chat.send_text("hi")
-        m = ac2.wait_next_incoming_message()
+        m = ac2._evtracker.wait_next_incoming_message()
         m.mark_seen()
         # we can only indirectly wait for mark-seen to cause an smtp-error
         lp.sec("try to wait for markseen to complete and check error states")
@@ -132,7 +132,7 @@ def test_hide_senders_ip_address(cmfactory):
     chat = cmfactory.get_accepted_chat(user1, user2)
 
     chat.send_text("testing submission header cleanup")
-    user2.wait_next_incoming_message()
+    user2._evtracker.wait_next_incoming_message()
     user2.direct_imap.select_folder("Inbox")
     msg = user2.direct_imap.get_all_messages()[0]
     assert public_ip not in msg.obj.as_string()
@@ -146,5 +146,5 @@ def test_echobot(cmfactory, chatmail_config, lp):
     text = "hi, I hope you text me back"
     chat.send_text(text)
     lp.sec("Wait for reply from echobot")
-    reply = ac.wait_next_incoming_message()
+    reply = ac._evtracker.wait_next_incoming_message()
     assert reply.text == text