diff --git a/helper-scripts/mailbox_cleaner.py b/helper-scripts/mailbox_cleaner.py new file mode 100755 index 0000000000..df8ff8b143 --- /dev/null +++ b/helper-scripts/mailbox_cleaner.py @@ -0,0 +1,140 @@ +#!/usr/bin/env python3 +""" +This script cleans up old messages from specified mailboxes (e.g., Trash, Junk) +in a mailcow environment. It can process a single user or all users, and it +supports dry-run mode. + +Ideally, this script should be run daily via cron. +""" + +import argparse +import logging +import os +import re +import subprocess +import sys + +logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') + +DEFAULT_DAYS_BACK: int = 30 +DEFAULT_MAILCOW_DIR: str = "/opt/mailcow-dockerized" +DEFAULT_MAILBOXES: list[str] = ["Trash", "Junk"] + + +def _run_doveadm_command(mailcow_dir: str, user: str | None, command: list[str]) -> str: + """ + Runs a doveadm command within the dovecot-mailcow container. + + Args: + mailcow_dir: The path to the mailcow-dockerized directory. + user: The email address of the user to run the command for, or None for all users. + command: The doveadm command to run as a list of strings. + + Returns: + The standard output of the command. + """ + command = ["docker", "compose", "--project-directory", mailcow_dir, "exec", "-T", "dovecot-mailcow", + "doveadm"] + command + if user: + command.extend(["-u", user]) + logging.debug(f"Executing command: {' '.join(command)}") + try: + result = subprocess.run(command, capture_output=True, text=True, check=True) + return result.stdout.strip() + except subprocess.CalledProcessError as e: + logging.error(f"Command execution failed: {' '.join(command)} (return code: {e.returncode})") + logging.error(f"Stderr: {e.stderr}") + logging.error(f"Stdout: {e.stdout}") + raise + + +def main() -> None: + """ + Main function to parse arguments and execute the cleanup process. + """ + parser = argparse.ArgumentParser( + description="Clean up old messages from specified mailboxes in a mailcow environment.", + formatter_class=argparse.ArgumentDefaultsHelpFormatter, + ) + group = parser.add_mutually_exclusive_group(required=True) + group.add_argument("--user", help="Email address of the single user to process.") + group.add_argument("--all", action="store_true", help="Process all users found via doveadm.") + parser.add_argument("--days-back", type=int, default=DEFAULT_DAYS_BACK, + help="Number of days back to consider for message deletion.") + parser.add_argument("--mailcow-directory", default=DEFAULT_MAILCOW_DIR, + help="Path to the mailcow-dockerized directory.") + parser.add_argument("--mailboxes", nargs='+', default=DEFAULT_MAILBOXES, + help="List of top-level mailboxes (and their subfolders) to process (e.g., Trash Junk).") + parser.add_argument("--debug", action="store_true", help="Enable debug logging.") + parser.add_argument("--dry-run", action="store_true", help="Perform a dry run without deleting anything.") + + args = parser.parse_args() + + if args.debug: + logging.getLogger().setLevel(logging.DEBUG) + + if not os.path.isdir(args.mailcow_directory): + raise FileNotFoundError( + f"mailcow directory '{args.mailcow_directory}' does not exist or is not a directory.") + # If --all is specified, get all users + if args.all: + doveadm_output = _run_doveadm_command(args.mailcow_directory, None, ["user", "*"]) + users = [line.strip() for line in doveadm_output.splitlines() if line.strip()] + # Otherwise, use the specified user + else: + try: + _run_doveadm_command(args.mailcow_directory, None, ["user", args.user]) + except subprocess.CalledProcessError: + logging.error(f"User '{args.user}' not found.") + sys.exit(1) + users = [args.user] + logging.info(f"Starting processing for {len(users)} users.") + logging.debug(f"Users to process: {', '.join(users)}") + # Iterate over each user + for user in users: + # Get all mailboxes for the current user + logging.info(f"Processing user: '{user}'.") + doveadm_output = _run_doveadm_command(args.mailcow_directory, user, ["mailbox", "list"]) + # get all user mailboxes, sorted in reverse order + mailboxes = sorted([line.strip() for line in doveadm_output.splitlines() if line.strip()], reverse=True) + logging.info(f"User '{user}' has {len(mailboxes)} mailboxes.") + logging.debug(f"Mailboxes for user '{user}': {', '.join(mailboxes)}") + for mailbox in mailboxes: + # Iterate over each mailbox + logging.debug(f"Processing mailbox '{mailbox}' for user '{user}'.") + # Check if the mailbox is a target mailbox + if not any(re.match(rf"{re.escape(tmb)}(/|$)", mailbox, re.IGNORECASE) for tmb in args.mailboxes): + logging.debug(f"Skipping mailbox '{mailbox}' for user '{user}' as it is not a target mailbox.") + continue + # Expunge old messages from the mailbox + logging.info( + f"Expunging messages older than {args.days_back} days from mailbox '{mailbox}' for user '{user}'.") + if args.dry_run: + logging.info(f"[DRY-RUN] Skipping expunge command for mailbox '{mailbox}' of user '{user}'.") + else: + # Run the expunge command + _run_doveadm_command(args.mailcow_directory, user, + ["expunge", "mailbox", mailbox, "savedbefore", f"{args.days_back}d"]) + # Check if the mailbox is a sub-mailbox + if "/" not in mailbox: + logging.debug( + f"Skipping deletion check for top-level mailbox '{mailbox}' to preserve standard folders.") + continue + # Check if the mailbox is empty + doveadm_output = _run_doveadm_command(args.mailcow_directory, user, ["mailbox", "status", "messages", mailbox]) + messages_count = int(doveadm_output.split("=")[1]) + logging.debug(f"Mailbox '{mailbox}' for user '{user}' contains {messages_count} messages.") + if messages_count > 0: + logging.info(f"Skipping deletion of mailbox '{mailbox}' for user '{user}' as it is not empty.") + continue + # Delete the mailbox if it's empty + logging.info(f"Deleting mailbox '{mailbox}' for user '{user}' (only if empty).") + if args.dry_run: + logging.info(f"[DRY-RUN] Skipping delete command for mailbox '{mailbox}' of user '{user}'.") + else: + # As a safeguard, -e flag prevents mailbox deletion in case it's not empty + _run_doveadm_command(args.mailcow_directory, user, ["mailbox", "delete", "-e", "-s", mailbox]) + + +if __name__ == "__main__": + main()