test: add pytest coverage and run tests in CI

.dockerignore

@@ -0,0 +1,6 @@
+__pycache__/
+*.pyc
+*.pyo
+*.pyd
+.env
+*.swp

.gitea/workflows/build-and-publish.yaml

@@ -0,0 +1,158 @@
+name: Build and publish container
+
+on:
+  # On merge to main, only build/release when image-affecting files change;
+  # CI-config, Renovate-config and docs changes do not produce a new image.
+  push:
+    branches: [main]
+    paths:
+      - 'Dockerfile'
+      - '.dockerignore'
+      - 'proxy_server.py'
+      - 'requirements.txt'
+  # Pull requests always run (the build is a required check); no path filter.
+  pull_request:
+    branches: [main]
+  workflow_dispatch:
+
+# A newer run cancels an older in-flight run in the same group (keyed by ref),
+# so a fresh merge to main supersedes the previous build and only the latest
+# release is produced, avoiding tags that would be immediately replaced. Each
+# pull request likewise supersedes only its own earlier runs.
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
+defaults:
+  run:
+    shell: bash
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: write
+      packages: write
+    steps:
+      - name: Checkout
+        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6
+        with:
+          # Full history and tags are required to derive the next version
+          # from the conventional-commit messages since the last release.
+          fetch-depth: 0
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: 3.12
+
+      - name: Install test dependencies
+        run: python -m pip install --upgrade pip && pip install -r requirements.txt
+
+      - name: Run unit tests
+        run: pytest -q
+      - name: Determine registry host
+        run: echo "REGISTRY=${GITHUB_SERVER_URL#*://}" >> "$GITHUB_ENV"
+
+      # Derive the release version from conventional commits since the last
+      # v* tag: feat -> minor, fix/perf -> patch, ! or BREAKING CHANGE -> major.
+      # Anything else (chore, ci, docs, build) produces no release; those builds
+      # are published under a sha-<short> tag only.
+      - name: Compute version and image tags
+        id: version
+        run: |
+          set -euo pipefail
+          image="${REGISTRY}/${GITHUB_REPOSITORY,,}"
+
+          last_tag="$(git tag --list 'v*' --sort=-v:refname | head -n1 || true)"
+          if [ -n "$last_tag" ]; then
+            range="${last_tag}..HEAD"
+            base="${last_tag#v}"
+          else
+            range=""
+            base="0.0.0"
+          fi
+
+          subjects="$(git log ${range} --format='%s')"
+          bodies="$(git log ${range} --format='%B')"
+
+          bump="none"
+          if printf '%s\n' "$bodies" | grep -qiE 'BREAKING[ -]CHANGE' \
+             || printf '%s\n' "$subjects" | grep -qE '^[a-z]+([(][^)]*[)])?!:'; then
+            bump="major"
+          elif printf '%s\n' "$subjects" | grep -qE '^feat([(][^)]*[)])?:'; then
+            bump="minor"
+          elif printf '%s\n' "$subjects" | grep -qE '^(fix|perf)([(][^)]*[)])?:'; then
+            bump="patch"
+          fi
+
+          major="${base%%.*}"
+          rest="${base#*.}"
+          minor="${rest%%.*}"
+          patch="${rest##*.}"
+
+          release="false"
+          if [ "${GITHUB_EVENT_NAME}" = "push" ] && [ "$bump" != "none" ]; then
+            release="true"
+            case "$bump" in
+              major) major=$((major + 1)); minor=0; patch=0 ;;
+              minor) minor=$((minor + 1)); patch=0 ;;
+              patch) patch=$((patch + 1)) ;;
+            esac
+            version="${major}.${minor}.${patch}"
+            {
+              echo "tags<<__EOT__"
+              echo "${image}:${version}"
+              echo "${image}:${major}.${minor}"
+              echo "${image}:${major}"
+              echo "${image}:latest"
+              echo "__EOT__"
+            } >> "$GITHUB_OUTPUT"
+            echo "version=${version}" >> "$GITHUB_OUTPUT"
+          else
+            short="$(git rev-parse --short HEAD)"
+            {
+              echo "tags<<__EOT__"
+              echo "${image}:sha-${short}"
+              echo "__EOT__"
+            } >> "$GITHUB_OUTPUT"
+          fi
+          echo "release=${release}" >> "$GITHUB_OUTPUT"
+          echo "Computed bump=${bump}, release=${release}, base=${base}"
+
+      - name: Set up QEMU
+        uses: docker/setup-qemu-action@06116385d9baf250c9f4dcb4858b16962ea869c3 # v4
+
+      - name: Set up Buildx
+        uses: docker/setup-buildx-action@d7f5e7f509e45cec5c76c4d5afdd7de93d0b3df5 # v4
+
+      - name: Log in to the Gitea container registry
+        if: github.event_name != 'pull_request'
+        uses: docker/login-action@650006c6eb7dba73a995cc03b0b2d7f5ca915bee # v4
+        with:
+          registry: ${{ env.REGISTRY }}
+          username: ${{ github.repository_owner }}
+          password: ${{ secrets.PACKAGES_TOKEN }}
+
+      - name: Build and push
+        uses: docker/build-push-action@f9f3042f7e2789586610d6e8b85c8f03e5195baf # v7
+        with:
+          context: .
+          platforms: linux/amd64,linux/arm64
+          push: ${{ github.event_name != 'pull_request' }}
+          tags: ${{ steps.version.outputs.tags }}
+          labels: |
+            org.opencontainers.image.source=${{ github.server_url }}/${{ github.repository }}
+            org.opencontainers.image.revision=${{ github.sha }}
+
+      # Record the release as an annotated git tag so the next run computes the
+      # following version from it. This push does not re-trigger the workflow,
+      # which only listens on the main branch and pull requests.
+      - name: Tag the release
+        if: steps.version.outputs.release == 'true'
+        run: |
+          set -euo pipefail
+          v="v${{ steps.version.outputs.version }}"
+          git config user.name "${{ github.actor }}"
+          git config user.email "${{ github.actor }}@users.noreply.${REGISTRY}"
+          git tag -a "$v" -m "$v"
+          git push origin "$v"

