#!/usr/bin/env python3
# SPDX-FileCopyrightText: 2026 Eric Ireland
# SPDX-License-Identifier: GPL-3.0-or-later
#
# This file is part of vmailctl.
"""Safely manage a small Postfix/Dovecot flat-file virtual mail setup."""

from __future__ import annotations

import argparse
import configparser
import datetime as dt
import fcntl
import getpass
import os
from pathlib import Path
import re
import secrets
import shutil
import stat
import string
import subprocess
import sys
import tempfile
from contextlib import contextmanager
from dataclasses import dataclass
from typing import Iterable


VERSION = "0.1.0"
DEFAULT_CONFIG = Path("/etc/vmailctl.conf")
LOCALPART_RE = re.compile(r"^[a-z0-9][a-z0-9._-]{0,62}$")
DOMAIN_LABEL_RE = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$")
GENERATED_PASSWORD_ALPHABET = string.ascii_letters + string.digits + "-._~"


class VmailError(Exception):
    """A safe, user-facing failure."""


@dataclass(frozen=True)
class Settings:
    domain: str
    dovecot_users: Path
    postfix_mailboxes: Path
    postfix_aliases: Path
    mail_root: Path
    uid: int
    gid: int
    hash_scheme: str
    folders: tuple[str, ...]
    backup_dir: Path
    lock_file: Path
    postmap: Path
    postfix: Path
    doveadm: Path
    doveconf: Path
    directory_mode: int

    @classmethod
    def load(cls, path: Path) -> "Settings":
        """Load configuration only after establishing that root can trust it.

        Command paths come from this file and are later executed as root, so
        the configuration file's ownership, permissions and symbolic-link
        status are checked before any values are parsed.
        """
        parser = configparser.ConfigParser(interpolation=None)
        descriptor = -1
        try:
            flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0)
            if hasattr(os, "O_NOFOLLOW"):
                flags |= os.O_NOFOLLOW
            descriptor = os.open(path, flags)
            metadata = os.fstat(descriptor)
            validate_regular_metadata(metadata, os.fspath(path), private=False)
            if not hasattr(os, "O_NOFOLLOW"):
                # Detect a link or path replacement on systems without
                # O_NOFOLLOW by comparing the opened file with the live path.
                path_metadata = path.lstat()
                if (
                    path_metadata.st_dev != metadata.st_dev
                    or path_metadata.st_ino != metadata.st_ino
                    or stat.S_ISLNK(path_metadata.st_mode)
                ):
                    raise VmailError(f"{path} must not be a symbolic link")
            with os.fdopen(descriptor, encoding="utf-8") as handle:
                descriptor = -1
                parser.read_file(handle)
        except VmailError:
            raise
        except (OSError, UnicodeError, configparser.Error) as exc:
            raise VmailError(f"cannot read configuration {path}: {exc}") from exc
        finally:
            if descriptor >= 0:
                os.close(descriptor)
        if "vmailctl" not in parser:
            raise VmailError(f"{path} has no [vmailctl] section")
        section = parser["vmailctl"]

        def required(name: str) -> str:
            value = section.get(name, "").strip()
            if not value:
                raise VmailError(f"{path}: missing {name}")
            return value

        def absolute_path(name: str) -> Path:
            value = Path(required(name))
            if not value.is_absolute():
                raise VmailError(f"{path}: {name} must be an absolute path")
            return value

        domain = validate_domain(required("domain"), os.fspath(path))
        try:
            uid = int(required("uid"))
            gid = int(required("gid"))
            directory_mode = int(required("directory_mode"), 8)
        except ValueError as exc:
            raise VmailError(f"{path}: uid, gid and directory_mode must be numeric") from exc
        if uid < 100 or gid < 1:
            raise VmailError(f"{path}: refusing unsafe uid/gid {uid}:{gid}")
        if directory_mode & 0o077 or directory_mode & 0o700 != 0o700:
            raise VmailError(
                f"{path}: directory_mode must grant owner rwx and no group/world access"
            )
        folders = tuple(x.strip() for x in required("folders").split(",") if x.strip())
        validate_folders(folders, os.fspath(path))

        settings = cls(
            domain=domain,
            dovecot_users=absolute_path("dovecot_users"),
            postfix_mailboxes=absolute_path("postfix_mailboxes"),
            postfix_aliases=absolute_path("postfix_aliases"),
            mail_root=absolute_path("mail_root"),
            uid=uid,
            gid=gid,
            hash_scheme=required("hash_scheme").upper(),
            folders=folders,
            backup_dir=absolute_path("backup_dir"),
            lock_file=absolute_path("lock_file"),
            postmap=absolute_path("postmap"),
            postfix=absolute_path("postfix"),
            doveadm=absolute_path("doveadm"),
            doveconf=absolute_path("doveconf"),
            directory_mode=directory_mode,
        )
        for command in (settings.postmap, settings.postfix, settings.doveadm, settings.doveconf):
            validate_secure_directory(
                command.parent,
                f"command directory {command.parent}",
                private=False,
            )
            validate_root_owned_regular(command, f"required command {command}", executable=True)
        return settings

    @property
    def domain_mail_root(self) -> Path:
        return self.mail_root / self.domain

    def mailbox_home(self, localpart: str) -> Path:
        return self.domain_mail_root / localpart

    def mailbox_target(self, localpart: str) -> str:
        return f"{self.domain}/{localpart}/Maildir/"


