mailman-sync/mailman_sync.py

#!/usr/bin/env python3

"""
Update Mailman 2 lists via a json API of the form:
{
  "LIST": {
    "real_name": "REAL_NAME",
    "moderator": ["ADDRESS", ...],
    "subject_prefix": "PREFIX",
    "reply_to_address": "REPLY_TO_ADDRESS",
    "members": ["ADDRESS", ...]
  },
  ...
}
"""

import argparse
from dataclasses import dataclass
import os
from pathlib import Path
import secrets
import string
import subprocess
import tempfile

import requests

PASSWORD_CHARS = string.ascii_letters + string.digits + string.punctuation
PASSWORD_LEN = 18


@dataclass
class ListManager:
    mailman_bin: Path
    list_name: str
    dry_run: bool

    def _call_script(self, script: str, args: list[str], **kwargs):
        output = subprocess.run(
            [self.mailman_bin / script, *args],
            encoding="ascii",
            capture_output=True,
            check=True,
            **kwargs,
        )
        for line in output.stdout.splitlines():
            print(f"[{script} {self.list_name}] {line}")

    def newlist(self, urlhost: str, admin: str):
        password = "".join(secrets.choice(PASSWORD_CHARS) for i in range(PASSWORD_LEN))

        self._call_script(
            "newlist",
            [
                "--quiet",
                f"--urlhost={urlhost}",
                self.list_name,
                admin,
                password,
            ],
        )

    def config_list(
        self,
        real_name: str,
        moderator: list[str],
        subject_prefix: str,
        reply_to_address: str,
    ):
        config_changes = "\n".join(
            [
                "real_name = " + repr(real_name),
                "moderator = " + repr(moderator),
                "subject_prefix = " + repr(subject_prefix),
                "from_is_list = 1",
                "anonymous_list = 1",
                "first_strip_reply_to = 1",
                "reply_to_address = " + repr(reply_to_address),
                # Use explicit address for Reply-To
                "reply_goes_to_list = 2",
                # quiet member management
                "send_reminders = 0",
                "send_welcome_msg = 0",
                "send_goodbye_msg = 0",
                # ConcealSubscription | DontReceiveDuplicates
                "new_member_options = 272",
                "advertised = 0",
                "private_roster = 2",  # only admins can view the roster
                "default_member_moderation = 1",
                "generic_nonmember_action = 3",  # discard non-member emails
                "forward_auto_discards = 0",  # don't notify admin about discards
            ]
        )

        with tempfile.NamedTemporaryFile("w", suffix=".py") as config_file:
            config_file.write(config_changes)
            config_file.flush()

            args = [
                "--inputfile",
                config_file.name,
                self.list_name,
            ]
            if self.dry_run:
                args.append("--checkonly")

            self._call_script("config_list", args)

    def sync_members(self, members: list[str]):
        args = [
            "--welcome-msg=no",
            "--goodbye-msg=no",
            "--notifyadmin=no",
            "--file",
            "-",
            self.list_name,
        ]
        if self.dry_run:
            args.append("--no-change")

        members_data = "\n".join(members)
        self._call_script("sync_members", args, input=members_data)


def main(
    mailman_bin: Path,
    api: str,
    api_auth: str,
    list_suffix: str,
    dry_run: bool,
    urlhost: str,
    admin: str,
):
    r = requests.get(api, headers={"Authorization": api_auth})
    if not r.ok:
        print(f"Failed to get mailing list data from api: {r.status_code} {r.text}")
        return
    expected_lists = r.json()

    existing_lists = subprocess.run(
        [mailman_bin / "list_lists", "-b"],
        encoding="ascii",
        capture_output=True,
        check=True,
    ).stdout.split("\n")
    for name, props in expected_lists.items():
        list_name = name + list_suffix
        list_manager = ListManager(mailman_bin, list_name, dry_run)
        if list_name not in existing_lists:
            if dry_run:
                print(f"Skipping non-existing list {list_name} in dry run mode")
                continue
            else:
                list_manager.newlist(urlhost, admin)

        print(f"Configuring/syncing {list_name}...")
        list_manager.config_list(
            props["real_name"],
            props["moderator"],
            props["subject_prefix"],
            props["reply_to_address"],
        )
        list_manager.sync_members(props["members"])


def parse_arguments():
    argp = argparse.ArgumentParser(description=__doc__)
    argp.add_argument(
        "--bin",
        default="/usr/local/mailman",
        type=Path,
        help="Path to Mailman site admin scripts (default %(default)s)",
    )
    argp.add_argument("--api", required=True, help="API endpoint to retrieve JSON from")
    argp.add_argument("--list-suffix", help="Suffix for mailing lists")
    argp.add_argument("--urlhost", help="Urlhost to use when creating new lists")
    argp.add_argument(
        "--admin", help="Admin email address to use when creating new lists"
    )
    argp.add_argument(
        "-n",
        "--dry-run",
        action="store_true",
        help="Don't make changes, just print what would happen",
    )
    return argp.parse_args()


if __name__ == "__main__":
    args = parse_arguments()

    if "API_AUTH" in os.environ:
        api_auth = os.environ.get("API_AUTH")
    else:
        print("Missing API_AUTH environment variable")
        exit(-1)

    try:
        main(
            args.bin,
            args.api,
            api_auth,
            args.list_suffix,
            args.dry_run,
            args.urlhost,
            args.admin,
        )
    except subprocess.CalledProcessError as e:
        print(e.stderr)
        raise