Dockerfile

@@ -0,0 +1,13 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+ENV PYTHONUNBUFFERED=1
+
+COPY requirements.txt ./
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY proxy_server.py ./
+
+EXPOSE 110 25
+
+CMD ["python", "proxy_server.py"]

README.md

@@ -0,0 +1,63 @@
+# Legacy Email Proxy
+
+Proxy an unauthenticated, unencrypted POP3 / SMTP server to authenticated IMAPS and SMTPS backends.
+
+## Features
+
+- Exposes legacy `POP3` on `0.0.0.0:110` and legacy `SMTP` on `0.0.0.0:25`
+- Forwards POP3 mailbox access to an IMAP backend
+- Forwards SMTP submissions to an SMTPS backend
+- Backend host, ports, and credentials are configured via environment variables
+
+## Environment Variables
+
+- `POP3_BIND_ADDR` (default `0.0.0.0`)
+- `POP3_BIND_PORT` (default `110`)
+- `SMTP_BIND_ADDR` (default `0.0.0.0`)
+- `SMTP_BIND_PORT` (default `25`)
+
+- `BACKEND_IMAP_HOST`
+- `BACKEND_IMAP_PORT` (default `993`)
+- `BACKEND_IMAP_USER`
+- `BACKEND_IMAP_PASS`
+- `BACKEND_IMAP_USE_SSL` (default `true`)
+- `BACKEND_IMAP_USE_STARTTLS` (default `false`)
+
+- `BACKEND_SMTP_HOST`
+- `BACKEND_SMTP_PORT` (default `465`)
+- `BACKEND_SMTP_USER`
+- `BACKEND_SMTP_PASS`
+- `BACKEND_SMTP_USE_SSL` (default `true`)
+- `BACKEND_SMTP_USE_TLS` (default `false`)
+
+## 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.
+
+```bash
+docker build -t legacy-email-proxy .
+docker run --rm -p 110:110 -p 25:25 \
+  -e BACKEND_IMAP_HOST=imap.example.com \
+  -e BACKEND_IMAP_PORT=993 \
+  -e BACKEND_IMAP_USER=imap-user \
+  -e BACKEND_IMAP_PASS=imap-pass \
+  -e BACKEND_SMTP_HOST=smtp.example.com \
+  -e BACKEND_SMTP_PORT=465 \
+  -e BACKEND_SMTP_USER=smtp-user \
+  -e BACKEND_SMTP_PASS=smtp-pass \
+  legacy-email-proxy
+```
+
+## Tests
+
+Run tests locally with:
+
+```bash
+python -m pip install --upgrade pip
+pip install -r requirements.txt
+pytest -q
+```
+
+## Notes
+
+This implementation begins the proxy with a minimal POP3 command set and SMTP delivery path. It is designed to start development on the required application architecture.