@dataclass(frozen=True)
class PasswdEntry:
    line_number: int
    address: str
    fields: tuple[str, ...]


def require_root() -> None:
    if os.geteuid() != 0:
        raise VmailError("this command must be run as root (use sudo)")


def validate_regular_metadata(
    metadata: os.stat_result,
    label: str,
    *,
    private: bool,
    executable: bool = False,
) -> None:
    """Apply the trust rules shared by configuration and managed files."""
    mode = stat.S_IMODE(metadata.st_mode)
    if not stat.S_ISREG(metadata.st_mode):
        raise VmailError(f"{label} must be a regular file, not a symlink or special file")
    if metadata.st_uid != 0:
        raise VmailError(f"{label} must be owned by root")
    if mode & 0o022:
        raise VmailError(f"{label} must not be group/world writable ({mode:04o})")
    if private and mode & 0o077:
        raise VmailError(f"{label} must be accessible only to root ({mode:04o})")
    if executable and not mode & 0o111:
        raise VmailError(f"{label} is not executable")


def validate_root_owned_regular(
    path: Path,
    label: str,
    *,
    private: bool = False,
    executable: bool = False,
) -> None:
    try:
        metadata = path.lstat()
    except OSError as exc:
        raise VmailError(f"cannot inspect {label}: {exc}") from exc
    validate_regular_metadata(
        metadata,
        label,
        private=private,
        executable=executable,
    )


def validate_domain(value: str, source: str) -> str:
    domain = value.strip().lower()
    labels = domain.split(".")
    if (
        len(domain) > 253
        or len(labels) < 2
        or any(not DOMAIN_LABEL_RE.fullmatch(label) for label in labels)
    ):
        raise VmailError(f"{source}: invalid domain {domain!r}")
    return domain


def validate_folders(folders: tuple[str, ...], source: str) -> None:
    seen: set[str] = set()
    for folder in folders:
        if (
            folder in {".", ".."}
            or folder.startswith("-")
            or "/" in folder
            or any(ord(character) < 32 or ord(character) == 127 for character in folder)
        ):
            raise VmailError(f"{source}: invalid standard folder name {folder!r}")
        if folder in seen:
            raise VmailError(f"{source}: duplicate standard folder name {folder!r}")
        seen.add(folder)


def validate_secure_directory(
    path: Path,
    label: str,
    *,
    private: bool,
    allow_sticky_world_writable: bool = False,
) -> None:
    """Reject directories in which an untrusted user could replace a file."""
    try:
        metadata = path.lstat()
    except OSError as exc:
        raise VmailError(f"cannot inspect {label}: {exc}") from exc
    mode = stat.S_IMODE(metadata.st_mode)
    if not stat.S_ISDIR(metadata.st_mode):
        raise VmailError(f"{label} must be a directory, not a symlink or special file")
    if metadata.st_uid != 0:
        raise VmailError(f"{label} must be owned by root")
    if private and mode & 0o077:
        raise VmailError(f"{label} must be accessible only to root ({mode:04o})")
    if not private:
        if mode & 0o002 and not (
            allow_sticky_world_writable and metadata.st_mode & stat.S_ISVTX
        ):
            raise VmailError(f"{label} is unsafely world writable ({mode:04o})")
        if mode & 0o020 and metadata.st_gid != 0:
            raise VmailError(f"{label} is writable by a non-root group ({mode:04o})")


def ensure_secure_backup_directory(path: Path) -> None:
    if not path.exists():
        validate_secure_directory(path.parent, f"backup parent {path.parent}", private=False)
        try:
            path.mkdir(mode=0o700)
        except OSError as exc:
            raise VmailError(f"cannot create backup directory {path}: {exc}") from exc
    validate_secure_directory(path, f"backup directory {path}", private=True)


def validate_mail_domain_root(settings: Settings) -> None:
    path = settings.domain_mail_root
    try:
        metadata = path.lstat()
    except OSError as exc:
        raise VmailError(f"cannot inspect mail domain root {path}: {exc}") from exc
    mode = stat.S_IMODE(metadata.st_mode)
    if not stat.S_ISDIR(metadata.st_mode):
        raise VmailError(f"mail domain root {path} must be a real directory")
    if metadata.st_uid not in {0, settings.uid} or metadata.st_gid not in {0, settings.gid}:
        raise VmailError(
            f"mail domain root {path} must be owned by root or virtual-mail "
            f"UID/GID {settings.uid}:{settings.gid}"
        )
    if mode & 0o002:
        raise VmailError(f"mail domain root {path} must not be world writable ({mode:04o})")
    if mode & 0o020 and metadata.st_gid not in {0, settings.gid}:
        raise VmailError(f"mail domain root {path} has an unsafe writable group ({mode:04o})")


