bongo/contrib/testing/imap-append-check.py

#!/usr/bin/env python3
# This program is free software, licensed under the terms of the GNU GPL.
# See the Bongo COPYING file for full details.
# Copyright (c) 2026 Bongo Project contributors

"""Verify live IMAP APPEND, LITERAL+, and APPENDLIMIT behavior."""

from __future__ import annotations

import os
import re
import socket
import ssl
import time


HOST = os.environ.get("BONGO_TEST_HOST", "127.0.0.1")
PORT = int(os.environ.get("BONGO_TEST_IMAPS_PORT", "993"))
TIMEOUT = float(os.environ.get("BONGO_TEST_TIMEOUT", "15"))
USERNAME = os.environ.get("BONGO_TEST_USER", "test1")
PASSWORD = os.environ.get("BONGO_TEST_PASSWORD", "")
APPENDLIMIT = 999999
TAGGED = re.compile(rb"^([A-Za-z0-9]+) (OK|NO|BAD)(?: .*)?\r\n$")


class IMAPCheckError(RuntimeError):
    """Raised when the live IMAP peer violates the tested contract."""


def quote(value: str) -> str:
    return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"'


class IMAPConnection:
    def __init__(self) -> None:
        raw = socket.create_connection((HOST, PORT), timeout=TIMEOUT)
        context = ssl.create_default_context()
        context.check_hostname = False
        context.verify_mode = ssl.CERT_NONE
        context.minimum_version = ssl.TLSVersion.TLSv1_2
        self.socket = context.wrap_socket(raw, server_hostname=HOST)
        self.socket.settimeout(TIMEOUT)
        self.reader = self.socket.makefile("rb")
        self.counter = 0
        greeting = self.line()
        if not greeting.startswith(b"* OK "):
            raise IMAPCheckError(f"unexpected IMAP greeting: {greeting!r}")

    def close(self) -> None:
        try:
            self.reader.close()
        finally:
            self.socket.close()

    def line(self) -> bytes:
        line = self.reader.readline(65538)
        if not line:
            raise IMAPCheckError("connection closed while waiting for IMAP response")
        if len(line) > 65536 or not line.endswith(b"\r\n"):
            raise IMAPCheckError(f"invalid IMAP response line: {line!r}")
        return line

    def next_tag(self) -> str:
        self.counter += 1
        return f"A{self.counter:04d}"

    def response(
        self, tag: str, expected: tuple[str, ...] = ("OK",)
    ) -> list[bytes]:
        lines: list[bytes] = []
        while True:
            line = self.line()
            match = TAGGED.match(line)
            if match is not None and match.group(1).decode("ascii") == tag:
                status = match.group(2).decode("ascii")
                if status not in expected:
                    raise IMAPCheckError(
                        f"{tag} expected {expected}, received {line!r}"
                    )
                return lines + [line]
            lines.append(line)

    def command(
        self, command: str, expected: tuple[str, ...] = ("OK",)
    ) -> list[bytes]:
        tag = self.next_tag()
        self.socket.sendall(f"{tag} {command}\r\n".encode("utf-8"))
        return self.response(tag, expected)

    def append(
        self,
        mailbox: str,
        payload: bytes,
        *,
        options: str = "",
        nonsynchronizing: bool = False,
        expected: tuple[str, ...] = ("OK",),
    ) -> list[bytes]:
        tag = self.next_tag()
        marker = f"{{{len(payload)}{'+' if nonsynchronizing else ''}}}"
        prefix = f"{tag} APPEND {quote(mailbox)}"
        if options:
            prefix += f" {options}"
        self.socket.sendall(f"{prefix} {marker}\r\n".encode("utf-8"))
        lines: list[bytes] = []
        if not nonsynchronizing:
            continuation = self.line()
            if not continuation.startswith(b"+"):
                raise IMAPCheckError(
                    f"synchronized APPEND did not continue: {continuation!r}"
                )
            lines.append(continuation)
        self.socket.sendall(payload + b"\r\n")
        return lines + self.response(tag, expected)

    def rejected_marker(
        self,
        mailbox: str,
        marker: str,
        *,
        options: str = "",
        payload: bytes | None = None,
    ) -> list[bytes]:
        tag = self.next_tag()
        prefix = f"{tag} APPEND {quote(mailbox)}"
        if options:
            prefix += f" {options}"
        self.socket.sendall(f"{prefix} {marker}\r\n".encode("utf-8"))
        if payload is not None:
            self.socket.sendall(payload + b"\r\n")
        return self.response(tag, ("NO", "BAD"))


def require_line(lines: list[bytes], pattern: bytes, description: str) -> None:
    if not any(re.search(pattern, line, re.IGNORECASE) for line in lines):
        raise IMAPCheckError(f"{description} missing from {lines!r}")


def message(subject: str) -> bytes:
    return (
        b"From: imap08@bongo.test\r\n"
        b"To: test1@bongo.test\r\n"
        + f"Subject: {subject}\r\n".encode("ascii")
        + f"Message-ID: <{subject.lower()}@bongo.test>\r\n".encode("ascii")
        + b"\r\n"
        + b"IMAP-08 APPEND fixture\r\n"
    )