proxy_server.py

@@ -0,0 +1,388 @@
+"""Legacy POP3/SMTP proxy to IMAP/SMTP backends.
+
+This module implements a lightweight proxy that exposes legacy, unauthenticated
+POP3 and SMTP interfaces and translates them to authenticated encrypted backend
+connections. It is intended to run inside Docker on a supported Python LTS
+release.
+"""
+
+import asyncio
+import imaplib
+import logging
+import os
+import re
+import smtplib
+import ssl
+from aiosmtpd.controller import Controller
+
+logging.basicConfig(level=logging.INFO, format="[%(asctime)s] %(levelname)s: %(message)s")
+logger = logging.getLogger(__name__)
+
+
+def env_bool(name, default=False):
+    """Return a boolean environment variable value.
+
+    Accepts common truthy strings: 1, true, yes, on.
+    """
+    raw = os.getenv(name)
+    if raw is None:
+        return default
+    return raw.strip().lower() in ("1", "true", "yes", "on")
+
+
+class Settings:
+    """Configuration values read from environment variables."""
+
+    POP3_BIND_ADDR = os.getenv("POP3_BIND_ADDR", "0.0.0.0")
+    POP3_BIND_PORT = int(os.getenv("POP3_BIND_PORT", "110"))
+    SMTP_BIND_ADDR = os.getenv("SMTP_BIND_ADDR", "0.0.0.0")
+    SMTP_BIND_PORT = int(os.getenv("SMTP_BIND_PORT", "25"))
+
+    BACKEND_IMAP_HOST = os.getenv("BACKEND_IMAP_HOST")
+    BACKEND_IMAP_PORT = int(os.getenv("BACKEND_IMAP_PORT", "993"))
+    BACKEND_IMAP_USER = os.getenv("BACKEND_IMAP_USER")
+    BACKEND_IMAP_PASS = os.getenv("BACKEND_IMAP_PASS")
+    BACKEND_IMAP_USE_SSL = env_bool("BACKEND_IMAP_USE_SSL", True)
+    BACKEND_IMAP_USE_STARTTLS = env_bool("BACKEND_IMAP_USE_STARTTLS", False)
+
+    BACKEND_SMTP_HOST = os.getenv("BACKEND_SMTP_HOST")
+    BACKEND_SMTP_PORT = int(os.getenv("BACKEND_SMTP_PORT", "465"))
+    BACKEND_SMTP_USER = os.getenv("BACKEND_SMTP_USER")
+    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)
+
+    @classmethod
+    def validate(cls):
+        """Validate that required backend settings are configured."""
+        missing = []
+        if not cls.BACKEND_IMAP_HOST:
+            missing.append("BACKEND_IMAP_HOST")
+        if not cls.BACKEND_SMTP_HOST:
+            missing.append("BACKEND_SMTP_HOST")
+        if missing:
+            raise RuntimeError(f"Missing required environment variables: {', '.join(missing)}")
+
+
+class IMAPBackend:
+    """Minimal IMAP wrapper for backend mailbox access."""
+
+    UID_PATTERN = re.compile(rb"\(UID (\d+) RFC822\.SIZE (\d+)\)")
+
+    def __init__(self, username, password):
+        self.username = username
+        self.password = password
+        self.connection = None
+
+    def _connect(self):
+        if Settings.BACKEND_IMAP_USE_SSL:
+            client = imaplib.IMAP4_SSL(Settings.BACKEND_IMAP_HOST, Settings.BACKEND_IMAP_PORT)
+        else:
+            client = imaplib.IMAP4(Settings.BACKEND_IMAP_HOST, Settings.BACKEND_IMAP_PORT)
+            if Settings.BACKEND_IMAP_USE_STARTTLS:
+                client.starttls(ssl_context=ssl.create_default_context())
+        return client
+
+    def login(self):
+        """Open an IMAP connection and select INBOX."""
+        client = self._connect()
+        client.login(self.username, self.password)
+        client.select("INBOX")
+        self.connection = client
+        return client
+
+    def logout(self):
+        """Safely close the IMAP connection."""
+        if self.connection is not None:
+            try:
+                self.connection.logout()
+            except Exception:
+                pass
+            self.connection = None
+
+    def list_uids(self):
+        typ, data = self.connection.uid("search", None, "ALL")
+        if typ != "OK":
+            raise RuntimeError("IMAP UID search failed")
+        return data[0].split() if data and data[0] else []
+
+    def fetch_message_size(self, uid):
+        typ, data = self.connection.uid("fetch", uid, "(RFC822.SIZE)")
+        if typ != "OK" or not data:
+            raise RuntimeError("IMAP size fetch failed")
+        response = data[0]
+        if isinstance(response, tuple):
+            response = response[0]
+        match = re.search(rb"RFC822\.SIZE (\d+)", response)
+        return int(match.group(1)) if match else 0
+
+    def fetch_message(self, uid):
+        typ, data = self.connection.uid("fetch", uid, "(RFC822)")
+        if typ != "OK" or not data:
+            raise RuntimeError("IMAP message fetch failed")
+        parts = [chunk for chunk in data if isinstance(chunk, bytes)]
+        return b"\r\n".join(parts) + b"\r\n"
+
+    def mark_deleted(self, uid):
+        self.connection.uid("STORE", uid, "+FLAGS.SILENT", "(\\Deleted)")
+
+    def expunge(self):
+        self.connection.expunge()
+
+
+class POP3Session:
+    """Session state and command handling for a single POP3 client."""
+
+    def __init__(self, reader, writer):
+        self.reader = reader
+        self.writer = writer
+        self._authenticated = False
+        self._imap = None
+        self.username = None
+        self.password = None
+        self.message_ids = []
+        self.deleted = set()
+
+    async def run(self):
+        """Process POP3 commands until the client disconnects."""
+        await self.send_line("+OK POP3 proxy ready")
+        while True:
+            line = await self.reader.readline()
+            if not line:
+                break
+            command = line.decode("utf-8", errors="ignore").strip()
+            if not command:
+                continue
+            logger.info("POP3 command: %s", command)
+            try:
+                response = await self.handle_command(command)
+            except Exception as exc:
+                logger.exception("POP3 handling failed")
+                await self.send_line(f"-ERR {exc}")
+                break
+            if response is False:
+                break
+        await self.close()
+
+    async def handle_command(self, command_line):
+        parts = command_line.split()
+        command = parts[0].upper()
+        args = parts[1:]
+
+        if command == "USER":
+            return await self.handle_user(args)
+        if command == "PASS":
+            return await self.handle_pass(args)
+        if command == "STAT":
+            return await self.handle_stat()
+        if command == "LIST":
+            return await self.handle_list(args)
+        if command == "RETR":
+            return await self.handle_retr(args)
+        if command == "DELE":
+            return await self.handle_dele(args)
+        if command == "NOOP":
+            return await self.send_line("+OK")
+        if command == "RSET":
+            self.deleted.clear()
+            return await self.send_line("+OK")
+        if command == "QUIT":
+            await self.handle_quit()
+            return False
+        if command == "UIDL":
+            return await self.handle_uidl(args)
+        return await self.send_line("-ERR Unsupported command")
+
+    async def handle_user(self, args):
+        if len(args) != 1:
+            return await self.send_line("-ERR USER requires username")
+        self.username = args[0]
+        return await self.send_line("+OK")
+
+    async def handle_pass(self, args):
+        if len(args) != 1:
+            return await self.send_line("-ERR PASS requires password")
+        self.password = args[0]
+        await asyncio.to_thread(self.authenticate)
+        return await self.send_line("+OK User authenticated")
+
+    def authenticate(self):
+        """Authenticate to the IMAP backend using configured credentials."""
+        if Settings.BACKEND_IMAP_USER and Settings.BACKEND_IMAP_PASS:
+            username = Settings.BACKEND_IMAP_USER
+            password = Settings.BACKEND_IMAP_PASS
+        elif self.username and self.password:
+            username = self.username
+            password = self.password
+        else:
+            raise RuntimeError("No IMAP credentials available")
+
+        backend = IMAPBackend(username, password)
+        backend.login()
+        self._imap = backend
+        self._refresh_mailbox()
+
+    def _refresh_mailbox(self):
+        self.message_ids = self._imap.list_uids()
+        self.deleted.clear()
+
+    async def handle_stat(self):
+        self._require_auth()
+        self._refresh_mailbox()
+        count = len(self.message_ids) - len(self.deleted)
+        total_size = sum(self._imap.fetch_message_size(uid) for uid in self.message_ids if uid not in self.deleted)
+        return await self.send_line(f"+OK {count} {total_size}")
+
+    async def handle_list(self, args):
+        self._require_auth()
+        self._refresh_mailbox()
+        if not args:
+            lines = [f"+OK {len(self.message_ids)} messages"]
+            for index, uid in enumerate(self.message_ids, start=1):
+                if uid in self.deleted:
+                    continue
+                size = self._imap.fetch_message_size(uid)
+                lines.append(f"{index} {size}")
+            lines.append(".")
+            return await self.send_lines(lines)
+        if len(args) != 1 or not args[0].isdigit():
+            return await self.send_line("-ERR LIST takes optional message number")
+        message_number = int(args[0])
+        if message_number < 1 or message_number > len(self.message_ids):
+            return await self.send_line("-ERR no such message")
+        uid = self.message_ids[message_number - 1]
+        if uid in self.deleted:
+            return await self.send_line("-ERR message deleted")
+        size = self._imap.fetch_message_size(uid)
+        return await self.send_line(f"+OK {message_number} {size}")
+
+    async def handle_retr(self, args):
+        self._require_auth()
+        if len(args) != 1 or not args[0].isdigit():
+            return await self.send_line("-ERR RETR requires message number")
+        index = int(args[0])
+        if index < 1 or index > len(self.message_ids):
+            return await self.send_line("-ERR no such message")
+        uid = self.message_ids[index - 1]
+        if uid in self.deleted:
+            return await self.send_line("-ERR message deleted")
+        message = await asyncio.to_thread(self._imap.fetch_message, uid)
+        await self.send_line(f"+OK {len(message)} octets")
+        await self.send_raw(message)
+        await self.send_line(".")
+
+    async def handle_dele(self, args):
+        self._require_auth()
+        if len(args) != 1 or not args[0].isdigit():
+            return await self.send_line("-ERR DELE requires message number")
+        index = int(args[0])
+        if index < 1 or index > len(self.message_ids):
+            return await self.send_line("-ERR no such message")
+        uid = self.message_ids[index - 1]
+        self.deleted.add(uid)
+        return await self.send_line("+OK message marked for deletion")
+
+    async def handle_uidl(self, args):
+        self._require_auth()
+        self._refresh_mailbox()
+        if not args:
+            lines = ["+OK UID list follows"]
+            for index, uid in enumerate(self.message_ids, start=1):
+                if uid in self.deleted:
+                    continue
+                lines.append(f"{index} {uid.decode('ascii')}")
+            lines.append(".")
+            return await self.send_lines(lines)
+        if len(args) != 1 or not args[0].isdigit():
+            return await self.send_line("-ERR UIDL takes a single message number")
+        message_number = int(args[0])
+        if message_number < 1 or message_number > len(self.message_ids):
+            return await self.send_line("-ERR no such message")
+        uid = self.message_ids[message_number - 1]
+        if uid in self.deleted:
+            return await self.send_line("-ERR message deleted")
+        return await self.send_line(f"+OK {message_number} {uid.decode('ascii')}")
+
+    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)
+            self._imap.logout()
+        await self.send_line("+OK Goodbye")
+
+    async def send_line(self, line):
+        self.writer.write((line + "\r\n").encode("utf-8"))
+        await self.writer.drain()
+
+    async def send_lines(self, lines):
+        for line in lines:
+            self.writer.write((line + "\r\n").encode("utf-8"))
+        await self.writer.drain()
+
+    async def send_raw(self, data):
+        self.writer.write(data)
+        await self.writer.drain()
+
+    def _require_auth(self):
+        if self._imap is None:
+            raise RuntimeError("Not authenticated")
+
+    async def close(self):
+        try:
+            self.writer.close()
+            await self.writer.wait_closed()
+        except Exception:
+            pass
+
+
+class SMTPProxyHandler:
+    """SMTP handler that relays inbound messages to the backend SMTP server."""
+
+    async def handle_DATA(self, server, session, envelope):
+        logger.info("SMTP message received from %s to %s", envelope.mail_from, envelope.rcpt_tos)
+        await asyncio.to_thread(self.send_message, envelope.mail_from, envelope.rcpt_tos, envelope.content)
+        return "250 Message accepted for delivery"
+
+        def send_message(self, sender, recipients, data):
+        """Forward a complete SMTP message to the backend SMTP server."""
+        message = data.decode("utf-8", errors="replace")
+        if Settings.BACKEND_SMTP_USE_SSL:
+            smtp = smtplib.SMTP_SSL(Settings.BACKEND_SMTP_HOST, Settings.BACKEND_SMTP_PORT, timeout=30)
+        else:
+            smtp = smtplib.SMTP(Settings.BACKEND_SMTP_HOST, Settings.BACKEND_SMTP_PORT, timeout=30)
+            if Settings.BACKEND_SMTP_USE_TLS:
+                smtp.starttls(context=ssl.create_default_context())
+        try:
+            if Settings.BACKEND_SMTP_USER and Settings.BACKEND_SMTP_PASS:
+                smtp.login(Settings.BACKEND_SMTP_USER, Settings.BACKEND_SMTP_PASS)
+            smtp.sendmail(sender, recipients, message)
+        finally:
+            smtp.quit()
+
+
+async def handle_pop3_client(reader, writer):
+    """Create and run a POP3 session for a new client connection."""
+    session = POP3Session(reader, writer)
+    await session.run()
+
+
+async def main():
+    """Initialize the proxy and start the POP3 and SMTP servers."""
+    Settings.validate()
+    logger.info("Starting POP3 server on %s:%s", Settings.POP3_BIND_ADDR, Settings.POP3_BIND_PORT)
+    pop3_server = await asyncio.start_server(handle_pop3_client, Settings.POP3_BIND_ADDR, Settings.POP3_BIND_PORT)
+
+    logger.info("Starting SMTP server on %s:%s", Settings.SMTP_BIND_ADDR, Settings.SMTP_BIND_PORT)
+    smtp_controller = Controller(SMTPProxyHandler(), hostname=Settings.SMTP_BIND_ADDR, port=Settings.SMTP_BIND_PORT)
+    smtp_controller.start()
+
+    async with pop3_server:
+        await pop3_server.serve_forever()
+
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        logger.info("Shutdown requested")