def validate_runtime_security(settings: Settings, *, modifying: bool) -> None:
    """Validate every configured path before reading or changing live state."""
    checked_directories: set[Path] = set()
    for path in (
        settings.dovecot_users,
        settings.postfix_mailboxes,
        settings.postfix_aliases,
    ):
        if path.parent not in checked_directories:
            validate_secure_directory(
                path.parent,
                f"managed-file directory {path.parent}",
                private=False,
            )
            checked_directories.add(path.parent)
    validate_root_owned_regular(
        settings.dovecot_users,
        f"Dovecot users file {settings.dovecot_users}",
    )
    for source in (settings.postfix_mailboxes, settings.postfix_aliases):
        validate_root_owned_regular(source, f"Postfix source map {source}")
        database = Path(str(source) + ".db")
        validate_root_owned_regular(database, f"Postfix compiled map {database}")
    validate_mail_domain_root(settings)
    if modifying:
        ensure_secure_backup_directory(settings.backup_dir)


def read_text(path: Path) -> str:
    try:
        return path.read_text(encoding="utf-8")
    except OSError as exc:
        raise VmailError(f"cannot read {path}: {exc}") from exc


def parse_passwd(text: str, source: str) -> dict[str, PasswdEntry]:
    entries: dict[str, PasswdEntry] = {}
    for number, raw in enumerate(text.splitlines(), 1):
        stripped = raw.strip()
        if not stripped or stripped.startswith("#"):
            continue
        fields = tuple(raw.split(":"))
        if len(fields) != 8:
            raise VmailError(f"{source}:{number}: expected 8 colon-separated fields")
        address = fields[0].lower()
        if not address or "@" not in address:
            raise VmailError(f"{source}:{number}: invalid login field")
        if address in entries:
            raise VmailError(f"{source}:{number}: duplicate login {address}")
        entries[address] = PasswdEntry(number, address, fields)
    return entries


def parse_map(text: str, source: str) -> dict[str, tuple[int, str]]:
    entries: dict[str, tuple[int, str]] = {}
    for number, raw in enumerate(text.splitlines(), 1):
        stripped = raw.strip()
        if not stripped or stripped.startswith("#"):
            continue
        fields = stripped.split()
        if len(fields) != 2:
            raise VmailError(f"{source}:{number}: expected exactly two whitespace-separated fields")
        address = fields[0].lower()
        if address in entries:
            raise VmailError(f"{source}:{number}: duplicate map key {address}")
        entries[address] = (number, fields[1])
    return entries


def append_line(text: str, line: str) -> str:
    if text and not text.endswith("\n"):
        text += "\n"
    return text + line + "\n"


def normalize_localpart(value: str, domain: str) -> tuple[str, str]:
    candidate = value.strip().lower()
    if "@" in candidate:
        if candidate.count("@") != 1:
            raise VmailError(f"invalid address: {value!r}")
        localpart, supplied_domain = candidate.rsplit("@", 1)
        if supplied_domain != domain:
            raise VmailError(f"only the {domain} domain is managed")
    else:
        localpart = candidate
    if (
        not LOCALPART_RE.fullmatch(localpart)
        or localpart.endswith(".")
        or ".." in localpart
        or "+" in localpart
    ):
        raise VmailError(
            "local part must be lowercase letters, digits, dots, underscores or hyphens; "
            "it cannot end in a dot or contain consecutive dots"
        )
    return localpart, f"{localpart}@{domain}"


def run(
    argv: Iterable[os.PathLike[str] | str],
    *,
    input_text: str | None = None,
    check: bool = True,
) -> subprocess.CompletedProcess[str]:
    """Run a fixed argument vector without invoking a shell."""
    command = [os.fspath(item) for item in argv]
    result = subprocess.run(
        command,
        input=input_text,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        check=False,
    )
    if check and result.returncode != 0:
        detail = result.stderr.strip() or result.stdout.strip() or f"exit {result.returncode}"
        raise VmailError(f"{command[0]} failed: {detail}")
    return result


def prompt_password() -> str:
    first = getpass.getpass("New mailbox password: ")
    second = getpass.getpass("Confirm password: ")
    if first != second:
        raise VmailError("passwords do not match")
    if len(first) < 12:
        raise VmailError("password must contain at least 12 characters")
    if any(char in first for char in ("\n", "\r", "\x00")):
        raise VmailError("password contains an unsupported control character")
    return first


def generate_password(length: int = 24) -> str:
    return "".join(secrets.choice(GENERATED_PASSWORD_ALPHABET) for _ in range(length))


def hash_password(settings: Settings, password: str) -> str:
    result = run(
        [settings.doveadm, "pw", "-s", settings.hash_scheme],
        input_text=f"{password}\n{password}\n",
    )
    expected_prefix = "{" + settings.hash_scheme + "}"
    hashes = [line.strip() for line in result.stdout.splitlines() if line.strip()]
    if len(hashes) != 1 or not hashes[0].startswith(expected_prefix):
        raise VmailError(f"doveadm did not return a {expected_prefix} password hash")
    return hashes[0]


