Add emailhost argument, set it in newlist and config_list

mailman_sync.py

@@ -1,90 +1,149 @@
 #!/usr/bin/env python3
 
 """
-Update Mailman 2 lists via a json API of the form {"LIST": ["ADDRESS", ...]}
+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
 
-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
+@dataclass
+class ListManager:
+    mailman_bin: Path
+    list_name: str
+    dry_run: bool
 
-new_member_options = 272
-advertised = 0
-private_roster = 2  # only admins can view the roster
-default_member_moderation = 1
+    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}")
 
-generic_nonmember_action = 3  # discard non-member emails
-forward_auto_discards = 0  # don't notify admin about discards
-"""
+    def newlist(self, urlhost: str, emailhost: str, admin: str):
+        password = "".join(secrets.choice(PASSWORD_CHARS) for i in range(PASSWORD_LEN))
+
+        self._call_script(
+            "newlist",
+            [
+                "--quiet",
+                f"--urlhost={urlhost}",
+                f"--emailhost={emailhost}",
+                self.list_name,
+                admin,
+                password,
+            ],
+        )
+
+    def config_list(
+        self,
+        emailhost: str,
+        real_name: str,
+        moderator: list[str],
+        subject_prefix: str,
+        reply_to_address: str,
+    ):
+        config_changes = "\n".join(
+            [
+                # have to modify mlist directly to allow removing suffix
+                "mlist.real_name = " + repr(real_name),
+                "host_name = " + repr(emailhost),
+                "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",
+                "subscribe_policy = 3",  # require approval to join
+                "unsubscribe_policy = 1",  #  require approval to unsubscribe
+                "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
+                "archive_private = 1",
+            ]
+        )
 
         with tempfile.NamedTemporaryFile("w", suffix=".py") as config_file:
             config_file.write(config_changes)
             config_file.flush()
-        output = subprocess.run(
-            [
-                mailman_bin / "config_list",
+
+            args = [
                 "--inputfile",
                 config_file.name,
-                mailing_list,
-            ],
-            encoding="ascii",
-            capture_output=True,
-            check=True,
-        )
-        for line in output.stdout.splitlines():
-            print(f"[Configuring {mailing_list}] {line}")
+                self.list_name,
+            ]
+            if self.dry_run:
+                args.append("--checkonly")
 
+            self._call_script("config_list", args)
 
-def sync_members(
-    mailman_bin: Path, mailing_list: str, members: list[str], dry_run: bool
-):
-    command = [
-        mailman_bin / "sync_members",
+    def sync_members(self, members: list[str]):
+        args = [
             "--welcome-msg=no",
             "--goodbye-msg=no",
             "--notifyadmin=no",
             "--file",
             "-",
-        mailing_list,
+            self.list_name,
         ]
-    if dry_run:
-        command.append("--no-change")
+        if self.dry_run:
+            args.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}")
+        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):
+def main(
+    mailman_bin: Path,
+    api: str,
+    api_auth: str,
+    list_suffix: str,
+    dry_run: bool,
+    urlhost: str,
+    emailhost: 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"],
@@ -92,18 +151,28 @@ def main(mailman_bin: Path, api: str, api_auth: str, list_suffix: str, dry_run:
         capture_output=True,
         check=True,
     ).stdout.split("\n")
-    certification_lists = r.json()
-    for name, members in certification_lists.items():
+    for name, props in expected_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)
+        list_manager = ListManager(mailman_bin, list_name, dry_run)
+        if list_name.lower() not in existing_lists:
+            if dry_run:
+                print(f"Skipping non-existing list {list_name} in dry run mode")
+                continue
             else:
-            print(f"Skipping {list_name}, as it does not exist in Mailman")
+                list_manager.newlist(urlhost, emailhost, admin)
+
+        print(f"Configuring/syncing {list_name}...")
+        list_manager.config_list(
+            emailhost,
+            props["real_name"],
+            props["moderator"],
+            props["subject_prefix"],
+            props["reply_to_address"],
+        )
+        list_manager.sync_members(props["members"])
 
 
-if __name__ == "__main__":
+def parse_arguments():
     argp = argparse.ArgumentParser(description=__doc__)
     argp.add_argument(
         "--bin",
@@ -112,14 +181,29 @@ if __name__ == "__main__":
         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("--list-suffix", help="Suffix for mailing lists", default="")
+    argp.add_argument(
+        "--urlhost", help="Urlhost to use when creating new lists", required=True
+    )
+    argp.add_argument(
+        "--emailhost", help="Emailhost to use when creating new lists", required=True
+    )
+    argp.add_argument(
+        "--admin",
+        help="Admin email address to use when creating new lists",
+        required=True,
+    )
     argp.add_argument(
         "-n",
         "--dry-run",
         action="store_true",
         help="Don't make changes, just print what would happen",
     )
-    args = argp.parse_args()
+    return argp.parse_args()
+
+
+if __name__ == "__main__":
+    args = parse_arguments()
 
     if "API_AUTH" in os.environ:
         api_auth = os.environ.get("API_AUTH")
@@ -128,7 +212,16 @@ if __name__ == "__main__":
         exit(-1)
 
     try:
-        main(args.bin, args.api, api_auth, args.list_suffix, args.dry_run)
+        main(
+            args.bin,
+            args.api,
+            api_auth,
+            args.list_suffix,
+            args.dry_run,
+            args.urlhost,
+            args.emailhost,
+            args.admin,
+        )
     except subprocess.CalledProcessError as e:
         print(e.stderr)
         raise

pyproject.toml

@@ -0,0 +1 @@
+[tool.black]