mailman-sync/mailman_sync.py

#!/usr/bin/env python3

"""
Update Mailman 2 lists via a json API of the form {"LIST": ["ADDRESS", ...]}
"""

import argparse
import os
from pathlib import Path
import subprocess
import tempfile

import requests


def config_list(mailman_bin: Path, mailing_list: str):
    config_changes = """
# TODO: set real_name, moderator, subject prefix, and reply-to address
from_is_list = 1
anonymous_list = 1
first_strip_reply_to = 1
reply_goes_to_list = 2

# quiet member management
send_reminders = 0
send_welcome_msg = 0
send_goodbye_msg = 0

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()
        output = subprocess.run(
            [
                mailman_bin / "config_list",
                "--verbose",
                "--inputfile",
                config_file.name,
                mailing_list,
            ],
            encoding="ascii",
            stdout=PIPE,
            stderr=STDOUT,
            check=True,
        )
        for line in output.stdout.splitlines():
            print(f"[Configuring {mailing_list}] {line}")


def sync_members(
    mailman_bin: Path, mailing_list: str, members: list[str], dry_run: bool
):
    command = [
        mailman_bin / "sync_members",
        "--welcome-msg=no",
        "--goodbye-msg=no",
        "--notifyadmin=no",
        "--file",
        "-",
        mailing_list,
    ]
    if dry_run:
        command.append("--no-change")

    members_data = "\n".join(members)
    output = subprocess.run(
        command,
        input=members_data,
        encoding="ascii",
        capture_output=True,
        check=True,
    )
    for line in output.stdout.splitlines():
        print(f"[Syncing {mailing_list}] {line}")


def main(mailman_bin: Path, api: str, api_auth: str, list_suffix: str, dry_run: bool):
    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

    existing_lists = subprocess.run(
        [mailman_bin / "list_lists", "-b"],
        encoding="ascii",
        capture_output=True,
        check=True,
    ).stdout.split("\n")
    certification_lists = r.json()
    for name, members in certification_lists.items():
        list_name = name + list_suffix
        if list_name in existing_lists:
            print(f"Configuring/syncing {list_name}...")
            config_list(mailman_bin, list_name)
            sync_members(mailman_bin, list_name, members, dry_run)
        else:
            print(f"Skipping {list_name}, as it does not exist in Mailman")


if __name__ == "__main__":
    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(
        "-n",
        "--dry-run",
        action="store_true",
        help="Don't make changes, just print what would happen",
    )
    args = argp.parse_args()

    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)
    except subprocess.CalledProcessError as e:
        print(e.stderr)
        raise