def write_temp_like(path: Path, content: str) -> Path:
    """Write and sync a same-directory replacement with live-file metadata.

    Keeping the temporary file on the destination filesystem makes the later
    os.replace operation atomic.
    """
    try:
        metadata = path.stat()
    except OSError as exc:
        raise VmailError(f"cannot stat {path}: {exc}") from exc
    fd, name = tempfile.mkstemp(prefix=f".{path.name}.vmailctl.", dir=path.parent)
    temp_path = Path(name)
    try:
        os.fchmod(fd, stat.S_IMODE(metadata.st_mode))
        os.fchown(fd, metadata.st_uid, metadata.st_gid)
        with os.fdopen(fd, "w", encoding="utf-8") as handle:
            fd = -1
            handle.write(content)
            handle.flush()
            os.fsync(handle.fileno())
    except (Exception, KeyboardInterrupt):
        if fd >= 0:
            os.close(fd)
        temp_path.unlink(missing_ok=True)
        raise
    return temp_path


def fsync_directory(path: Path) -> None:
    """Persist directory-entry changes after an atomic rename."""
    descriptor = os.open(path, os.O_RDONLY | os.O_DIRECTORY)
    try:
        os.fsync(descriptor)
    finally:
        os.close(descriptor)


def replace_text(path: Path, content: str) -> None:
    """Atomically replace a text file while preserving its ownership and mode."""
    temp_path = write_temp_like(path, content)
    try:
        os.replace(temp_path, path)
        fsync_directory(path.parent)
    finally:
        temp_path.unlink(missing_ok=True)


@dataclass
class PreparedMap:
    """A compiled Postfix source/database pair awaiting atomic renames."""

    source: Path
    database: Path

    def cleanup(self) -> None:
        self.source.unlink(missing_ok=True)
        self.database.unlink(missing_ok=True)


def prepare_map(settings: Settings, path: Path, content: str) -> PreparedMap:
    """Compile a candidate Postfix map without touching the live pair."""
    temp_source = write_temp_like(path, content)
    temp_database = Path(str(temp_source) + ".db")
    try:
        run([settings.postmap, temp_source])
        if not temp_database.is_file():
            raise VmailError(f"postmap did not create {temp_database}")
        existing_database = Path(str(path) + ".db")
        if existing_database.exists():
            metadata = existing_database.stat()
            os.chmod(temp_database, stat.S_IMODE(metadata.st_mode))
            os.chown(temp_database, metadata.st_uid, metadata.st_gid)
        return PreparedMap(temp_source, temp_database)
    except (Exception, KeyboardInterrupt):
        temp_source.unlink(missing_ok=True)
        temp_database.unlink(missing_ok=True)
        raise


def commit_map(path: Path, prepared: PreparedMap) -> None:
    """Install a prepared source and database pair using atomic renames."""
    database = Path(str(path) + ".db")
    try:
        os.replace(prepared.source, path)
        os.replace(prepared.database, database)
        fsync_directory(path.parent)
    finally:
        prepared.cleanup()


@contextmanager
def exclusive_lock(path: Path):
    """Serialize operations with a root-only, symlink-resistant advisory lock."""
    validate_secure_directory(
        path.parent,
        f"lock directory {path.parent}",
        private=False,
        allow_sticky_world_writable=True,
    )
    flags = os.O_RDWR | os.O_CREAT | getattr(os, "O_CLOEXEC", 0)
    if hasattr(os, "O_NOFOLLOW"):
        flags |= os.O_NOFOLLOW
    try:
        descriptor = os.open(path, flags, 0o600)
    except OSError as exc:
        raise VmailError(f"cannot open lock file {path}: {exc}") from exc
    try:
        metadata = os.fstat(descriptor)
        validate_regular_metadata(
            metadata,
            f"lock file {path}",
            private=True,
        )
        os.fchmod(descriptor, 0o600)
        with os.fdopen(descriptor, "w") as handle:
            descriptor = -1
            fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
            yield
            fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
    finally:
        if descriptor >= 0:
            os.close(descriptor)


