#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Mailcow Assisted Installer V3 This tool automates the Linux and mailcow installation path while keeping DNS, rDNS/PTR, DKIM and deliverability steps explicit and auditable. Core goals: - Preflight checks before installation. - DNS plan generation. - Optional Cloudflare DNS apply mode. - Docker + mailcow bootstrap for Debian/Ubuntu. - Deliverability doctor after installation. - Django SMTP snippet generation. - JSON and HTML customer reports. The script intentionally does not set PTR/rDNS automatically. PTR is managed by the server/IP provider, not by the domain DNS zone. Examples: python3 mailcow_assisted_installer_v2.py dns-plan --domain example.com --hostname mail.example.com --public-ip 1.2.3.4 sudo python3 mailcow_assisted_installer_v2.py preflight --domain example.com --hostname mail.example.com sudo python3 mailcow_assisted_installer_v2.py all --domain example.com --hostname mail.example.com --yes python3 mailcow_assisted_installer_v2.py django-snippet --domain example.com --smtp-user noreply@example.com CLOUDFLARE_API_TOKEN=xxx python3 mailcow_assisted_installer_v2.py dns-apply-cloudflare --domain example.com --hostname mail.example.com --public-ip 1.2.3.4 --cloudflare-zone-id ZONEID --yes """ from __future__ import annotations import argparse import csv import tarfile import zipfile import datetime as dt import getpass import html import ipaddress import json import os import platform import re import shutil import socket import ssl import subprocess import sys import textwrap import time import urllib.error import urllib.parse import urllib.request from dataclasses import dataclass, field from pathlib import Path from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple APP_NAME = "Mailcow Assisted Installer V3" APP_VERSION = "2.0.0" DEFAULT_INSTALL_DIR = Path("/opt/mailcow-dockerized") DEFAULT_TIMEZONE = "Europe/Madrid" DEFAULT_REPORT_DIR = Path("/root/mailcow-installer-reports") MIN_RAM_GIB = 6 MIN_SWAP_GIB = 1 MIN_DISK_GIB = 20 SUPPORTED_ARCHES = {"x86_64", "amd64", "aarch64", "arm64"} REQUIRED_MAILCOW_PORTS = [25, 80, 110, 143, 443, 465, 587, 993, 995, 4190] SYSTEM_PACKAGES_DEBIAN = [ "git", "openssl", "curl", "gawk", "coreutils", "grep", "jq", "ca-certificates", "gnupg", "lsb-release", "dnsutils", "iproute2", "ufw", ] DOCKER_PACKAGES = [ "docker-ce", "docker-ce-cli", "containerd.io", "docker-buildx-plugin", "docker-compose-plugin", ] CONFLICTING_DOCKER_PACKAGES = [ "docker.io", "docker-compose", "docker-compose-v2", "docker-doc", "podman-docker", "containerd", "runc", ] CLOUDFLARE_API_BASE = "https://api.cloudflare.com/client/v4" COMMON_DNSBLS = [ "zen.spamhaus.org", "bl.spamcop.net", "b.barracudacentral.org", "dnsbl.sorbs.net", ] class InstallerError(RuntimeError): pass @dataclass class CommandResult: command: str rc: int stdout: str = "" stderr: str = "" @dataclass class CheckResult: category: str name: str status: str message: str details: Dict[str, Any] = field(default_factory=dict) @property def ok(self) -> bool: return self.status == "OK" @property def is_fail(self) -> bool: return self.status == "FAIL" @dataclass class DnsRecord: name: str type: str content: str ttl: int = 1 priority: Optional[int] = None proxied: bool = False required: bool = True purpose: str = "" def full_name(self, domain: str) -> str: if self.name in {"@", domain}: return domain if self.name.endswith("." + domain): return self.name return f"{self.name}.{domain}" def cloudflare_payload(self, domain: str) -> Dict[str, Any]: payload: Dict[str, Any] = { "type": self.type, "name": self.full_name(domain), "content": self.content, "ttl": self.ttl, "proxied": self.proxied, "comment": "Managed by Mailcow Assisted Installer V3", } if self.type == "MX" and self.priority is not None: payload["priority"] = self.priority return payload @dataclass class InstallContext: action: str domain: Optional[str] hostname: Optional[str] public_ip: Optional[str] ipv6: Optional[str] timezone: str install_dir: Path report_dir: Path dry_run: bool yes: bool quiet: bool low_memory: bool create_swap_gb: int skip_docker_install: bool remove_conflicting_docker: bool update_existing_repo: bool pull_images: bool start_mailcow: bool configure_ufw: bool check_outbound_25: bool cloudflare_zone_id: Optional[str] cloudflare_token: Optional[str] cloudflare_overwrite: bool smtp_user: Optional[str] smtp_password_env: str django_settings_module: Optional[str] mailcow_api_url: Optional[str] mailcow_api_key: Optional[str] mailcow_create_domain: bool mailcow_create_mailbox: Optional[str] mailcow_mailbox_password_env: str client_name: Optional[str] target_volume: int pack_name: Optional[str] service_tier: str class Console: def __init__(self, quiet: bool = False) -> None: self.quiet = quiet def line(self, text: str = "") -> None: if not self.quiet: print(text) def title(self, text: str) -> None: if self.quiet: return print() print(text) print("=" * len(text)) def section(self, text: str) -> None: if self.quiet: return print() print(text) print("-" * len(text)) def info(self, text: str) -> None: self.line(f"[INFO] {text}") def ok(self, text: str) -> None: self.line(f"[OK] {text}") def warn(self, text: str) -> None: self.line(f"[WARN] {text}") def fail(self, text: str) -> None: self.line(f"[FAIL] {text}") def progress(self, done: int, total: int, text: str) -> None: if self.quiet: return pct = int((done / max(total, 1)) * 100) width = 24 filled = int(width * pct / 100) bar = "#" * filled + "." * (width - filled) print(f"[{bar}] {pct:3d}% {text}") class Runner: def __init__(self, console: Console, dry_run: bool = False) -> None: self.console = console self.dry_run = dry_run self.history: List[CommandResult] = [] def run( self, cmd: Sequence[str] | str, *, cwd: Optional[Path] = None, check: bool = True, capture: bool = True, shell: bool = False, input_text: Optional[str] = None, env: Optional[Dict[str, str]] = None, ) -> CommandResult: printable = cmd if isinstance(cmd, str) else " ".join(cmd) if cwd: printable = f"(cd {cwd} && {printable})" if self.dry_run: self.console.info(f"DRY-RUN: {printable}") result = CommandResult(str(printable), 0, "", "") self.history.append(result) return result self.console.info(f"RUN: {printable}") completed = subprocess.run( cmd, cwd=str(cwd) if cwd else None, shell=shell, text=True, input=input_text, capture_output=capture, check=False, env=env, ) result = CommandResult( str(printable), completed.returncode, completed.stdout or "", completed.stderr or "", ) self.history.append(result) if check and completed.returncode != 0: raise InstallerError( f"Command failed with rc={completed.returncode}: {printable}\n" f"STDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}" ) return result class ReportStore: def __init__(self, ctx: InstallContext, console: Console) -> None: self.ctx = ctx self.console = console self.checks: List[CheckResult] = [] self.artifacts: Dict[str, str] = {} self.started_at = dt.datetime.now(dt.timezone.utc) def add(self, result: CheckResult) -> CheckResult: self.checks.append(result) if result.status == "OK": self.console.ok(f"{result.category} / {result.name}: {result.message}") elif result.status == "WARN": self.console.warn(f"{result.category} / {result.name}: {result.message}") else: self.console.fail(f"{result.category} / {result.name}: {result.message}") return result def extend(self, items: Iterable[CheckResult]) -> None: for item in items: self.add(item) def add_artifact(self, key: str, path: Path) -> None: self.artifacts[key] = str(path) def summary(self) -> Dict[str, int]: counts = {"OK": 0, "WARN": 0, "FAIL": 0} for item in self.checks: counts[item.status] = counts.get(item.status, 0) + 1 return counts def utc_stamp() -> str: return dt.datetime.now(dt.timezone.utc).strftime("%Y%m%d_%H%M%S_utc") def read_text(path: Path) -> str: try: return path.read_text(encoding="utf-8", errors="replace") except FileNotFoundError: return "" def write_text(path: Path, content: str, mode: int = 0o644) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(content, encoding="utf-8") try: os.chmod(path, mode) except PermissionError: pass def command_exists(name: str) -> bool: return shutil.which(name) is not None def normalize_domain(domain: Optional[str]) -> Optional[str]: if not domain: return None domain = domain.strip().lower().rstrip(".") if not re.match(r"^[a-z0-9][a-z0-9.-]*\.[a-z]{2,}$", domain): raise InstallerError(f"Invalid domain: {domain}") return domain def normalize_hostname(hostname: Optional[str], domain: Optional[str]) -> Optional[str]: if hostname: value = hostname.strip().lower().rstrip(".") elif domain: value = f"mail.{domain}" else: return None if not re.match(r"^[a-z0-9][a-z0-9.-]*\.[a-z]{2,}$", value): raise InstallerError(f"Invalid hostname: {value}") return value def normalize_ip(value: Optional[str]) -> Optional[str]: if not value: return None value = value.strip() try: ipaddress.ip_address(value) except ValueError as exc: raise InstallerError(f"Invalid IP address: {value}") from exc return value def require_root(action: str) -> None: if action in {"install", "all", "postcheck", "configure-firewall", "mailcow-bootstrap", "backup-config"} and os.geteuid() != 0: raise InstallerError("This action must be run as root.") def confirm(ctx: InstallContext, console: Console, question: str) -> None: if ctx.yes or ctx.dry_run: return answer = input(f"{question} [y/N] ").strip().lower() if answer not in {"y", "yes"}: raise InstallerError("Cancelled by user.") def load_os_release() -> Dict[str, str]: data: Dict[str, str] = {} for line in read_text(Path("/etc/os-release")).splitlines(): if "=" not in line: continue key, value = line.split("=", 1) data[key] = value.strip().strip('"') return data def distro() -> Tuple[str, str, str]: data = load_os_release() return ( data.get("ID", "").lower(), data.get("VERSION_ID", ""), data.get("VERSION_CODENAME") or data.get("UBUNTU_CODENAME") or "", ) def is_supported_debian_ubuntu() -> bool: os_id, version, _codename = distro() if os_id == "debian": try: return int(version.split(".")[0]) in {11, 12, 13} except Exception: return False if os_id == "ubuntu": try: major, minor = (version.split(".") + ["0"])[:2] return int(major) > 22 or (int(major) == 22 and int(minor) >= 4) except Exception: return False return False def memory_swap_gib() -> Tuple[float, float]: mem_kb = 0 swap_kb = 0 for line in read_text(Path("/proc/meminfo")).splitlines(): if line.startswith("MemTotal:"): mem_kb = int(line.split()[1]) elif line.startswith("SwapTotal:"): swap_kb = int(line.split()[1]) return mem_kb / 1024 / 1024, swap_kb / 1024 / 1024 def free_disk_gib(path: Path) -> float: probe = path if path.exists() else path.parent usage = shutil.disk_usage(str(probe)) return usage.free / 1024 / 1024 / 1024 def detect_public_ipv4(console: Console) -> Optional[str]: for endpoint in ["https://ip4.mailcow.email", "https://api.ipify.org"]: try: console.info(f"Detecting public IPv4 from {endpoint}") with urllib.request.urlopen(endpoint, timeout=7) as response: value = response.read().decode("ascii", errors="ignore").strip() ipaddress.IPv4Address(value) return value except Exception: continue return None def parse_version_major(value: str) -> Optional[int]: match = re.search(r"(\d+)", value) if not match: return None return int(match.group(1)) def docker_version() -> Tuple[bool, str]: if not command_exists("docker"): return False, "docker command not found" result = subprocess.run( ["docker", "version", "--format", "{{.Server.Version}}"], text=True, capture_output=True, check=False, ) version = (result.stdout or "").strip() if result.returncode != 0 or not version: return False, "docker daemon not reachable" major = parse_version_major(version) return bool(major and major >= 24), version def compose_version() -> Tuple[bool, str]: if not command_exists("docker"): return False, "docker command not found" result = subprocess.run( ["docker", "compose", "version", "--short"], text=True, capture_output=True, check=False, ) version = (result.stdout or "").strip() if result.returncode != 0 or not version: return False, "docker compose plugin not reachable" major = parse_version_major(version) return bool(major and major >= 2), version def listening_ports(runner: Runner) -> Dict[int, str]: ports: Dict[int, str] = {} if not command_exists("ss"): return ports result = runner.run(["ss", "-tlpn"], check=False, capture=True) for line in result.stdout.splitlines(): port_match = re.search(r":(\d+)\s+", line) if not port_match: continue port = int(port_match.group(1)) users = "" user_match = re.search(r"users:\(\((.*)\)\)", line) if user_match: users = user_match.group(1) ports[port] = users or line.strip() return ports def systemd_virt(runner: Runner) -> str: if not command_exists("systemd-detect-virt"): return "unknown" result = runner.run(["systemd-detect-virt"], check=False, capture=True) if result.rc == 0: return result.stdout.strip() or "unknown" return "none" def resolve_ipv4(name: str) -> List[str]: try: infos = socket.getaddrinfo(name, None, socket.AF_INET, socket.SOCK_STREAM) except socket.gaierror: return [] return sorted({item[4][0] for item in infos}) def dig(runner: Runner, name: str, record_type: str) -> List[str]: if not command_exists("dig"): return [] result = runner.run(["dig", "+short", record_type, name], check=False, capture=True) return [line.strip() for line in result.stdout.splitlines() if line.strip()] def dig_reverse(runner: Runner, ip: str) -> List[str]: if not command_exists("dig"): return [] result = runner.run(["dig", "+short", "-x", ip], check=False, capture=True) return [line.strip().rstrip(".") for line in result.stdout.splitlines() if line.strip()] def tcp_connect(host: str, port: int, timeout: float = 6.0) -> bool: try: with socket.create_connection((host, port), timeout=timeout): return True except OSError: return False def check_tls(host: str, port: int = 443, timeout: float = 6.0) -> Tuple[bool, str]: try: context = ssl.create_default_context() with socket.create_connection((host, port), timeout=timeout) as sock: with context.wrap_socket(sock, server_hostname=host) as ssock: cert = ssock.getpeercert() subject = cert.get("subject", []) return True, str(subject) except Exception as exc: return False, str(exc) def build_dns_records(domain: str, hostname: str, public_ip: Optional[str], ipv6: Optional[str] = None) -> List[DnsRecord]: host_label = hostname if hostname.endswith("." + domain): host_label = hostname[: -(len(domain) + 1)] if not host_label: host_label = "@" ip = public_ip or "" records = [ DnsRecord(host_label, "A", ip, required=True, purpose="Mail host IPv4 address."), DnsRecord("@", "MX", hostname, priority=10, required=True, purpose="Inbound mail routing."), DnsRecord("@", "TXT", "v=spf1 mx a -all", required=True, purpose="SPF authorization."), DnsRecord( "_dmarc", "TXT", f"v=DMARC1; p=quarantine; rua=mailto:dmarc@{domain}; ruf=mailto:dmarc@{domain}; fo=1", required=True, purpose="DMARC policy and reporting.", ), DnsRecord("autodiscover", "CNAME", hostname, required=False, purpose="Outlook auto-configuration."), DnsRecord("autoconfig", "CNAME", hostname, required=False, purpose="Thunderbird auto-configuration."), DnsRecord( "dkim._domainkey", "TXT", "v=DKIM1; k=rsa; p=", required=True, purpose="DKIM key generated by mailcow after domain creation.", ), DnsRecord("_submission._tcp", "SRV", f"0 1 587 {hostname}", required=False, purpose="SMTP submission discovery."), DnsRecord("_imaps._tcp", "SRV", f"0 1 993 {hostname}", required=False, purpose="IMAPS discovery."), DnsRecord("_smtps._tcp", "SRV", f"0 1 465 {hostname}", required=False, purpose="SMTPS discovery."), ] if ipv6: records.insert(1, DnsRecord(host_label, "AAAA", ipv6, required=False, purpose="Mail host IPv6 address.")) return records def render_dns_plan(ctx: InstallContext) -> str: if not ctx.domain or not ctx.hostname: raise InstallerError("domain and hostname are required.") records = build_dns_records(ctx.domain, ctx.hostname, ctx.public_ip, ctx.ipv6) lines = [ f"# DNS plan for {ctx.domain}", "", f"Client: `{ctx.client_name or '-'}`", f"Mail host: `{ctx.hostname}`", f"Public IPv4: `{ctx.public_ip or ''}`", f"Public IPv6: `{ctx.ipv6 or '-'}`", "", "## Records to create", "", "| Required | Name | Type | Priority | TTL | Value | Purpose |", "|---|---|---|---:|---:|---|---|", ] for record in records: lines.append( f"| {'yes' if record.required else 'recommended'} | `{record.full_name(ctx.domain)}` | `{record.type}` | " f"`{record.priority or ''}` | `{record.ttl}` | `{record.content}` | {record.purpose} |" ) lines.extend( [ "", "## PTR / reverse DNS", "", f"Set PTR/rDNS for `{ctx.public_ip or ''}` to `{ctx.hostname}` in the server provider panel.", "This cannot normally be done in the domain DNS zone.", "", "## DKIM workflow", "", "1. Install and start mailcow.", f"2. Add the domain `{ctx.domain}` in the mailcow UI.", "3. Generate DKIM for the domain.", "4. Replace the placeholder DKIM TXT record in DNS.", "5. Run this installer again with `doctor` to validate SPF, DKIM, DMARC and PTR.", "", "## Warm-up rule", "", "Do not send 100k messages immediately from a new IP. Increase volume progressively and monitor bounces, complaints, queue size and blacklist status.", ] ) return "\n".join(lines) + "\n" def save_dns_plan(ctx: InstallContext, report: ReportStore) -> Path: content = render_dns_plan(ctx) path = ctx.report_dir / f"dns_plan_{ctx.domain}_{utc_stamp()}.md" write_text(path, content) report.add_artifact("dns_plan", path) return path def render_django_snippet(ctx: InstallContext) -> str: if not ctx.domain: raise InstallerError("domain is required for django-snippet.") host = ctx.hostname or f"mail.{ctx.domain}" smtp_user = ctx.smtp_user or f"noreply@{ctx.domain}" password_env = ctx.smtp_password_env default_from = smtp_user return textwrap.dedent( f""" # Django SMTP settings for mailcow # Store secrets in environment variables. Do not hardcode SMTP passwords. import os EMAIL_BACKEND = "django.core.mail.backends.smtp.EmailBackend" EMAIL_HOST = os.getenv("EMAIL_HOST", "{host}") EMAIL_PORT = int(os.getenv("EMAIL_PORT", "587")) EMAIL_HOST_USER = os.getenv("EMAIL_HOST_USER", "{smtp_user}") EMAIL_HOST_PASSWORD = os.getenv("{password_env}", "") EMAIL_USE_TLS = os.getenv("EMAIL_USE_TLS", "1") == "1" EMAIL_USE_SSL = False DEFAULT_FROM_EMAIL = os.getenv("DEFAULT_FROM_EMAIL", "{default_from}") SERVER_EMAIL = os.getenv("SERVER_EMAIL", DEFAULT_FROM_EMAIL) # Recommended environment values: # EMAIL_HOST={host} # EMAIL_PORT=587 # EMAIL_HOST_USER={smtp_user} # {password_env}= # DEFAULT_FROM_EMAIL={default_from} """ ).strip() + "\n" def save_django_snippet(ctx: InstallContext, report: ReportStore) -> Path: content = render_django_snippet(ctx) path = ctx.report_dir / f"django_smtp_settings_{ctx.domain}_{utc_stamp()}.py" write_text(path, content) report.add_artifact("django_snippet", path) return path def check_preflight(ctx: InstallContext, runner: Runner) -> List[CheckResult]: results: List[CheckResult] = [] os_id, version, codename = distro() supported = is_supported_debian_ubuntu() results.append(CheckResult("system", "os", "OK" if supported else "FAIL", f"{os_id} {version} {codename}".strip())) arch = platform.machine().lower() results.append(CheckResult("system", "architecture", "OK" if arch in SUPPORTED_ARCHES else "FAIL", arch)) virt = systemd_virt(runner) if virt in {"openvz", "lxc", "container", "docker", "podman"}: results.append(CheckResult("system", "virtualization", "FAIL", f"Unsupported or risky virtualization detected: {virt}")) elif virt == "unknown": results.append(CheckResult("system", "virtualization", "WARN", "Could not detect virtualization type")) else: results.append(CheckResult("system", "virtualization", "OK", virt)) mem_gib, swap_gib = memory_swap_gib() results.append( CheckResult( "system", "memory", "OK" if mem_gib >= MIN_RAM_GIB else "WARN" if ctx.low_memory else "FAIL", f"{mem_gib:.2f} GiB RAM detected; target is {MIN_RAM_GIB} GiB", {"ram_gib": round(mem_gib, 2), "low_memory": ctx.low_memory}, ) ) swap_status = "OK" if swap_gib >= MIN_SWAP_GIB or ctx.create_swap_gb > 0 else "WARN" results.append( CheckResult( "system", "swap", swap_status, f"{swap_gib:.2f} GiB swap detected; auto create target is {ctx.create_swap_gb} GiB", {"swap_gib": round(swap_gib, 2)}, ) ) disk_gib = free_disk_gib(ctx.install_dir) results.append( CheckResult( "system", "disk", "OK" if disk_gib >= MIN_DISK_GIB else "FAIL", f"{disk_gib:.2f} GiB free near {ctx.install_dir}; target is {MIN_DISK_GIB} GiB", ) ) ports = listening_ports(runner) busy = {port: ports[port] for port in REQUIRED_MAILCOW_PORTS if port in ports} results.append( CheckResult( "network", "local ports", "OK" if not busy else "FAIL", "Required ports are free" if not busy else f"Busy ports: {busy}", {"busy": busy}, ) ) docker_ok, docker_msg = docker_version() compose_ok, compose_msg = compose_version() if ctx.skip_docker_install: results.append(CheckResult("docker", "engine", "OK" if docker_ok else "FAIL", docker_msg)) results.append(CheckResult("docker", "compose", "OK" if compose_ok else "FAIL", compose_msg)) else: results.append(CheckResult("docker", "engine", "OK" if docker_ok else "WARN", docker_msg if docker_ok else "Docker will be installed or updated")) results.append(CheckResult("docker", "compose", "OK" if compose_ok else "WARN", compose_msg if compose_ok else "Docker Compose plugin will be installed or updated")) if ctx.hostname and ctx.public_ip: resolved = resolve_ipv4(ctx.hostname) status = "OK" if ctx.public_ip in resolved else "WARN" results.append(CheckResult("dns", "A record", status, f"{ctx.hostname} resolves to {resolved}; expected {ctx.public_ip}")) if ctx.check_outbound_25: ok = tcp_connect("gmail-smtp-in.l.google.com", 25, timeout=7) results.append(CheckResult("network", "outbound tcp 25", "OK" if ok else "WARN", "Outbound TCP/25 reachable" if ok else "Outbound TCP/25 seems blocked")) return results def install_system_packages(runner: Runner) -> None: runner.run(["apt-get", "update"]) runner.run(["apt-get", "install", "-y", *SYSTEM_PACKAGES_DEBIAN]) def installed_packages(packages: Sequence[str], runner: Runner) -> List[str]: found: List[str] = [] for package in packages: result = runner.run(["dpkg-query", "-W", "-f=${Status}", package], check=False, capture=True) if "install ok installed" in result.stdout: found.append(package) return found def install_docker(ctx: InstallContext, runner: Runner, console: Console) -> None: if ctx.skip_docker_install: console.info("Docker installation skipped by flag.") return docker_ok, _docker_msg = docker_version() compose_ok, _compose_msg = compose_version() if docker_ok and compose_ok: console.ok("Docker and Docker Compose are already ready.") return os_id, _version, codename = distro() if os_id not in {"debian", "ubuntu"} or not codename: raise InstallerError("Automatic Docker install currently supports Debian/Ubuntu only.") conflicts = installed_packages(CONFLICTING_DOCKER_PACKAGES, runner) if conflicts: if not ctx.remove_conflicting_docker: raise InstallerError(f"Conflicting Docker packages found: {conflicts}. Add --remove-conflicting-docker on a fresh host.") runner.run(["apt-get", "remove", "-y", *conflicts], check=False) runner.run(["apt-get", "update"]) runner.run(["apt-get", "install", "-y", "ca-certificates", "curl"]) runner.run(["install", "-m", "0755", "-d", "/etc/apt/keyrings"]) runner.run(["curl", "-fsSL", f"https://download.docker.com/linux/{os_id}/gpg", "-o", "/etc/apt/keyrings/docker.asc"]) runner.run(["chmod", "a+r", "/etc/apt/keyrings/docker.asc"]) arch = runner.run(["dpkg", "--print-architecture"], capture=True).stdout.strip() source = textwrap.dedent( f"""\ Types: deb URIs: https://download.docker.com/linux/{os_id} Suites: {codename} Components: stable Architectures: {arch} Signed-By: /etc/apt/keyrings/docker.asc """ ) if ctx.dry_run: console.info("DRY-RUN: write /etc/apt/sources.list.d/docker.sources") else: write_text(Path("/etc/apt/sources.list.d/docker.sources"), source) runner.run(["apt-get", "update"]) runner.run(["apt-get", "install", "-y", *DOCKER_PACKAGES]) runner.run(["systemctl", "enable", "--now", "docker"], check=False) def ensure_swap(ctx: InstallContext, runner: Runner, console: Console) -> None: _mem, swap = memory_swap_gib() if swap >= MIN_SWAP_GIB: console.ok(f"Swap already present: {swap:.2f} GiB") return if ctx.create_swap_gb <= 0: console.warn("No swap detected and swap creation disabled.") return swapfile = Path("/swapfile_mailcow") if swapfile.exists(): console.warn(f"Swap file already exists: {swapfile}") return runner.run(["fallocate", "-l", f"{ctx.create_swap_gb}G", str(swapfile)]) runner.run(["chmod", "600", str(swapfile)]) runner.run(["mkswap", str(swapfile)]) runner.run(["swapon", str(swapfile)]) entry = f"{swapfile} none swap sw 0 0" current = read_text(Path("/etc/fstab")) if entry not in current: if ctx.dry_run: console.info(f"DRY-RUN: append swap entry to /etc/fstab: {entry}") else: with Path("/etc/fstab").open("a", encoding="utf-8") as handle: handle.write("\n" + entry + "\n") def configure_ufw(ctx: InstallContext, runner: Runner, console: Console) -> None: if not ctx.configure_ufw: return if not command_exists("ufw"): console.warn("ufw not available; firewall configuration skipped.") return for port in REQUIRED_MAILCOW_PORTS: runner.run(["ufw", "allow", str(port)], check=False) runner.run(["ufw", "allow", "22"], check=False) runner.run(["ufw", "--force", "enable"], check=False) runner.run(["ufw", "status", "verbose"], check=False) def clone_or_update_mailcow(ctx: InstallContext, runner: Runner, console: Console) -> None: repo = ctx.install_dir if repo.exists() and (repo / ".git").exists(): console.ok(f"Existing mailcow repository found at {repo}") if ctx.update_existing_repo: runner.run(["git", "fetch", "--all", "--tags"], cwd=repo) runner.run(["git", "pull", "--ff-only"], cwd=repo) return if repo.exists(): raise InstallerError(f"Install directory exists but is not a git repository: {repo}") runner.run(["mkdir", "-p", str(repo.parent)]) runner.run(["git", "clone", "https://github.com/mailcow/mailcow-dockerized", str(repo)]) def update_key_values(path: Path, values: Dict[str, str], ctx: InstallContext, console: Console) -> None: content = read_text(path) if not content: raise InstallerError(f"Cannot read {path}") lines: List[str] = [] seen: set[str] = set() for line in content.splitlines(): if not line or line.lstrip().startswith("#") or "=" not in line: lines.append(line) continue key, _value = line.split("=", 1) key = key.strip() if key in values: lines.append(f"{key}={values[key]}") seen.add(key) else: lines.append(line) for key, value in values.items(): if key not in seen: lines.append(f"{key}={value}") if ctx.dry_run: console.info(f"DRY-RUN: update {path} keys {sorted(values.keys())}") return backup = path.with_name(path.name + f".backup_{utc_stamp()}") shutil.copy2(path, backup) path.write_text("\n".join(lines) + "\n", encoding="utf-8") console.ok(f"Updated {path}; backup at {backup}") def generate_mailcow_config(ctx: InstallContext, runner: Runner, console: Console) -> None: if not ctx.hostname: raise InstallerError("hostname is required for mailcow configuration.") conf = ctx.install_dir / "mailcow.conf" if not conf.exists(): input_text = f"{ctx.hostname}\n{ctx.timezone}\n" runner.run(["./generate_config.sh"], cwd=ctx.install_dir, input_text=input_text) values = {"MAILCOW_HOSTNAME": ctx.hostname, "TZ": ctx.timezone} if ctx.low_memory: values["SKIP_CLAMD"] = "y" values["SKIP_FTS"] = "y" update_key_values(conf, values, ctx, console) def start_mailcow(ctx: InstallContext, runner: Runner) -> None: if ctx.pull_images: runner.run(["docker", "compose", "pull"], cwd=ctx.install_dir) if ctx.start_mailcow: runner.run(["docker", "compose", "up", "-d"], cwd=ctx.install_dir) def run_install(ctx: InstallContext, runner: Runner, console: Console, report: ReportStore) -> None: require_root(ctx.action) console.title("Installation") steps = [ "system packages", "swap", "docker", "firewall", "repository", "mailcow config", "start containers", ] console.progress(0, len(steps), "starting") install_system_packages(runner) console.progress(1, len(steps), steps[0]) ensure_swap(ctx, runner, console) console.progress(2, len(steps), steps[1]) install_docker(ctx, runner, console) console.progress(3, len(steps), steps[2]) configure_ufw(ctx, runner, console) console.progress(4, len(steps), steps[3]) clone_or_update_mailcow(ctx, runner, console) console.progress(5, len(steps), steps[4]) generate_mailcow_config(ctx, runner, console) console.progress(6, len(steps), steps[5]) start_mailcow(ctx, runner) console.progress(7, len(steps), steps[6]) report.add(CheckResult("install", "mailcow", "OK", "Installation workflow completed")) def doctor_checks(ctx: InstallContext, runner: Runner) -> List[CheckResult]: if not ctx.domain or not ctx.hostname: raise InstallerError("domain and hostname are required for doctor.") results: List[CheckResult] = [] a_records = dig(runner, ctx.hostname, "A") or resolve_ipv4(ctx.hostname) results.append(CheckResult("dns", "A", "OK" if a_records else "FAIL", f"{ctx.hostname} A records: {a_records}")) if ctx.public_ip: results.append(CheckResult("dns", "A matches expected IP", "OK" if ctx.public_ip in a_records else "WARN", f"Expected {ctx.public_ip}; got {a_records}")) mx_records = dig(runner, ctx.domain, "MX") mx_ok = any(ctx.hostname.rstrip(".") in record.rstrip(".") for record in mx_records) results.append(CheckResult("dns", "MX", "OK" if mx_ok else "FAIL", f"MX records: {mx_records}")) txt_root = dig(runner, ctx.domain, "TXT") spf_records = [record for record in txt_root if "v=spf1" in record] if not spf_records: results.append(CheckResult("dns", "SPF", "FAIL", "No SPF record found")) elif len(spf_records) > 1: results.append(CheckResult("dns", "SPF", "WARN", f"Multiple SPF records found: {spf_records}")) else: results.append(CheckResult("dns", "SPF", "OK", spf_records[0])) dmarc_records = dig(runner, f"_dmarc.{ctx.domain}", "TXT") dmarc_ok = any("v=DMARC1" in record for record in dmarc_records) results.append(CheckResult("dns", "DMARC", "OK" if dmarc_ok else "FAIL", f"DMARC records: {dmarc_records}")) dkim_records = dig(runner, f"dkim._domainkey.{ctx.domain}", "TXT") dkim_ok = any("v=DKIM1" in record for record in dkim_records) status = "OK" if dkim_ok else "WARN" results.append(CheckResult("dns", "DKIM", status, "DKIM found" if dkim_ok else "DKIM not found yet; generate it in mailcow UI and add TXT record")) if ctx.public_ip: ptr = dig_reverse(runner, ctx.public_ip) ptr_ok = any(item.rstrip(".") == ctx.hostname.rstrip(".") for item in ptr) results.append(CheckResult("dns", "PTR", "OK" if ptr_ok else "FAIL", f"PTR for {ctx.public_ip}: {ptr}; expected {ctx.hostname}")) for name in ["autodiscover", "autoconfig"]: cname = dig(runner, f"{name}.{ctx.domain}", "CNAME") status = "OK" if cname else "WARN" results.append(CheckResult("dns", name, status, f"CNAME: {cname}")) for port in [25, 443, 587, 993]: ok = tcp_connect(ctx.hostname, port, timeout=6) status = "OK" if ok else "WARN" if port in {587, 993} else "FAIL" results.append(CheckResult("network", f"tcp {port}", status, f"{ctx.hostname}:{port} {'reachable' if ok else 'not reachable'}")) tls_ok, tls_msg = check_tls(ctx.hostname, 443) results.append(CheckResult("tls", "https certificate", "OK" if tls_ok else "WARN", tls_msg[:300])) return results def postcheck(ctx: InstallContext, runner: Runner) -> List[CheckResult]: require_root(ctx.action) results: List[CheckResult] = [] repo_ok = ctx.install_dir.exists() and (ctx.install_dir / "docker-compose.yml").exists() results.append(CheckResult("mailcow", "repository", "OK" if repo_ok else "FAIL", str(ctx.install_dir))) conf_ok = (ctx.install_dir / "mailcow.conf").exists() results.append(CheckResult("mailcow", "configuration", "OK" if conf_ok else "FAIL", str(ctx.install_dir / "mailcow.conf"))) if repo_ok and command_exists("docker"): result = runner.run(["docker", "compose", "ps"], cwd=ctx.install_dir, check=False, capture=True) status = "OK" if result.rc == 0 and result.stdout.strip() else "WARN" results.append(CheckResult("docker", "compose ps", status, "docker compose ps executed", {"stdout_tail": result.stdout[-4000:]})) ports = listening_ports(runner) present = sorted([port for port in REQUIRED_MAILCOW_PORTS if port in ports]) results.append(CheckResult("network", "listening ports", "OK" if present else "WARN", f"Listening mailcow ports detected: {present}")) return results class CloudflareClient: def __init__(self, token: str, zone_id: str, dry_run: bool, console: Console) -> None: self.token = token self.zone_id = zone_id self.dry_run = dry_run self.console = console def request(self, method: str, path: str, payload: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: url = CLOUDFLARE_API_BASE + path data = None headers = {"Authorization": f"Bearer {self.token}", "Content-Type": "application/json"} if payload is not None: data = json.dumps(payload).encode("utf-8") if self.dry_run and method in {"POST", "PUT", "PATCH", "DELETE"}: self.console.info(f"DRY-RUN Cloudflare {method} {url} payload={payload}") return {"success": True, "result": {"id": "dry-run"}} request = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(request, timeout=20) as response: return json.loads(response.read().decode("utf-8")) except urllib.error.HTTPError as exc: body = exc.read().decode("utf-8", errors="replace") raise InstallerError(f"Cloudflare API error {exc.code}: {body}") from exc def list_records(self, name: str, record_type: Optional[str] = None) -> List[Dict[str, Any]]: query = {"name": name} if record_type: query["type"] = record_type path = f"/zones/{self.zone_id}/dns_records?" + urllib.parse.urlencode(query) response = self.request("GET", path) if not response.get("success"): raise InstallerError(f"Cloudflare list_records failed: {response}") return list(response.get("result") or []) def create_record(self, payload: Dict[str, Any]) -> Dict[str, Any]: response = self.request("POST", f"/zones/{self.zone_id}/dns_records", payload) if not response.get("success"): raise InstallerError(f"Cloudflare create_record failed: {response}") return response def update_record(self, record_id: str, payload: Dict[str, Any]) -> Dict[str, Any]: response = self.request("PUT", f"/zones/{self.zone_id}/dns_records/{record_id}", payload) if not response.get("success"): raise InstallerError(f"Cloudflare update_record failed: {response}") return response def apply_cloudflare_dns(ctx: InstallContext, console: Console, report: ReportStore) -> None: if not ctx.domain or not ctx.hostname: raise InstallerError("domain and hostname are required for Cloudflare DNS apply.") if not ctx.cloudflare_zone_id: raise InstallerError("--cloudflare-zone-id is required.") if not ctx.cloudflare_token: raise InstallerError("Cloudflare token is required. Use CLOUDFLARE_API_TOKEN or --cloudflare-token.") records = build_dns_records(ctx.domain, ctx.hostname, ctx.public_ip, ctx.ipv6) records = [record for record in records if " Dict[str, Any]: if not ctx.mailcow_api_url or not ctx.mailcow_api_key: raise InstallerError("Mailcow API URL and key are required.") url = ctx.mailcow_api_url.rstrip("/") + endpoint headers = {"X-API-Key": ctx.mailcow_api_key, "Content-Type": "application/json"} if ctx.dry_run: console.info(f"DRY-RUN mailcow API POST {url} payload={payload}") return {"dry_run": True} request = urllib.request.Request(url, data=json.dumps(payload).encode("utf-8"), headers=headers, method="POST") try: with urllib.request.urlopen(request, timeout=20) as response: raw = response.read().decode("utf-8", errors="replace") try: return json.loads(raw) except json.JSONDecodeError: return {"raw": raw} except urllib.error.HTTPError as exc: body = exc.read().decode("utf-8", errors="replace") raise InstallerError(f"mailcow API error {exc.code}: {body}") from exc def mailcow_bootstrap(ctx: InstallContext, console: Console, report: ReportStore) -> None: if not ctx.domain: raise InstallerError("domain is required for mailcow bootstrap.") if ctx.mailcow_create_domain: payload = { "domain": ctx.domain, "description": ctx.client_name or ctx.domain, "aliases": 400, "mailboxes": 400, "defquota": 3072, "maxquota": 10240, "quota": 10240, "active": 1, } response = mailcow_api_request(ctx, "/api/v1/add/domain", payload, console) report.add(CheckResult("mailcow-api", "create domain", "OK", f"Domain request sent: {response}")) if ctx.mailcow_create_mailbox: password = os.getenv(ctx.mailcow_mailbox_password_env, "") if not password and not ctx.dry_run: raise InstallerError(f"Mailbox password env var is missing: {ctx.mailcow_mailbox_password_env}") local_part = ctx.mailcow_create_mailbox.split("@", 1)[0] payload = { "local_part": local_part, "domain": ctx.domain, "name": local_part, "password": password or "", "password2": password or "", "quota": 3072, "active": 1, } response = mailcow_api_request(ctx, "/api/v1/add/mailbox", payload, console) report.add(CheckResult("mailcow-api", "create mailbox", "OK", f"Mailbox request sent: {response}")) def target_volume(ctx: InstallContext) -> int: volume = int(getattr(ctx, "target_volume", 100000) or 100000) return max(volume, 1000) def service_tier(ctx: InstallContext) -> str: return str(getattr(ctx, "service_tier", "100k") or "100k") def render_env_template(ctx: InstallContext) -> str: if not ctx.domain: raise InstallerError("domain is required for env-template.") host = ctx.hostname or f"mail.{ctx.domain}" smtp_user = ctx.smtp_user or f"noreply@{ctx.domain}" password_env = ctx.smtp_password_env return textwrap.dedent( f""" # Mailcow SMTP environment template # Copy this file to the application environment manager and replace secrets. EMAIL_HOST={host} EMAIL_PORT=587 EMAIL_HOST_USER={smtp_user} {password_env}= EMAIL_USE_TLS=1 DEFAULT_FROM_EMAIL={smtp_user} SERVER_EMAIL={smtp_user} # Optional deployment metadata MAIL_DOMAIN={ctx.domain} MAIL_HOSTNAME={host} MAIL_TARGET_VOLUME_MONTHLY={target_volume(ctx)} MAIL_SERVICE_TIER={service_tier(ctx)} """ ).strip() + "\n" def save_env_template(ctx: InstallContext, report: ReportStore) -> Path: content = render_env_template(ctx) path = ctx.report_dir / f"django_email_env_{ctx.domain}_{utc_stamp()}.env.example" write_text(path, content, mode=0o600) report.add_artifact("env_template", path) return path def warmup_rows(monthly_target: int) -> List[Dict[str, Any]]: # Conservative schedule for a new dedicated IP/domain pair. daily_target = max(100, int(monthly_target / 30)) plan = [ (1, 50), (2, 100), (3, 200), (4, 400), (5, 800), (6, 1200), (7, 1800), (8, 2500), (9, 3500), (10, 5000), (11, 6500), (12, 8000), (13, 10000), (14, 12500), (15, 15000), (16, 18000), (17, 22000), (18, 26000), (19, 30000), (20, 35000), (21, 40000), (22, 45000), (23, 50000), (24, 60000), (25, 70000), (26, 80000), (27, 90000), (28, 100000), ] rows: List[Dict[str, Any]] = [] for day, raw_limit in plan: limit = min(raw_limit, daily_target) if monthly_target <= 100000 and day > 18 and limit >= daily_target: phase = "stabilize" elif day <= 7: phase = "seed" elif day <= 14: phase = "ramp" elif day <= 21: phase = "scale" else: phase = "production" rows.append( { "day": day, "phase": phase, "daily_limit": limit, "monthly_equivalent": limit * 30, "action": "Increase only if bounce, complaint and queue metrics are clean.", } ) return rows def render_warmup_plan(ctx: InstallContext) -> str: if not ctx.domain: raise InstallerError("domain is required for warmup-plan.") volume = target_volume(ctx) rows = warmup_rows(volume) lines = [ f"# Warm-up plan for {ctx.domain}", "", f"Client: `{ctx.client_name or '-'}`", f"Mail host: `{ctx.hostname or 'mail.' + ctx.domain}`", f"Monthly target: `{volume}` messages/month", f"Service tier: `{service_tier(ctx)}`", "", "## Operational rule", "", "Never jump directly to the target volume on a fresh IP or a fresh sending domain. Increase progressively and stop the ramp if bounce rate, complaint rate, queue size or blacklist signals deteriorate.", "", "## Daily ramp", "", "| Day | Phase | Daily limit | Monthly equivalent | Go / no-go rule |", "|---:|---|---:|---:|---|", ] for row in rows: lines.append( f"| {row['day']} | {row['phase']} | {row['daily_limit']} | {row['monthly_equivalent']} | {row['action']} |" ) lines.extend( [ "", "## Daily checks during warm-up", "", "- Queue size must return to normal after each campaign window.", "- Bounce addresses must be suppressed quickly.", "- Complaint sources must be removed from future sends.", "- SPF, DKIM, DMARC and PTR must stay valid.", "- Do not mix marketing lists with critical transactional email on day one.", "", "## Go / no-go thresholds", "", "| Metric | OK | Warning | Stop ramp |", "|---|---:|---:|---:|", "| Hard bounce rate | < 2% | 2-5% | > 5% |", "| Complaint rate | < 0.05% | 0.05-0.10% | > 0.10% |", "| Unknown users | rare | increasing | repeated pattern |", "| Queue backlog | clears quickly | delayed | persistent |", "| Blacklist signal | none | one low-impact hit | major DNSBL hit |", ] ) return "\n".join(lines) + "\n" def save_warmup_plan(ctx: InstallContext, report: ReportStore) -> Tuple[Path, Path]: md = render_warmup_plan(ctx) md_path = ctx.report_dir / f"warmup_plan_{ctx.domain}_{target_volume(ctx)}_{utc_stamp()}.md" write_text(md_path, md) csv_path = ctx.report_dir / f"warmup_plan_{ctx.domain}_{target_volume(ctx)}_{utc_stamp()}.csv" csv_path.parent.mkdir(parents=True, exist_ok=True) with csv_path.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=["day", "phase", "daily_limit", "monthly_equivalent", "action"]) writer.writeheader() writer.writerows(warmup_rows(target_volume(ctx))) report.add_artifact("warmup_plan_md", md_path) report.add_artifact("warmup_plan_csv", csv_path) return md_path, csv_path def reverse_ipv4(ip: str) -> str: return ".".join(reversed(ip.split("."))) def dnsbl_lookup(ctx: InstallContext, runner: Runner) -> List[CheckResult]: if not ctx.public_ip: raise InstallerError("public IP is required for blacklist-doctor.") try: ipaddress.IPv4Address(ctx.public_ip) except ValueError as exc: raise InstallerError("DNSBL checks currently require an IPv4 address.") from exc results: List[CheckResult] = [] reversed_ip = reverse_ipv4(ctx.public_ip) for dnsbl in COMMON_DNSBLS: query = f"{reversed_ip}.{dnsbl}" answers = dig(runner, query, "A") if answers: results.append(CheckResult("dnsbl", dnsbl, "FAIL", f"Listed: {answers}", {"query": query, "answers": answers})) else: results.append(CheckResult("dnsbl", dnsbl, "OK", "No listing detected", {"query": query})) return results def readiness_score(checks: List[CheckResult]) -> Tuple[int, List[str]]: score = 100 blockers: List[str] = [] for item in checks: if item.status == "FAIL": if item.category in {"dns", "dnsbl", "network", "system"}: score -= 18 blockers.append(f"{item.category}/{item.name}: {item.message}") else: score -= 10 elif item.status == "WARN": score -= 6 score = max(0, min(100, score)) return score, blockers[:12] def add_readiness_score(report: ReportStore) -> None: score, blockers = readiness_score(report.checks) status = "OK" if score >= 85 and not blockers else "WARN" if score >= 65 else "FAIL" message = f"Readiness score is {score}/100" if blockers: message += f"; blockers: {' | '.join(blockers[:4])}" report.add(CheckResult("readiness", "client delivery", status, message, {"score": score, "blockers": blockers})) def render_client_checklist(ctx: InstallContext, score: Optional[int] = None) -> str: if not ctx.domain: raise InstallerError("domain is required for client checklist.") return textwrap.dedent( f""" # Client delivery checklist - {ctx.domain} Client: {ctx.client_name or '-'} Mail host: {ctx.hostname or 'mail.' + ctx.domain} Public IPv4: {ctx.public_ip or '-'} Target monthly volume: {target_volume(ctx)} Service tier: {service_tier(ctx)} Readiness score: {score if score is not None else 'not computed'} ## DNS - [ ] A record points the mail hostname to the public IP. - [ ] MX record points the root domain to the mail hostname. - [ ] SPF has one valid TXT record only. - [ ] DKIM public key is copied from mailcow to DNS. - [ ] DMARC exists and starts at p=quarantine or stricter after validation. - [ ] PTR/rDNS is configured at the server provider and matches the mail hostname. ## Mailcow - [ ] Admin password changed after first login. - [ ] Domain created in mailcow. - [ ] DKIM generated for the domain. - [ ] First mailbox or SMTP sender created. - [ ] Quotas and sender policies reviewed. - [ ] Backups configured. ## Application integration - [ ] Django SMTP settings deployed from the generated snippet. - [ ] SMTP password stored in environment variables only. - [ ] Test email sent from production-like environment. - [ ] Bounce handling strategy defined. - [ ] Critical transactional mail separated from marketing traffic. ## Warm-up - [ ] Warm-up plan approved. - [ ] Daily sending limit configured outside the application if possible. - [ ] Queue, bounces, complaints and blacklist signals monitored daily. - [ ] Ramp stopped automatically or manually when thresholds are exceeded. """ ).strip() + "\n" def save_client_checklist(ctx: InstallContext, report: ReportStore, score: Optional[int] = None) -> Path: content = render_client_checklist(ctx, score) path = ctx.report_dir / f"client_delivery_checklist_{ctx.domain}_{utc_stamp()}.md" write_text(path, content) report.add_artifact("client_checklist", path) return path def render_operations_runbook(ctx: InstallContext) -> str: if not ctx.domain: raise InstallerError("domain is required for operations runbook.") host = ctx.hostname or f"mail.{ctx.domain}" return textwrap.dedent( f""" # Operations runbook - {ctx.domain} ## Useful URLs - Mailcow admin: https://{host}/admin - Webmail: https://{host}/SOGo/ ## Daily checks ```bash cd {ctx.install_dir} docker compose ps docker compose logs --tail=120 postfix-mailcow docker compose logs --tail=120 rspamd-mailcow ``` ## Queue inspection ```bash cd {ctx.install_dir} docker compose exec postfix-mailcow postqueue -p ``` ## Restart stack ```bash cd {ctx.install_dir} docker compose restart ``` ## Update mailcow ```bash cd {ctx.install_dir} ./update.sh ``` ## Emergency actions 1. Stop marketing or bulk sends immediately. 2. Keep transactional traffic only if reputation is clean. 3. Check DNS, queue, bounces, complaints and DNSBL results. 4. Fix the root cause before resuming the warm-up ramp. """ ).strip() + "\n" def save_operations_runbook(ctx: InstallContext, report: ReportStore) -> Path: content = render_operations_runbook(ctx) path = ctx.report_dir / f"operations_runbook_{ctx.domain}_{utc_stamp()}.md" write_text(path, content) report.add_artifact("operations_runbook", path) return path def backup_mailcow_config(ctx: InstallContext, report: ReportStore, console: Console) -> Path: require_root(ctx.action) if not ctx.install_dir.exists(): raise InstallerError(f"Install directory does not exist: {ctx.install_dir}") backup_path = ctx.report_dir / f"mailcow_config_backup_{ctx.domain or 'unknown'}_{utc_stamp()}.tar.gz" include_names = [ "mailcow.conf", "docker-compose.yml", "data/conf", ] if ctx.dry_run: console.info(f"DRY-RUN: create config backup {backup_path}") report.add_artifact("config_backup_dry_run", backup_path) return backup_path backup_path.parent.mkdir(parents=True, exist_ok=True) with tarfile.open(backup_path, "w:gz") as tar: for name in include_names: item = ctx.install_dir / name if item.exists(): tar.add(item, arcname=f"mailcow/{name}") report.add_artifact("config_backup", backup_path) report.add(CheckResult("backup", "mailcow config", "OK", f"Config backup created: {backup_path}")) return backup_path def create_delivery_pack(ctx: InstallContext, report: ReportStore, runner: Runner, console: Console) -> Path: if not ctx.domain: raise InstallerError("domain is required for delivery-pack.") console.title("Client delivery pack") # Generate or refresh key artifacts. dns_path = save_dns_plan(ctx, report) django_path = save_django_snippet(ctx, report) env_path = save_env_template(ctx, report) warmup_md, warmup_csv = save_warmup_plan(ctx, report) runbook_path = save_operations_runbook(ctx, report) # Run non-invasive checks when possible, then compute score. if ctx.hostname: try: report.extend(doctor_checks(ctx, runner)) except InstallerError as exc: report.add(CheckResult("doctor", "delivery pack", "WARN", str(exc))) if ctx.public_ip: try: report.extend(dnsbl_lookup(ctx, runner)) except InstallerError as exc: report.add(CheckResult("dnsbl", "delivery pack", "WARN", str(exc))) add_readiness_score(report) score, _blockers = readiness_score(report.checks) checklist_path = save_client_checklist(ctx, report, score) # Save an intermediate report before zipping. json_path = ctx.report_dir / f"delivery_pack_manifest_{ctx.domain}_{utc_stamp()}.json" manifest = { "app": APP_NAME, "version": APP_VERSION, "client_name": ctx.client_name, "domain": ctx.domain, "hostname": ctx.hostname, "public_ip": ctx.public_ip, "target_volume": target_volume(ctx), "service_tier": service_tier(ctx), "readiness_score": score, "artifacts": report.artifacts, } write_text(json_path, json.dumps(manifest, indent=2, sort_keys=True)) report.add_artifact("delivery_manifest", json_path) pack_label = getattr(ctx, "pack_name", None) or f"mailcow_delivery_pack_{ctx.domain}_{utc_stamp()}" safe_label = re.sub(r"[^a-zA-Z0-9_.-]+", "_", pack_label).strip("_") zip_path = ctx.report_dir / f"{safe_label}.zip" artifact_paths = [ dns_path, django_path, env_path, warmup_md, warmup_csv, runbook_path, checklist_path, json_path, ] with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED) as zf: for artifact_path in artifact_paths: if artifact_path.exists(): zf.write(artifact_path, arcname=artifact_path.name) report.add_artifact("delivery_pack_zip", zip_path) report.add(CheckResult("delivery-pack", "zip", "OK", f"Delivery pack created: {zip_path}", {"readiness_score": score})) return zip_path def report_payload(ctx: InstallContext, report: ReportStore, runner: Runner) -> Dict[str, Any]: return { "app": APP_NAME, "version": APP_VERSION, "generated_at_utc": utc_stamp(), "client_name": ctx.client_name, "action": ctx.action, "domain": ctx.domain, "hostname": ctx.hostname, "public_ip": ctx.public_ip, "ipv6": ctx.ipv6, "install_dir": str(ctx.install_dir), "dry_run": ctx.dry_run, "summary": report.summary(), "checks": [ { "category": item.category, "name": item.name, "status": item.status, "message": item.message, "details": item.details, } for item in report.checks ], "artifacts": report.artifacts, "commands": [ { "command": item.command, "rc": item.rc, "stdout_tail": item.stdout[-2000:], "stderr_tail": item.stderr[-2000:], } for item in runner.history ], } def save_json_report(ctx: InstallContext, report: ReportStore, runner: Runner) -> Path: payload = report_payload(ctx, report, runner) path = ctx.report_dir / f"mailcow_v3_report_{ctx.action}_{utc_stamp()}.json" write_text(path, json.dumps(payload, indent=2, sort_keys=True)) report.add_artifact("json_report", path) return path def html_status_class(status: str) -> str: return {"OK": "ok", "WARN": "warn", "FAIL": "fail"}.get(status, "warn") def save_html_report(ctx: InstallContext, report: ReportStore, runner: Runner) -> Path: payload = report_payload(ctx, report, runner) rows = [] for item in payload["checks"]: rows.append( "" f"{html.escape(item['category'])}" f"{html.escape(item['name'])}" f"{html.escape(item['status'])}" f"{html.escape(item['message'])}" "" ) artifact_rows = [] for key, value in report.artifacts.items(): artifact_rows.append(f"{html.escape(key)}{html.escape(value)}") doc = f""" Mailcow V3 Report - {html.escape(ctx.domain or '')}