renovate.json

@@ -0,0 +1,17 @@
+{
+  "extends": ["config:base"],
+  "timezone": "UTC",
+  "packageRules": [
+    {
+      "matchPackagePatterns": ["*"],
+      "groupName": "all dependencies",
+      "enabled": true
+    }
+  ],
+  "labels": ["dependencies"],
+  "automerge": false,
+  "prHourlyLimit": 2,
+  "rangeStrategy": "bump",
+  "postUpdateOptions": ["gomodTidy"],
+  "ignorePaths": ["**/node_modules/**"]
+}

requirements.txt

@@ -0,0 +1,2 @@
+aiosmtpd>=1.6.3
+pytest>=8.0.0

tests/test_proxy_server.py

@@ -0,0 +1,121 @@
+import imaplib
+import smtplib
+
+import pytest
+
+from proxy_server import IMAPBackend, SMTPProxyHandler, Settings, env_bool
+
+
+class DummyIMAP:
+    def __init__(self, host, port):
+        self.host = host
+        self.port = port
+        self.logged_in = False
+        self.selected = None
+
+    def login(self, username, password):
+        self.logged_in = True
+
+    def select(self, mailbox):
+        self.selected = mailbox
+
+    def uid(self, command, *args):
+        if command == "search":
+            return "OK", [b"1 2 3"]
+        if command == "fetch" and args[1] == "(RFC822.SIZE)":
+            return "OK", [(b"1 (RFC822.SIZE 1024)", b"")]
+        if command == "fetch" and args[1] == "(RFC822)":
+            return "OK", [b"1 (RFC822 {10}", b"Hello", b" World", b")"]
+        return "NO", []
+
+    def logout(self):
+        pass
+
+    def expunge(self):
+        pass
+
+
+def test_env_bool_interprets_truthy_values(monkeypatch):
+    monkeypatch.setenv("FEATURE_ENABLED", "true")
+    assert env_bool("FEATURE_ENABLED") is True
+    monkeypatch.setenv("FEATURE_ENABLED", "1")
+    assert env_bool("FEATURE_ENABLED") is True
+    monkeypatch.setenv("FEATURE_ENABLED", "off")
+    assert env_bool("FEATURE_ENABLED") is False
+    monkeypatch.delenv("FEATURE_ENABLED", raising=False)
+    assert env_bool("FEATURE_ENABLED", default=True) is True
+
+
+def test_settings_validate_requires_backend_hosts():
+    original_imap = Settings.BACKEND_IMAP_HOST
+    original_smtp = Settings.BACKEND_SMTP_HOST
+    Settings.BACKEND_IMAP_HOST = None
+    Settings.BACKEND_SMTP_HOST = None
+    with pytest.raises(RuntimeError, match="Missing required environment variables"):
+        Settings.validate()
+    Settings.BACKEND_IMAP_HOST = original_imap
+    Settings.BACKEND_SMTP_HOST = original_smtp
+
+
+def test_settings_validate_succeeds_with_backends():
+    original_imap = Settings.BACKEND_IMAP_HOST
+    original_smtp = Settings.BACKEND_SMTP_HOST
+    Settings.BACKEND_IMAP_HOST = "imap.example.com"
+    Settings.BACKEND_SMTP_HOST = "smtp.example.com"
+    Settings.validate()
+    Settings.BACKEND_IMAP_HOST = original_imap
+    Settings.BACKEND_SMTP_HOST = original_smtp
+
+
+def test_imap_backend_can_login_and_fetch(monkeypatch):
+    monkeypatch.setattr(imaplib, "IMAP4_SSL", lambda host, port: DummyIMAP(host, port))
+    backend = IMAPBackend("user", "pass")
+    backend.login()
+    assert backend.connection.logged_in
+    assert backend.connection.selected == "INBOX"
+    assert backend.list_uids() == [b"1", b"2", b"3"]
+    assert backend.fetch_message_size(b"1") == 1024
+    assert b"Hello" in backend.fetch_message(b"1")
+
+
+def test_smtp_proxy_handler_forwards_message_over_ssl(monkeypatch):
+    captured = {}
+
+    class DummySMTP:
+        def __init__(self, host, port, timeout=0):
+            captured["instance"] = self
+            self.host = host
+            self.port = port
+            self.timeout = timeout
+            self.logged_in = False
+            self.sent = None
+
+        def login(self, user, password):
+            self.logged_in = True
+
+        def sendmail(self, sender, recipients, message):
+            self.sent = (sender, recipients, message)
+
+        def quit(self):
+            captured["quit"] = True
+
+    monkeypatch.setattr(smtplib, "SMTP_SSL", DummySMTP)
+    previous_ssl = Settings.BACKEND_SMTP_USE_SSL
+    previous_user = Settings.BACKEND_SMTP_USER
+    previous_pass = Settings.BACKEND_SMTP_PASS
+    Settings.BACKEND_SMTP_USE_SSL = True
+    Settings.BACKEND_SMTP_USER = "smtp-user"
+    Settings.BACKEND_SMTP_PASS = "smtp-pass"
+
+    handler = SMTPProxyHandler()
+    handler.send_message("from@example.com", ["to@example.com"], b"Subject: Test\r\n\r\nBody\r\n")
+
+    assert captured["instance"].host == Settings.BACKEND_SMTP_HOST
+    assert captured["instance"].logged_in is True
+    assert captured["instance"].sent[0] == "from@example.com"
+    assert captured["instance"].sent[1] == ["to@example.com"]
+    assert b"Subject: Test" in captured["instance"].sent[2]
+
+    Settings.BACKEND_SMTP_USE_SSL = previous_ssl
+    Settings.BACKEND_SMTP_USER = previous_user
+    Settings.BACKEND_SMTP_PASS = previous_pass