class BackupSet:
    """Root-only snapshots used to roll a transaction back."""

    def __init__(self, settings: Settings, action: str, paths: Iterable[Path]):
        ensure_secure_backup_directory(settings.backup_dir)
        timestamp = dt.datetime.now(dt.timezone.utc).strftime("%Y%m%dT%H%M%SZ")
        safe_action = re.sub(r"[^a-zA-Z0-9_.-]+", "-", action).strip("-")
        destination = settings.backup_dir / f"{timestamp}-{safe_action}-{os.getpid()}"
        destination.mkdir(mode=0o700)
        self.destination = destination
        self.records: list[tuple[Path, Path | None]] = []
        for original in paths:
            backup_name = original.as_posix().lstrip("/").replace("/", "__")
            backup = destination / backup_name
            if original.exists():
                shutil.copy2(original, backup)
                self.records.append((original, backup))
            else:
                self.records.append((original, None))
        manifest = destination / "manifest.txt"
        manifest.write_text(
            f"created_utc={timestamp}\naction={action}\n"
            + "".join(f"path={original}\n" for original, _ in self.records),
            encoding="utf-8",
        )
        os.chmod(manifest, 0o600)

    def restore(self) -> None:
        """Restore every recorded path, continuing after individual failures."""
        errors: list[str] = []
        for original, backup in self.records:
            try:
                if backup is None:
                    if original.exists():
                        quarantine = self.destination / f"created-{original.name}"
                        os.replace(original, quarantine)
                else:
                    temp = write_temp_like(original, backup.read_text(encoding="utf-8")) if not original.name.endswith(".db") else None
                    if temp is not None:
                        os.replace(temp, original)
                    else:
                        restore_temp = original.parent / f".{original.name}.restore-{os.getpid()}"
                        shutil.copy2(backup, restore_temp)
                        os.replace(restore_temp, original)
                    fsync_directory(original.parent)
            except BaseException as exc:  # Best effort across all transaction files.
                errors.append(f"{original}: {exc}")
        if errors:
            raise VmailError("rollback was incomplete: " + "; ".join(errors))


def make_maildir(settings: Settings, localpart: str) -> Path:
    home = settings.mailbox_home(localpart)
    if home.exists():
        raise VmailError(f"mailbox home already exists: {home}")
    for directory in (home, home / "Maildir", home / "Maildir" / "cur", home / "Maildir" / "new", home / "Maildir" / "tmp"):
        directory.mkdir(mode=settings.directory_mode)
        os.chown(directory, settings.uid, settings.gid)
        os.chmod(directory, settings.directory_mode)
    return home


def normalize_new_maildir_permissions(settings: Settings, home: Path) -> None:
    for item in [home, *home.rglob("*")]:
        if item.is_symlink():
            raise VmailError(f"unexpected symlink in newly created mailbox: {item}")
        os.chown(item, settings.uid, settings.gid)
        if item.is_dir():
            os.chmod(item, settings.directory_mode)
        elif item.is_file():
            os.chmod(item, 0o600)


def quarantine_failed_mailhome(home: Path, backup: BackupSet) -> None:
    """Preserve a partially created Maildir for inspection instead of deleting it."""
    if not home.exists():
        return
    destination = backup.destination / "rolled-back-mailhome"
    if destination.exists():
        destination = backup.destination / f"rolled-back-mailhome-{os.getpid()}"
    shutil.move(os.fspath(home), os.fspath(destination))


def verify_postfix(settings: Settings) -> None:
    run([settings.postfix, "check"])


def verify_dovecot(settings: Settings) -> None:
    run([settings.doveconf, "-n"])


def query_map(settings: Settings, path: Path, key: str) -> str:
    result = run([settings.postmap, "-q", key, f"hash:{path}"])
    return result.stdout.strip()


def new_passwd_line(settings: Settings, localpart: str, address: str, password_hash: str) -> str:
    home = settings.mailbox_home(localpart)
    return (
        f"{address}:{password_hash}:{settings.uid}:{settings.gid}::{home}::"
        "userdb_mail_path=~/Maildir"
    )


def load_live_state(settings: Settings):
    passwd_text = read_text(settings.dovecot_users)
    mailbox_text = read_text(settings.postfix_mailboxes)
    alias_text = read_text(settings.postfix_aliases)
    passwd = parse_passwd(passwd_text, os.fspath(settings.dovecot_users))
    mailboxes = parse_map(mailbox_text, os.fspath(settings.postfix_mailboxes))
    aliases = parse_map(alias_text, os.fspath(settings.postfix_aliases))
    return passwd_text, mailbox_text, alias_text, passwd, mailboxes, aliases


def ensure_address_available(address: str, passwd, mailboxes, aliases) -> None:
    locations = []
    if address in passwd:
        locations.append("Dovecot users")
    if address in mailboxes:
        locations.append("Postfix mailboxes")
    if address in aliases:
        locations.append("Postfix aliases")
    if locations:
        raise VmailError(f"{address} already exists in {', '.join(locations)}")


