#!/usr/bin/env python3 """Migrate mailbox data from a Mailu installation into Stalwart. Mailu stores mail as standard Dovecot Maildir under its ``dovecot_mail`` volume (one ``/`` tree per account, subfolders as ``.Name`` maildir folders). Reading that volume directly means NO live Mailu is required — the migration works against a stopped or legacy-parked instance and the test harness needs no full platform deploy. Scope (v1, deliberate): - Migrates MESSAGES (folder structure, flags, internal dates) per account via IMAP APPEND into Stalwart. - Idempotent: a message whose Message-ID already exists in the target folder is skipped, so re-runs converge instead of duplicating. - Accounts/aliases are NOT created here: on the platform they are provisioned from the inventory (web-app-stalwart tasks/04_manage_user.yml); the test harness creates its fixture account via the same JMAP API. - Sieve filters and CalDAV/CardDAV data are out of scope (documented in the role README). Usage: migrate_from_mailu.py --maildir-root /mail --imap-host mail.example.org \ --accounts-file accounts.json [--imap-port 993] [--imap-insecure] \ [--dest-separator /] ``accounts.json``: {"": "", ...} — the credentials of the already-provisioned Stalwart accounts (inventory passwords on the platform; fixture passwords in the test). """ from __future__ import annotations import argparse import contextlib import email.parser import imaplib import json import ssl import sys import time from dataclasses import dataclass, field from pathlib import Path _MAILDIR_FLAG_MAP = { "S": "\\Seen", "R": "\\Answered", "F": "\\Flagged", "T": "\\Deleted", "D": "\\Draft", } @dataclass class FolderStats: appended: int = 0 skipped: int = 0 failed: int = 0 @dataclass class AccountStats: folders: dict[str, FolderStats] = field(default_factory=dict) def folder(self, name: str) -> FolderStats: return self.folders.setdefault(name, FolderStats()) @property def failed(self) -> int: return sum(f.failed for f in self.folders.values()) def maildir_flags_to_imap(filename: str) -> str: """Map a maildir file's ``:2,`` suffix to an IMAP flag list string.""" if ":2," not in filename: return "" info = filename.rsplit(":2,", 1)[1] flags = [_MAILDIR_FLAG_MAP[c] for c in info if c in _MAILDIR_FLAG_MAP] return "(" + " ".join(flags) + ")" if flags else "" def maildir_folder_to_imap(maildir_name: str, separator: str) -> str: """Map a maildir folder dir name to the destination IMAP mailbox name. ``""`` (the maildir root) is INBOX; ``.Archive.2024`` becomes ``Archive2024`` (dovecot uses ``.`` as the on-disk separator). """ if not maildir_name: return "INBOX" return maildir_name.lstrip(".").replace(".", separator) def iter_maildir_folders(account_dir: Path) -> list[tuple[str, Path]]: """Yield (maildir_folder_name, path) for the root and every ``.Sub`` dir.""" folders: list[tuple[str, Path]] = [] if (account_dir / "cur").is_dir() or (account_dir / "new").is_dir(): folders.append(("", account_dir)) folders.extend( (child.name, child) for child in sorted(account_dir.iterdir()) if child.is_dir() and child.name.startswith(".") and ((child / "cur").is_dir() or (child / "new").is_dir()) ) return folders def iter_messages(folder_dir: Path) -> list[Path]: """All message files of one maildir folder (cur = seen-ish, new = unseen).""" messages: list[Path] = [] for sub in ("cur", "new"): subdir = folder_dir / sub if subdir.is_dir(): messages.extend(p for p in sorted(subdir.iterdir()) if p.is_file()) return messages def extract_message_id(raw: bytes) -> str: """Message-ID header value from raw RFC-5322 bytes ('' when absent).""" parser = email.parser.BytesHeaderParser() try: return (parser.parsebytes(raw).get("Message-ID") or "").strip() except Exception: return "" _FETCH_CHUNK = 500 def existing_message_ids(imap: imaplib.IMAP4) -> set[str]: """Message-IDs already present in the SELECTed folder. Fetched client-side (one header-fields FETCH per chunk) instead of ``SEARCH HEADER``: Stalwart indexes headers asynchronously, so a server-side search right after APPEND misses fresh messages — the harness's idempotency step caught exactly that as duplicates. """ typ, nums = imap.search(None, "ALL") if typ != "OK" or not nums or not nums[0].split(): return set() sequence = [n.decode() for n in nums[0].split()] ids: set[str] = set() for start in range(0, len(sequence), _FETCH_CHUNK): chunk = ",".join(sequence[start : start + _FETCH_CHUNK]) typ, data = imap.fetch(chunk, "(BODY.PEEK[HEADER.FIELDS (MESSAGE-ID)])") if typ != "OK": continue for part in data or []: if isinstance(part, tuple): message_id = extract_message_id(part[1]) if message_id: ids.add(message_id) return ids def ensure_mailbox(imap: imaplib.IMAP4, mailbox: str) -> None: """CREATE the mailbox; 'already exists' answers are fine.""" if mailbox.upper() == "INBOX": return imap.create(mailbox) def migrate_account( imap: imaplib.IMAP4, account_dir: Path, separator: str, stats: AccountStats, ) -> None: for maildir_name, folder_dir in iter_maildir_folders(account_dir): mailbox = maildir_folder_to_imap(maildir_name, separator) fstats = stats.folder(mailbox) ensure_mailbox(imap, mailbox) typ, _ = imap.select(mailbox) if typ != "OK": print(f" [FAIL] cannot SELECT {mailbox!r}", file=sys.stderr) fstats.failed += len(iter_messages(folder_dir)) continue present = existing_message_ids(imap) for msg_path in iter_messages(folder_dir): raw = msg_path.read_bytes() message_id = extract_message_id(raw) if message_id and message_id in present: fstats.skipped += 1 continue present.add(message_id) flags = maildir_flags_to_imap(msg_path.name) internal_date = imaplib.Time2Internaldate(msg_path.stat().st_mtime) typ, data = imap.append(mailbox, flags or None, internal_date, raw) if typ == "OK": fstats.appended += 1 else: print(f" [FAIL] APPEND {msg_path.name}: {data}", file=sys.stderr) fstats.failed += 1 def connect(host: str, port: int, insecure: bool) -> imaplib.IMAP4_SSL: context = ssl.create_default_context() if insecure: context.check_hostname = False context.verify_mode = ssl.CERT_NONE imap = imaplib.IMAP4_SSL(host, port, ssl_context=context, timeout=30) imap._encoding = "utf-8" return imap def run(args: argparse.Namespace) -> int: accounts: dict[str, str] = json.loads( Path(args.accounts_file).read_text(encoding="utf-8") ) maildir_root = Path(args.maildir_root) exit_code = 0 for address, password in accounts.items(): account_dir = maildir_root / address if not account_dir.is_dir(): print(f"[SKIP] no maildir for {address} under {maildir_root}") continue print(f"[ACCOUNT] {address}") stats = AccountStats() imap = connect(args.imap_host, args.imap_port, args.imap_insecure) try: imap.login(address, password) migrate_account(imap, account_dir, args.dest_separator, stats) finally: with contextlib.suppress(Exception): imap.logout() for mailbox, f in sorted(stats.folders.items()): print( f" {mailbox}: appended={f.appended} " f"skipped={f.skipped} failed={f.failed}" ) if stats.failed: exit_code = 1 return exit_code def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--maildir-root", required=True, help="Mailu dovecot mail volume root (contains one / tree per account)", ) parser.add_argument("--imap-host", required=True, help="Stalwart IMAP host") parser.add_argument("--imap-port", type=int, default=993) parser.add_argument( "--imap-insecure", action="store_true", help="Skip TLS verification (fixture/bootstrap certificates)", ) parser.add_argument( "--accounts-file", required=True, help='JSON file {"": ""} of provisioned Stalwart accounts', ) parser.add_argument( "--dest-separator", default="/", help="IMAP hierarchy separator on the destination (Stalwart: '/')", ) return parser.parse_args(argv) if __name__ == "__main__": start = time.time() rc = run(parse_args()) print(f"[DONE] rc={rc} in {time.time() - start:.1f}s") raise SystemExit(rc)