feat: respect BACKEND_MUTATE to avoid mutating backend mailboxes by default; add tests and docs

README.md

@@ -30,6 +30,11 @@ Proxy an unauthenticated, unencrypted POP3 / SMTP server to authenticated IMAPS
 - `BACKEND_SMTP_USE_SSL` (default `true`)
 - `BACKEND_SMTP_USE_TLS` (default `false`)
 
+- `BACKEND_MUTATE` (default `false`) - when `true`, POP3 deletions are propagated to the
+  backend IMAP server (STORE +FLAGS / EXPUNGE). Default behaviour is to never mutate
+  the backend mailbox on POP client deletions; the proxy only hides messages for the
+  duration of the client session.
+
 ## Build and run
 
 This project targets the latest Python LTS release. The included `Dockerfile` uses `python:3.12-slim`, which is compatible with Python 3.12 and later LTS releases.

proxy_server.py

@@ -51,6 +51,10 @@ class Settings:
     BACKEND_SMTP_PASS = os.getenv("BACKEND_SMTP_PASS")
     BACKEND_SMTP_USE_SSL = env_bool("BACKEND_SMTP_USE_SSL", True)
     BACKEND_SMTP_USE_TLS = env_bool("BACKEND_SMTP_USE_TLS", False)
+    # When false (default) the proxy will not mutate backend mailboxes
+    # (no STORE +FLAGS / EXPUNGE). Set to true only when deletions should
+    # be propagated to the backend IMAP server.
+    BACKEND_MUTATE = env_bool("BACKEND_MUTATE", False)
 
     @classmethod
     def validate(cls):
@@ -376,9 +380,12 @@ class POP3Session:
 
     async def handle_quit(self):
         if self._imap:
+            # Only propagate deletes to the backend when explicitly enabled.
+            if Settings.BACKEND_MUTATE:
                 for uid in self.deleted:
                     await asyncio.to_thread(self._imap.mark_deleted, uid)
                 await asyncio.to_thread(self._imap.expunge)
+            # Always logout the backend connection.
             self._imap.logout()
         await self.send_line("+OK Goodbye")
 

tests/test_proxy_server.py

@@ -143,6 +143,95 @@ def test_smtp_proxy_handler_forwards_message_over_ssl(monkeypatch):
     Settings.BACKEND_SMTP_PASS = previous_pass
 
 
+def test_pop3_quit_does_not_mutate_backend_by_default():
+    import asyncio
+
+    class DummyBackend:
+        def __init__(self):
+            self.deleted_called = []
+            self.expunge_called = False
+            self.logged_out = False
+
+        def mark_deleted(self, uid):
+            self.deleted_called.append(uid)
+
+        def expunge(self):
+            self.expunge_called = True
+
+        def logout(self):
+            self.logged_out = True
+
+    session = POP3Session(None, None)
+    backend = DummyBackend()
+    session._imap = backend
+    session.deleted = {b"1", b"2"}
+
+    # Provide a dummy writer so send_line can be awaited without a socket.
+    class DummyWriter:
+        def __init__(self):
+            self.buf = b""
+            self.closed = False
+
+        def write(self, data):
+            self.buf += data
+
+        async def drain(self):
+            return None
+
+        def close(self):
+            self.closed = True
+
+        async def wait_closed(self):
+            return None
+
+    session.writer = DummyWriter()
+
+    # Ensure default is not to mutate
+    previous_mutate = Settings.BACKEND_MUTATE
+    Settings.BACKEND_MUTATE = False
+    asyncio.run(session.handle_quit())
+    Settings.BACKEND_MUTATE = previous_mutate
+
+    assert backend.deleted_called == []
+    assert backend.expunge_called is False
+    assert backend.logged_out is True
+
+
+def test_pop3_quit_mutates_backend_when_enabled():
+    import asyncio
+
+    class DummyBackend:
+        def __init__(self):
+            self.deleted_called = []
+            self.expunge_called = False
+            self.logged_out = False
+
+        def mark_deleted(self, uid):
+            self.deleted_called.append(uid)
+
+        def expunge(self):
+            self.expunge_called = True
+
+        def logout(self):
+            self.logged_out = True
+
+    session = POP3Session(None, None)
+    backend = DummyBackend()
+    session._imap = backend
+    session.deleted = {b"1", b"2"}
+
+    session.writer = DummyWriter()
+
+    previous_mutate = Settings.BACKEND_MUTATE
+    Settings.BACKEND_MUTATE = True
+    asyncio.run(session.handle_quit())
+    Settings.BACKEND_MUTATE = previous_mutate
+
+    assert set(backend.deleted_called) == {b"1", b"2"}
+    assert backend.expunge_called is True
+    assert backend.logged_out is True
+
+
 class FakeWriter:
     def __init__(self):
         self.buffer = bytearray()
@@ -160,6 +249,26 @@ class FakeWriter:
         pass
 
 
+class DummyWriter:
+    """Simple writer used by tests where we only need write/drain/close."""
+
+    def __init__(self):
+        self.buf = b""
+        self.closed = False
+
+    def write(self, data):
+        self.buf += data
+
+    async def drain(self):
+        return None
+
+    def close(self):
+        self.closed = True
+
+    async def wait_closed(self):
+        return None
+
+
 class RecordingIMAP:
     """In-memory IMAPBackend stand-in for POP3 session tests."""
 
@@ -278,6 +387,7 @@ def test_dele_survives_stat_list_uidl_until_quit():
     asyncio.run(session.handle_list([]))
     asyncio.run(session.handle_uidl([]))
     assert b"2" in session.deleted
+    # Default behaviour is not to mutate the backend mailbox on QUIT.
     asyncio.run(session.handle_quit())
-    assert session._imap.marked == [b"2"]
-    assert session._imap.expunged is True
+    assert session._imap.marked == []
+    assert session._imap.expunged is False