def mailbox_add(settings: Settings, value: str, dry_run: bool, use_generated_password: bool) -> None:
    """Create a login, delivery map and Maildir as one recoverable transaction."""
    localpart, address = normalize_localpart(value, settings.domain)
    state = load_live_state(settings)
    ensure_address_available(address, state[3], state[4], state[5])
    if dry_run:
        print(f"DRY RUN: would create mailbox {address}")
        print(f"DRY RUN: would create {settings.mailbox_home(localpart) / 'Maildir'}")
        print(f"DRY RUN: would create and subscribe folders: {', '.join(settings.folders)}")
        return

    password = generate_password() if use_generated_password else prompt_password()
    password_hash = hash_password(settings, password)
    backup: BackupSet | None = None
    home: Path | None = None
    with exclusive_lock(settings.lock_file):
        passwd_text, mailbox_text, _, passwd, mailboxes, aliases = load_live_state(settings)
        ensure_address_available(address, passwd, mailboxes, aliases)
        backup = BackupSet(
            settings,
            f"mailbox-add-{address}",
            (
                settings.dovecot_users,
                settings.postfix_mailboxes,
                Path(str(settings.postfix_mailboxes) + ".db"),
            ),
        )
        new_passwd = append_line(
            passwd_text,
            new_passwd_line(settings, localpart, address, password_hash),
        )
        new_mailboxes = append_line(
            mailbox_text,
            f"{address}\t{settings.mailbox_target(localpart)}",
        )
        prepared = prepare_map(settings, settings.postfix_mailboxes, new_mailboxes)
        try:
            # From this point, any failure or terminal interrupt must restore
            # the files and quarantine a Maildir that was already created.
            home = settings.mailbox_home(localpart)
            home = make_maildir(settings, localpart)
            replace_text(settings.dovecot_users, new_passwd)
            commit_map(settings.postfix_mailboxes, prepared)
            verify_dovecot(settings)
            verify_postfix(settings)
            result = run([settings.doveadm, "user", address])
            if result.returncode != 0:
                raise VmailError(f"Dovecot cannot resolve {address}")
            expected = settings.mailbox_target(localpart)
            if query_map(settings, settings.postfix_mailboxes, address) != expected:
                raise VmailError(f"Postfix mailbox lookup failed for {address}")
            if settings.folders:
                run([settings.doveadm, "mailbox", "create", "-u", address, "-s", *settings.folders])
            normalize_new_maildir_permissions(settings, home)
        except (Exception, KeyboardInterrupt) as exc:
            prepared.cleanup()
            rollback_error = None
            try:
                backup.restore()
                if home is not None:
                    quarantine_failed_mailhome(home, backup)
            except BaseException as rollback_exc:
                rollback_error = rollback_exc
            if rollback_error:
                raise VmailError(f"{exc}; additionally, rollback failed: {rollback_error}") from exc
            if isinstance(exc, KeyboardInterrupt):
                raise
            raise VmailError(f"{exc}; changes were rolled back to {backup.destination}") from exc

    print(f"Created mailbox {address}")
    print(f"Backup: {backup.destination}")
    if use_generated_password:
        print(f"Generated password (shown once): {password}")


def mailbox_password(settings: Settings, value: str, dry_run: bool, use_generated_password: bool) -> None:
    """Change one password hash with validation and rollback."""
    _, address = normalize_localpart(value, settings.domain)
    state = load_live_state(settings)
    if address not in state[3]:
        raise VmailError(f"unknown Dovecot mailbox: {address}")
    if dry_run:
        print(f"DRY RUN: would change the password for {address}")
        return

    password = generate_password() if use_generated_password else prompt_password()
    password_hash = hash_password(settings, password)
    backup: BackupSet | None = None
    with exclusive_lock(settings.lock_file):
        passwd_text = read_text(settings.dovecot_users)
        entries = parse_passwd(passwd_text, os.fspath(settings.dovecot_users))
        if address not in entries:
            raise VmailError(f"unknown Dovecot mailbox: {address}")
        output: list[str] = []
        for raw in passwd_text.splitlines():
            if raw.strip() and not raw.lstrip().startswith("#") and raw.split(":", 1)[0].lower() == address:
                fields = raw.split(":")
                fields[1] = password_hash
                raw = ":".join(fields)
            output.append(raw)
        new_text = "\n".join(output) + ("\n" if passwd_text.endswith("\n") or output else "")
        backup = BackupSet(settings, f"mailbox-passwd-{address}", (settings.dovecot_users,))
        try:
            replace_text(settings.dovecot_users, new_text)
            verify_dovecot(settings)
            run([settings.doveadm, "user", address])
        except (Exception, KeyboardInterrupt) as exc:
            try:
                backup.restore()
            except BaseException as rollback_exc:
                raise VmailError(f"{exc}; additionally, rollback failed: {rollback_exc}") from exc
            if isinstance(exc, KeyboardInterrupt):
                raise
            raise VmailError(f"{exc}; change was rolled back to {backup.destination}") from exc

    print(f"Changed password for {address}")
    print(f"Backup: {backup.destination}")
    if use_generated_password:
        print(f"Generated password (shown once): {password}")


def mailbox_list(settings: Settings) -> None:
    passwd = parse_passwd(read_text(settings.dovecot_users), os.fspath(settings.dovecot_users))
    for address in sorted(passwd):
        print(address)
    print(f"{len(passwd)} mailbox login(s)")


