#!/usr/bin/env python3 """export_mailbox.py - runs ON the wahlkampf host. PARALLEL + SLICED IMAP export. For one mailbox id: read its address, set a known password via the mail-mgmt API, then IMAP-export every message using N parallel IMAP connections, writing byte-budgeted .tar.gz SLICES onto the SHARED NFS path. Usage: python3 export_mailbox.py [threads] [slice_mb] [pw] Emits: /_.sliceNN.tar.gz and a line per slice in index.tsv """ import concurrent.futures as cf import imaplib import io import json import os import re import sys import tarfile import threading import time import urllib.request KEY = "ZEA4R2sEkegRafwLsLPyqnpVAz2EwNnOUb9HscWqZV6TaXejfScxXPOG5ubxzRIhfBfMTBhMWcE7HD95frJt7OXO1b4s8G59z8MqeuCxMP5CiJJpTkdCtsnrZmzsBsxj" BASE = "https://mail-management.afd.de/api/v1/users" IMAP_HOST = "10.12.51.240" OUTDIR = "/var/www/vhosts/wahlkampf.afd.de/.maildump" HDR = {"Authorization": KEY, "User-Agent": "curl/8.5.0", "Accept": "application/json", "Content-Type": "application/json"} _tls = threading.local() _lock = threading.Lock() _counter = {"n": 0} def api(method, path, body=None): data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(BASE + path, data=data, headers=HDR, method=method) try: r = urllib.request.urlopen(req, timeout=40) return r.status, json.loads(r.read().decode()) except urllib.error.HTTPError as e: return e.code, e.read(500).decode("utf-8", "replace") except Exception as e: # noqa: BLE001 return None, str(e) def sanitize(s): return re.sub(r"[^A-Za-z0-9._@-]", "_", s) def q(folder): return '"' + folder.replace("\\", "\\\\").replace('"', '\\"') + '"' def _conn(addr, pw): c = getattr(_tls, "c", None) if c is None: c = imaplib.IMAP4_SSL(IMAP_HOST, 993, timeout=90) c.login(addr, pw) _tls.c = c return c def fetch_batch(addr, pw, folder, ids): """Fetch a list of message ids in ONE IMAP command. Returns (folder, [(mid, raw), ...]).""" for _ in range(4): try: c = _conn(addr, pw) if getattr(_tls, "folder", None) != folder: typ, _ = c.select(q(folder)) if typ != "OK": raise RuntimeError("select failed") _tls.folder = folder typ, data = c.fetch(",".join(ids), "(RFC822)") if typ != "OK": raise RuntimeError("fetch failed") out = [] for item in data: if isinstance(item, tuple) and len(item) == 2 and item[1]: m = re.match(rb"(\d+) ", item[0] or b"") out.append((m.group(1).decode() if m else "?", item[1])) return folder, out except Exception: # noqa: BLE001 _tls.c = None _tls.folder = None time.sleep(1) return folder, [] def list_work(addr, pw): """Enumerate (folder, msgid) with retries on the LIST connection.""" for attempt in range(4): work = [] try: M = imaplib.IMAP4_SSL(IMAP_HOST, 993) M.login(addr, pw) typ, folders = M.list() for f in folders: m = re.match(rb'\([^)]*\)\s+"[^"]*"\s+(?P.*)$', f) if not m: continue name = m.group("name").strip() if name.startswith(b'"') and name.endswith(b'"'): name = name[1:-1] folder = name.decode("utf-8", "replace") try: typ, _ = M.select(q(folder)) if typ != "OK": continue typ, data = M.search(None, "ALL") if typ != "OK": continue for i in data[0].split(): work.append((folder, i)) except Exception: # noqa: BLE001 continue M.logout() return work except Exception: # noqa: BLE001 time.sleep(5) return [] def main(): uid = int(sys.argv[1]) threads = int(sys.argv[2]) if len(sys.argv) > 2 else 8 slice_mb = int(sys.argv[3]) if len(sys.argv) > 3 else 20000 pw = sys.argv[4] if len(sys.argv) > 4 else "Rekey-AfD-2026!x9Qz" slice_bytes = slice_mb * 1024 * 1024 os.makedirs(OUTDIR, exist_ok=True) st, u = api("GET", f"/{uid}") if st != 200 or not isinstance(u, dict): print(f"ID {uid}: GET failed {st} {u}") return 1 addr = f"{u.get('name')}@{(u.get('domain') or {}).get('name')}" st, _ = api("POST", f"/{uid}/password", {"password": pw, "password_confirm": pw}) print(f"ID {uid} {addr}: pw set -> {st} (msgs={u.get('mailbox_usage_messages')})") base = f"{uid}_{sanitize(addr)}" for x in os.listdir(OUTDIR): if x.startswith(base + ".slice") and x.endswith(".part"): try: os.remove(os.path.join(OUTDIR, x)) except OSError: pass donef = os.path.join(OUTDIR, base + ".done") done = set() if os.path.exists(donef): with open(donef, encoding="utf-8", errors="replace") as f: done = {ln.strip() for ln in f if ln.strip()} work = list_work(addr, pw) alltotal = len(work) expected = u.get("mailbox_usage_messages") or 0 if alltotal == 0 and expected > 0 and len(done) == 0: print(f"ERROR {addr}: list_work returned 0 but API says {expected} msgs; " f"aborting without a completion marker so the pipeline retries", flush=True) return 2 work = [(fo, mi) for (fo, mi) in work if f"{fo}|{mi.decode()}" not in done] total = len(work) idxf = os.path.join(OUTDIR, base + ".nextidx") idx = 0 if os.path.exists(idxf): try: idx = int(open(idxf).read().strip() or "0") except ValueError: idx = 0 print(f"{addr}: {total}/{alltotal} remaining ({len(done)} done), " f"{threads} threads, slice={slice_mb}MB, start slice{idx:03d}") donelog = open(donef, "a", encoding="utf-8") cur_bytes = 0 cur = None cur_path = None cur_keys = [] def open_slice(): nonlocal cur, cur_bytes, cur_path, idx, cur_keys cur_path = os.path.join(OUTDIR, f"{base}.slice{idx:03d}.tar.gz") cur = tarfile.open(cur_path + ".part", "w:gz", compresslevel=1) cur_bytes = 0 idx += 1 with open(idxf, "w") as f: f.write(str(idx)) cur_keys = [] open_slice() BATCH = 100 byfolder = {} for fo, mi in work: byfolder.setdefault(fo, []).append(mi.decode()) batches = [] for fo, ids in byfolder.items(): for i in range(0, len(ids), BATCH): batches.append((fo, ids[i:i + BATCH])) print(f"{addr}: {len(batches)} batches of {BATCH}", flush=True) with cf.ThreadPoolExecutor(max_workers=threads) as ex: futs = [ex.submit(fetch_batch, addr, pw, fo, ids) for fo, ids in batches] for fut in cf.as_completed(futs): folder, msgs = fut.result() for mid, raw in msgs: if raw is None: continue info = tarfile.TarInfo(f"{sanitize(folder)}/{mid}.eml") info.size = len(raw) with _lock: cur.addfile(info, io.BytesIO(raw)) cur_bytes += len(raw) _counter["n"] += 1 cur_keys.append(f"{folder}|{mid}") if cur_bytes >= slice_bytes: cur.close() os.replace(cur_path + ".part", cur_path) sz = os.path.getsize(cur_path) with open(os.path.join(OUTDIR, "index.tsv"), "a") as f: f.write(f"{uid}\t{addr}\t{os.path.basename(cur_path)}\t{sz}\t" f"{time.strftime('%Y-%m-%dT%H:%M:%SZ')}\n") donelog.write("\n".join(cur_keys) + "\n") donelog.flush() print(f" slice {cur_path} {sz} bytes", flush=True) open_slice() if msgs: print(f" {addr}: {_counter['n']}/{total}...", flush=True) cur.close() os.replace(cur_path + ".part", cur_path) sz = os.path.getsize(cur_path) with open(os.path.join(OUTDIR, "index.tsv"), "a") as f: f.write(f"{uid}\t{addr}\t{os.path.basename(cur_path)}\t{sz}\t" f"{time.strftime('%Y-%m-%dT%H:%M:%SZ')}\n") donelog.write("\n".join(cur_keys) + "\n") print(f"DONE {addr}: {_counter['n']} msgs, {idx} slices -> {base}.slice*.tar.gz") donelog.flush() donelog.close() return 0 if __name__ == "__main__": sys.exit(main())