Mailcow Assisted Installer V3 Report

Generated UTC: {html.escape(payload['generated_at_utc'])}
Client: {html.escape(str(ctx.client_name or '-'))}
Domain: {html.escape(str(ctx.domain or '-'))} | Host: {html.escape(str(ctx.hostname or '-'))} | IP: {html.escape(str(ctx.public_ip or '-'))}

Summary

OK: {payload['summary'].get('OK', 0)} WARN: {payload['summary'].get('WARN', 0)} FAIL: {payload['summary'].get('FAIL', 0)}

Checks

{''.join(rows)}
CategoryNameStatusMessage

Artifacts

{''.join(artifact_rows)}
NamePath

Commercial next steps

  • Set or verify PTR/rDNS at the IP provider.
  • Generate DKIM from mailcow and publish the TXT record.
  • Run the doctor again after DNS propagation.
  • Start IP warm-up progressively before high-volume sending.
  • Monitor bounce rate, complaints, queue size and blacklist status.
""" path = ctx.report_dir / f"mailcow_v3_report_{ctx.action}_{utc_stamp()}.html" write_text(path, doc) report.add_artifact("html_report", path) return path def save_profile_template(ctx: InstallContext, report: ReportStore) -> Path: profile = { "client_name": "Client Name", "domain": "example.com", "hostname": "mail.example.com", "public_ip": "1.2.3.4", "ipv6": None, "timezone": "Europe/Madrid", "smtp_user": "noreply@example.com", "low_memory": False, "configure_ufw": True, "cloudflare_zone_id": "", "target_volume": 100000, "service_tier": "100k", "pack_name": "", } path = ctx.report_dir / f"mailcow_client_profile_template_{utc_stamp()}.json" write_text(path, json.dumps(profile, indent=2)) report.add_artifact("profile_template", path) return path def load_profile(path: Optional[str]) -> Dict[str, Any]: if not path: return {} data = json.loads(Path(path).expanduser().read_text(encoding="utf-8")) if not isinstance(data, dict): raise InstallerError("Profile must be a JSON object.") return data def value_from_args_profile(args: argparse.Namespace, profile: Dict[str, Any], name: str, default: Any = None) -> Any: value = getattr(args, name, None) if value not in (None, False, ""): return value return profile.get(name, default) def build_context(args: argparse.Namespace) -> InstallContext: profile = load_profile(args.profile) domain = normalize_domain(value_from_args_profile(args, profile, "domain")) hostname = normalize_hostname(value_from_args_profile(args, profile, "hostname"), domain) public_ip = normalize_ip(value_from_args_profile(args, profile, "public_ip")) ipv6 = normalize_ip(value_from_args_profile(args, profile, "ipv6")) report_dir = Path(value_from_args_profile(args, profile, "report_dir", DEFAULT_REPORT_DIR)).expanduser().resolve() install_dir = Path(value_from_args_profile(args, profile, "install_dir", DEFAULT_INSTALL_DIR)).expanduser().resolve() return InstallContext( action=args.action, domain=domain, hostname=hostname, public_ip=public_ip, ipv6=ipv6, timezone=value_from_args_profile(args, profile, "timezone", DEFAULT_TIMEZONE), install_dir=install_dir, report_dir=report_dir, dry_run=args.dry_run, yes=args.yes, quiet=args.quiet, low_memory=bool(value_from_args_profile(args, profile, "low_memory", False)), create_swap_gb=int(value_from_args_profile(args, profile, "create_swap_gb", 1)), skip_docker_install=args.skip_docker_install, remove_conflicting_docker=args.remove_conflicting_docker, update_existing_repo=args.update_existing_repo, pull_images=not args.no_pull, start_mailcow=not args.no_start, configure_ufw=bool(value_from_args_profile(args, profile, "configure_ufw", False)) or args.configure_ufw, check_outbound_25=args.check_outbound_25, cloudflare_zone_id=value_from_args_profile(args, profile, "cloudflare_zone_id"), cloudflare_token=args.cloudflare_token or os.getenv("CLOUDFLARE_API_TOKEN"), cloudflare_overwrite=args.cloudflare_overwrite, smtp_user=value_from_args_profile(args, profile, "smtp_user"), smtp_password_env=args.smtp_password_env, django_settings_module=args.django_settings_module, mailcow_api_url=args.mailcow_api_url, mailcow_api_key=args.mailcow_api_key or os.getenv("MAILCOW_API_KEY"), mailcow_create_domain=args.mailcow_create_domain, mailcow_create_mailbox=args.mailcow_create_mailbox, mailcow_mailbox_password_env=args.mailcow_mailbox_password_env, client_name=value_from_args_profile(args, profile, "client_name"), target_volume=int(value_from_args_profile(args, profile, "target_volume", 100000)), pack_name=value_from_args_profile(args, profile, "pack_name"), service_tier=value_from_args_profile(args, profile, "service_tier", "100k"), ) def print_summary(ctx: InstallContext, console: Console) -> None: console.title("Execution context") for key, value in [ ("Action", ctx.action), ("Client", ctx.client_name or "-"), ("Domain", ctx.domain or "-"), ("Hostname", ctx.hostname or "-"), ("Public IPv4", ctx.public_ip or "-"), ("IPv6", ctx.ipv6 or "-"), ("Timezone", ctx.timezone), ("Install dir", str(ctx.install_dir)), ("Report dir", str(ctx.report_dir)), ("Target volume", str(getattr(ctx, "target_volume", 100000))), ("Service tier", str(getattr(ctx, "service_tier", "100k"))), ("Dry-run", str(ctx.dry_run)), ]: console.line(f"{key:14s}: {value}") def maybe_detect_ip(ctx: InstallContext, console: Console) -> None: if ctx.public_ip or ctx.action == "profile-template": return detected = detect_public_ipv4(console) if detected: ctx.public_ip = detected console.ok(f"Detected public IPv4: {detected}") else: console.warn("Public IPv4 could not be detected.") def create_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="mailcow_assisted_installer_v3.py", description="Assisted mailcow installer with DNS, Cloudflare, doctor and Django helpers.", formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "action", choices=[ "profile-template", "preflight", "dns-plan", "dns-apply-cloudflare", "install", "postcheck", "doctor", "django-snippet", "mailcow-bootstrap", "configure-firewall", "env-template", "warmup-plan", "blacklist-doctor", "readiness-score", "backup-config", "delivery-pack", "all", ], ) parser.add_argument("--profile", help="JSON client profile file.") parser.add_argument("--client-name", dest="client_name", help="Client name for reports.") parser.add_argument("--target-volume", type=int, default=100000, help="Target monthly volume for warm-up and delivery pack.") parser.add_argument("--service-tier", default="100k", help="Commercial service tier label, for example 100k, 250k, 500k.") parser.add_argument("--pack-name", help="Optional delivery pack ZIP base name.") parser.add_argument("--domain", help="Mail domain, for example example.com.") parser.add_argument("--hostname", help="Mail server FQDN, for example mail.example.com.") parser.add_argument("--public-ip", help="Public IPv4 address.") parser.add_argument("--ipv6", help="Public IPv6 address.") parser.add_argument("--timezone", default=DEFAULT_TIMEZONE) parser.add_argument("--install-dir", default=str(DEFAULT_INSTALL_DIR)) parser.add_argument("--report-dir", default=str(DEFAULT_REPORT_DIR)) parser.add_argument("--dry-run", action="store_true") parser.add_argument("--yes", action="store_true") parser.add_argument("--quiet", action="store_true") parser.add_argument("--low-memory", action="store_true") parser.add_argument("--create-swap-gb", type=int, default=1) parser.add_argument("--skip-docker-install", action="store_true") parser.add_argument("--remove-conflicting-docker", action="store_true") parser.add_argument("--update-existing-repo", action="store_true") parser.add_argument("--no-pull", action="store_true") parser.add_argument("--no-start", action="store_true") parser.add_argument("--configure-ufw", action="store_true") parser.add_argument("--check-outbound-25", action="store_true") parser.add_argument("--cloudflare-zone-id", help="Cloudflare zone id.") parser.add_argument("--cloudflare-token", help="Cloudflare API token. Prefer CLOUDFLARE_API_TOKEN env var.") parser.add_argument("--cloudflare-overwrite", action="store_true", help="Update existing Cloudflare records.") parser.add_argument("--smtp-user", help="SMTP user for Django snippet, for example noreply@example.com.") parser.add_argument("--smtp-password-env", default="EMAIL_HOST_PASSWORD") parser.add_argument("--django-settings-module", help="Reserved for future direct patch mode.") parser.add_argument("--mailcow-api-url", help="Mailcow base URL, for example https://mail.example.com") parser.add_argument("--mailcow-api-key", help="Mailcow API key. Prefer MAILCOW_API_KEY env var.") parser.add_argument("--mailcow-create-domain", action="store_true") parser.add_argument("--mailcow-create-mailbox", help="Mailbox address to create, for example noreply@example.com.") parser.add_argument("--mailcow-mailbox-password-env", default="MAILCOW_MAILBOX_PASSWORD") return parser def execute(ctx: InstallContext, console: Console, runner: Runner, report: ReportStore) -> None: if ctx.action == "profile-template": path = save_profile_template(ctx, report) console.ok(f"Profile template written to {path}") return if ctx.action == "preflight": report.extend(check_preflight(ctx, runner)) return if ctx.action == "dns-plan": path = save_dns_plan(ctx, report) console.line(render_dns_plan(ctx)) console.ok(f"DNS plan written to {path}") return if ctx.action == "dns-apply-cloudflare": confirm(ctx, console, "Apply DNS records to Cloudflare?") apply_cloudflare_dns(ctx, console, report) return if ctx.action == "install": confirm(ctx, console, "Install mailcow on this host?") report.extend(check_preflight(ctx, runner)) failures = [item for item in report.checks if item.is_fail and item.category in {"system", "network"}] if failures: raise InstallerError("Critical preflight failures detected. Fix them before install.") run_install(ctx, runner, console, report) return if ctx.action == "postcheck": report.extend(postcheck(ctx, runner)) return if ctx.action == "doctor": report.extend(doctor_checks(ctx, runner)) return if ctx.action == "django-snippet": path = save_django_snippet(ctx, report) console.line(render_django_snippet(ctx)) console.ok(f"Django SMTP snippet written to {path}") return if ctx.action == "mailcow-bootstrap": confirm(ctx, console, "Call the mailcow API to create domain/mailbox?") mailcow_bootstrap(ctx, console, report) return if ctx.action == "configure-firewall": require_root(ctx.action) confirm(ctx, console, "Configure UFW firewall for mailcow ports?") configure_ufw(ctx, runner, console) report.add(CheckResult("firewall", "ufw", "OK", "Firewall workflow completed")) return if ctx.action == "env-template": path = save_env_template(ctx, report) console.line(render_env_template(ctx)) console.ok(f"Environment template written to {path}") return if ctx.action == "warmup-plan": md_path, csv_path = save_warmup_plan(ctx, report) console.line(render_warmup_plan(ctx)) console.ok(f"Warm-up plan written to {md_path} and {csv_path}") return if ctx.action == "blacklist-doctor": report.extend(dnsbl_lookup(ctx, runner)) return if ctx.action == "readiness-score": report.extend(check_preflight(ctx, runner)) if ctx.hostname: report.extend(doctor_checks(ctx, runner)) if ctx.public_ip: report.extend(dnsbl_lookup(ctx, runner)) add_readiness_score(report) return if ctx.action == "backup-config": confirm(ctx, console, "Create a mailcow configuration backup archive?") backup_mailcow_config(ctx, report, console) return if ctx.action == "delivery-pack": create_delivery_pack(ctx, report, runner, console) return if ctx.action == "all": confirm(ctx, console, "Run preflight, DNS plan, install, postcheck and doctor?") report.extend(check_preflight(ctx, runner)) failures = [item for item in report.checks if item.is_fail and item.category in {"system", "network"}] if failures: raise InstallerError("Critical preflight failures detected. Fix them before install.") save_dns_plan(ctx, report) save_django_snippet(ctx, report) run_install(ctx, runner, console, report) time.sleep(2) report.extend(postcheck(ctx, runner)) try: report.extend(doctor_checks(ctx, runner)) except InstallerError as exc: report.add(CheckResult("doctor", "skipped", "WARN", str(exc))) try: save_warmup_plan(ctx, report) save_env_template(ctx, report) save_operations_runbook(ctx, report) add_readiness_score(report) except InstallerError as exc: report.add(CheckResult("delivery", "artifacts", "WARN", str(exc))) return raise InstallerError(f"Unsupported action: {ctx.action}") def main(argv: Optional[Sequence[str]] = None) -> int: parser = create_parser() args = parser.parse_args(argv) console = Console(args.quiet) try: ctx = build_context(args) ctx.report_dir.mkdir(parents=True, exist_ok=True) maybe_detect_ip(ctx, console) print_summary(ctx, console) runner = Runner(console, ctx.dry_run) report = ReportStore(ctx, console) execute(ctx, console, runner, report) json_path = save_json_report(ctx, report, runner) html_path = save_html_report(ctx, report, runner) console.title("Final report") counts = report.summary() console.line(f"OK={counts.get('OK', 0)} WARN={counts.get('WARN', 0)} FAIL={counts.get('FAIL', 0)}") console.line(f"JSON: {json_path}") console.line(f"HTML: {html_path}") if ctx.action in {"install", "all"} and ctx.hostname: console.line(f"Admin URL: https://{ctx.hostname}/admin") console.line("Default admin user is usually: admin") console.line("Change the default password immediately at first login.") return 0 except KeyboardInterrupt: console.fail("Interrupted.") return 130 except InstallerError as exc: console.fail(str(exc)) return 2 except Exception as exc: console.fail(f"Unexpected error: {exc}") return 1 if __name__ == "__main__": raise SystemExit(main())