def alias_add(settings: Settings, alias_value: str, target_value: str, dry_run: bool) -> None:
    """Add and verify one Postfix alias as a recoverable transaction."""
    _, alias_address = normalize_localpart(alias_value, settings.domain)
    target_localpart, target_address = normalize_localpart(target_value, settings.domain)
    state = load_live_state(settings)
    ensure_address_available(alias_address, state[3], state[4], state[5])
    expected_target = settings.mailbox_target(target_localpart)
    if target_address not in state[4] or state[4][target_address][1] != expected_target:
        raise VmailError(f"alias target is not a real mailbox: {target_address}")
    if dry_run:
        print(f"DRY RUN: would create alias {alias_address} -> {target_address}")
        return

    backup: BackupSet | None = None
    with exclusive_lock(settings.lock_file):
        _, _, alias_text, passwd, mailboxes, aliases = load_live_state(settings)
        ensure_address_available(alias_address, passwd, mailboxes, aliases)
        if target_address not in mailboxes or mailboxes[target_address][1] != expected_target:
            raise VmailError(f"alias target is not a real mailbox: {target_address}")
        backup = BackupSet(
            settings,
            f"alias-add-{alias_address}",
            (settings.postfix_aliases, Path(str(settings.postfix_aliases) + ".db")),
        )
        new_aliases = append_line(alias_text, f"{alias_address}\t{target_address}")
        prepared = prepare_map(settings, settings.postfix_aliases, new_aliases)
        try:
            commit_map(settings.postfix_aliases, prepared)
            verify_postfix(settings)
            if query_map(settings, settings.postfix_aliases, alias_address) != target_address:
                raise VmailError(f"Postfix alias lookup failed for {alias_address}")
        except (Exception, KeyboardInterrupt) as exc:
            prepared.cleanup()
            try:
                backup.restore()
            except BaseException as rollback_exc:
                raise VmailError(f"{exc}; additionally, rollback failed: {rollback_exc}") from exc
            if isinstance(exc, KeyboardInterrupt):
                raise
            raise VmailError(f"{exc}; change was rolled back to {backup.destination}") from exc

    print(f"Created alias {alias_address} -> {target_address}")
    print(f"Backup: {backup.destination}")


def alias_list(settings: Settings) -> None:
    aliases = parse_map(read_text(settings.postfix_aliases), os.fspath(settings.postfix_aliases))
    for address in sorted(aliases):
        print(f"{address} -> {aliases[address][1]}")
    print(f"{len(aliases)} alias(es)")