def main() -> int:
    if not PASSWORD:
        raise IMAPCheckError("BONGO_TEST_PASSWORD must be set")

    token = f"{os.getpid():x}-{time.time_ns():x}"
    mailbox = f"Bongo 08 Append {token}"
    missing = f"Bongo 08 Missing {token}"
    client = IMAPConnection()
    created = False
    selected = False
    try:
        client.command(f"LOGIN {quote(USERNAME)} {quote(PASSWORD)}")
        capabilities = client.command("CAPABILITY")
        require_line(
            capabilities,
            rb"(?:^|\s)LITERAL\+(?:\s|\r\n$)",
            "LITERAL+ capability",
        )
        require_line(
            capabilities,
            rb"\bAPPENDLIMIT=999999\b",
            "global APPENDLIMIT capability",
        )

        client.command(f"CREATE {quote(mailbox)}")
        created = True
        status = client.command(f"STATUS {quote(mailbox)} (APPENDLIMIT)")
        require_line(
            status,
            rb"^\* STATUS .+ \(APPENDLIMIT 999999\)\r\n$",
            "mailbox APPENDLIMIT status",
        )

        first = client.append(
            mailbox,
            message("IMAP08-SYNC"),
            options=r'(\Seen \Flagged) "17-Jul-2026 12:34:56 +0200"',
        )
        require_line(
            first,
            rb" OK \[APPENDUID [1-9][0-9]* [1-9][0-9]*\] APPEND completed",
            "synchronized APPENDUID",
        )
        second = client.append(
            mailbox,
            message("IMAP08-LITERAL-PLUS"),
            nonsynchronizing=True,
        )
        if any(line.startswith(b"+") for line in second):
            raise IMAPCheckError(
                f"LITERAL+ unexpectedly received a continuation: {second!r}"
            )
        require_line(
            second,
            rb" OK \[APPENDUID [1-9][0-9]* [1-9][0-9]*\] APPEND completed",
            "LITERAL+ APPENDUID",
        )

        selected_lines = client.command(f"SELECT {quote(mailbox)}")
        selected = True
        require_line(selected_lines, rb"^\* 2 EXISTS\r\n$", "APPEND message count")
        fetched = client.command("FETCH 1 (FLAGS INTERNALDATE RFC822.SIZE)")
        require_line(fetched, rb"FLAGS \([^)]*\\Seen", "APPEND Seen flag")
        require_line(fetched, rb"FLAGS \([^)]*\\Flagged", "APPEND Flagged flag")
        require_line(
            fetched,
            rb'INTERNALDATE "17-Jul-2026 10:34:56 \+0000"',
            "APPEND internal date",
        )
        client.command("UNSELECT")
        selected = False

        invalid_date = client.rejected_marker(
            mailbox,
            f"{{{len(message('IMAP08-BAD-DATE'))}}}",
            options='"31-Feb-2026 12:34:56 +0200"',
        )
        require_line(invalid_date, rb" BAD ", "invalid APPEND date rejection")
        client.command("NOOP")

        invalid_flag = client.rejected_marker(
            mailbox,
            f"{{{len(message('IMAP08-BAD-FLAG'))}}}",
            options=r"(\Bogus)",
        )
        require_line(invalid_flag, rb" BAD ", "invalid APPEND flag rejection")
        client.command("NOOP")

        too_big_sync = client.rejected_marker(
            mailbox, f"{{{APPENDLIMIT + 1}}}"
        )
        require_line(
            too_big_sync,
            rb" NO \[TOOBIG\] APPEND ",
            "synchronized oversized APPEND",
        )
        client.command("NOOP")

        oversized_payload = b"x" * (APPENDLIMIT + 1)
        too_big_plus = client.rejected_marker(
            mailbox,
            f"{{{len(oversized_payload)}+}}",
            payload=oversized_payload,
        )
        require_line(
            too_big_plus,
            rb" NO \[TOOBIG\] APPEND ",
            "LITERAL+ oversized APPEND",
        )
        client.command("NOOP")

        missing_result = client.rejected_marker(
            missing,
            f"{{{len(message('IMAP08-MISSING'))}+}}",
            payload=message("IMAP08-MISSING"),
        )
        require_line(
            missing_result,
            rb" NO \[TRYCREATE\] APPEND",
            "missing-mailbox APPEND",
        )
        client.command("NOOP")

        client.command(f"DELETE {quote(mailbox)}")
        created = False
        client.command("LOGOUT")
    finally:
        if selected:
            try:
                client.command("UNSELECT", ("OK", "NO", "BAD"))
            except (IMAPCheckError, OSError):
                pass
        if created:
            try:
                client.command(f"DELETE {quote(mailbox)}", ("OK", "NO"))
            except (IMAPCheckError, OSError):
                pass
        client.close()

    print(
        "IMAP-08 PASS "
        f"host={HOST}:{PORT} "
        "append=sync/literal+ appendlimit=capability/status/toobig "
        "metadata=flags/date missing=trycreate stream=resynchronized cleanup=yes"
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())