From 4bde4f884db2150b1b5ae5d2ac3e3d7e82ab2567 Mon Sep 17 00:00:00 2001 From: Lyra Thorpe Date: Wed, 17 Jun 2026 18:43:22 +0100 Subject: [PATCH] feat: respect BACKEND_MUTATE to avoid mutating backend mailboxes by default; add tests and docs --- README.md | 5 ++ proxy_server.py | 13 ++++- tests/test_proxy_server.py | 114 ++++++++++++++++++++++++++++++++++++- 3 files changed, 127 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index b1f6eb5..4dd0a20 100644 --- a/README.md +++ b/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. diff --git a/proxy_server.py b/proxy_server.py index 6b7933f..16ced93 100644 --- a/proxy_server.py +++ b/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: - for uid in self.deleted: - await asyncio.to_thread(self._imap.mark_deleted, uid) - await asyncio.to_thread(self._imap.expunge) + # 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") diff --git a/tests/test_proxy_server.py b/tests/test_proxy_server.py index 139956a..4df30f7 100644 --- a/tests/test_proxy_server.py +++ b/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