def audit(settings: Settings) -> int:
    """Cross-check files, Maildirs, service configuration and live lookups."""
    errors: list[str] = []
    warnings: list[str] = []
    infos: list[str] = []
    try:
        _, _, _, passwd, mailboxes, aliases = load_live_state(settings)
    except VmailError as exc:
        print(f"ERROR: {exc}")
        return 2

    for path in (settings.dovecot_users, settings.postfix_mailboxes, settings.postfix_aliases):
        try:
            metadata = path.stat()
            mode = stat.S_IMODE(metadata.st_mode)
            if mode & 0o022:
                errors.append(f"{path} is group/world writable ({mode:04o})")
            if path == settings.dovecot_users and mode & 0o007:
                errors.append(f"{path} is accessible to other users ({mode:04o})")
        except OSError as exc:
            errors.append(f"cannot stat {path}: {exc}")

    collisions = sorted(set(mailboxes) & set(aliases))
    for address in collisions:
        alias_target = aliases[address][1].lower()
        direct_target = mailboxes[address][1]
        if alias_target in mailboxes and mailboxes[alias_target][1] == direct_target:
            infos.append(
                f"{address} has consistent alias and direct-delivery compatibility mappings"
            )
        else:
            errors.append(
                f"{address} exists in both mailbox and alias maps with conflicting destinations"
            )

    for address, entry in passwd.items():
        try:
            localpart, _ = normalize_localpart(address, settings.domain)
        except VmailError as exc:
            errors.append(f"{settings.dovecot_users}:{entry.line_number}: {exc}")
            continue
        fields = entry.fields
        if fields[2] != str(settings.uid) or fields[3] != str(settings.gid):
            errors.append(f"{address} has unexpected uid/gid {fields[2]}:{fields[3]}")
        expected_home = os.fspath(settings.mailbox_home(localpart))
        if fields[5] != expected_home:
            errors.append(f"{address} has unexpected home {fields[5]}")
        if fields[7] != "userdb_mail_path=~/Maildir":
            errors.append(f"{address} has unexpected Dovecot extra fields")
        expected_target = settings.mailbox_target(localpart)
        if address not in mailboxes:
            errors.append(f"{address} is missing from the Postfix mailbox map")
        elif mailboxes[address][1] != expected_target:
            errors.append(f"{address} has unexpected Postfix target {mailboxes[address][1]}")
        home = settings.mailbox_home(localpart)
        maildir = home / "Maildir"
        if not maildir.is_dir():
            errors.append(f"{address} has no Maildir at {maildir}")
        else:
            metadata = maildir.stat()
            if metadata.st_uid != settings.uid or metadata.st_gid != settings.gid:
                warnings.append(
                    f"{maildir} is owned by {metadata.st_uid}:{metadata.st_gid}, "
                    f"expected {settings.uid}:{settings.gid}"
                )

    for address, (_, target) in aliases.items():
        if target.lower() == address:
            errors.append(f"{address} is a direct alias loop")

    for source in (settings.postfix_mailboxes, settings.postfix_aliases):
        database = Path(str(source) + ".db")
        if not database.is_file():
            errors.append(f"compiled Postfix map is missing: {database}")
        elif database.stat().st_mtime < source.stat().st_mtime:
            warnings.append(f"compiled Postfix map is older than its source: {database}")

    for label, command in (
        ("Postfix configuration", [settings.postfix, "check"]),
        ("Dovecot configuration", [settings.doveconf, "-n"]),
    ):
        result = run(command, check=False)
        if result.returncode != 0:
            errors.append(f"{label} validation failed")

    for address, (_, target) in mailboxes.items():
        try:
            if query_map(settings, settings.postfix_mailboxes, address) != target:
                errors.append(f"Postfix mailbox lookup mismatch for {address}")
        except VmailError as exc:
            errors.append(f"Postfix mailbox lookup failed for {address}: {exc}")
    for address, (_, target) in aliases.items():
        try:
            if query_map(settings, settings.postfix_aliases, address) != target:
                errors.append(f"Postfix alias lookup mismatch for {address}")
        except VmailError as exc:
            errors.append(f"Postfix alias lookup failed for {address}: {exc}")
    for address in passwd:
        result = run([settings.doveadm, "user", address], check=False)
        if result.returncode != 0:
            errors.append(f"Dovecot user lookup failed for {address}")

    delivery_only = len(set(mailboxes) - set(passwd))
    print(
        f"Checked {len(passwd)} mailbox login(s), {len(mailboxes)} delivery mapping(s), "
        f"{len(aliases)} alias(es)"
    )
    if delivery_only:
        print(f"INFO: {delivery_only} delivery mapping(s) intentionally have no mailbox login")
    for information in infos:
        print(f"INFO: {information}")
    for warning in warnings:
        print(f"WARNING: {warning}")
    for error in errors:
        print(f"ERROR: {error}")
    if errors:
        print(f"Audit failed with {len(errors)} error(s) and {len(warnings)} warning(s)")
        return 2
    print(f"Audit passed with {len(warnings)} warning(s)")
    return 0


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="vmailctl",
        description="Manage flat-file Postfix/Dovecot virtual mailboxes safely.",
    )
    parser.add_argument("--config", type=Path, default=DEFAULT_CONFIG)
    parser.add_argument("--dry-run", action="store_true", help="show a planned change without writing")
    parser.add_argument("--version", action="version", version=f"%(prog)s {VERSION}")
    commands = parser.add_subparsers(dest="command", required=True)
    commands.add_parser("audit", help="validate account files, maps, paths and lookups")

    mailbox = commands.add_parser("mailbox", help="manage mailbox logins")
    mailbox_commands = mailbox.add_subparsers(dest="mailbox_command", required=True)
    mailbox_commands.add_parser("list", help="list mailbox logins")
    mailbox_add_parser = mailbox_commands.add_parser("add", help="create a mailbox")
    mailbox_add_parser.add_argument("address")
    mailbox_add_parser.add_argument(
        "--generate-password",
        action="store_true",
        help="generate a strong password and display it once after success",
    )
    mailbox_password_parser = mailbox_commands.add_parser("passwd", help="change a mailbox password")
    mailbox_password_parser.add_argument("address")
    mailbox_password_parser.add_argument(
        "--generate-password",
        action="store_true",
        help="generate a strong password and display it once after success",
    )

    alias = commands.add_parser("alias", help="manage aliases")
    alias_commands = alias.add_subparsers(dest="alias_command", required=True)
    alias_commands.add_parser("list", help="list aliases")
    alias_add_parser = alias_commands.add_parser("add", help="create an alias to a real mailbox")
    alias_add_parser.add_argument("alias")
    alias_add_parser.add_argument("target")
    return parser


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    try:
        require_root()
        settings = Settings.load(args.config)
        modifying = (
            not args.dry_run
            and (
                args.command == "mailbox"
                and args.mailbox_command in {"add", "passwd"}
                or args.command == "alias"
                and args.alias_command == "add"
            )
        )
        validate_runtime_security(settings, modifying=modifying)
        if args.command == "audit":
            if args.dry_run:
                raise VmailError("--dry-run is not meaningful with audit")
            with exclusive_lock(settings.lock_file):
                return audit(settings)
        if args.command == "mailbox":
            if args.mailbox_command == "list":
                if args.dry_run:
                    raise VmailError("--dry-run is not meaningful with mailbox list")
                mailbox_list(settings)
            elif args.mailbox_command == "add":
                mailbox_add(settings, args.address, args.dry_run, args.generate_password)
            elif args.mailbox_command == "passwd":
                mailbox_password(settings, args.address, args.dry_run, args.generate_password)
        elif args.command == "alias":
            if args.alias_command == "list":
                if args.dry_run:
                    raise VmailError("--dry-run is not meaningful with alias list")
                alias_list(settings)
            elif args.alias_command == "add":
                alias_add(settings, args.alias, args.target, args.dry_run)
        return 0
    except VmailError as exc:
        print(f"vmailctl: {exc}", file=sys.stderr)
        return 1
    except KeyboardInterrupt:
        print("vmailctl: cancelled", file=sys.stderr)
        return 130


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