From 58a5ebdc2f81e42e4fe9f538ebcf7bd7bdcbffb7 Mon Sep 17 00:00:00 2001
From: Magnus Hagander <magnus@hagander.net>
Date: Wed, 7 Jan 2026 17:42:09 +0100
Subject: [PATCH] Add small tools for testing email sending and parsing

These tools can be used for any app that uses the "pgweb style
mailqueue" module, including for example pgeu-system.
---
 tools/email/.gitignore         |  1 +
 tools/email/README.md          | 23 ++++++++++++
 tools/email/config.yaml.sample | 11 ++++++
 tools/email/parse_email.py     | 50 +++++++++++++++++++++++++
 tools/email/send_email.py      | 68 ++++++++++++++++++++++++++++++++++
 5 files changed, 153 insertions(+)
 create mode 100644 tools/email/.gitignore
 create mode 100644 tools/email/README.md
 create mode 100644 tools/email/config.yaml.sample
 create mode 100755 tools/email/parse_email.py
 create mode 100755 tools/email/send_email.py

diff --git a/tools/email/.gitignore b/tools/email/.gitignore
new file mode 100644
index 00000000..5b6b0720
--- /dev/null
+++ b/tools/email/.gitignore
@@ -0,0 +1 @@
+config.yaml
diff --git a/tools/email/README.md b/tools/email/README.md
new file mode 100644
index 00000000..130627ad
--- /dev/null
+++ b/tools/email/README.md
@@ -0,0 +1,23 @@
+# email tools
+
+This directory holds a few trivial email testing tools. They work on emails
+that are in the `mailqueue` app, so they first have to be generated (with pgweb
+that's typically done by approving news or using the `news_send_email` command),
+and then referenced by their id number. They are used to test formats and markups.
+
+## parse_email.py
+
+This tool will simply parse and print the MIME structure of the email in question.
+
+## send_email.py
+
+This tools will take the email and send it out using SMTP/AUTH (hardcoded to always
+have STARTTLS) according to the settings in `config.yaml` for end-to-end testing.
+
+Note that emails are *not* removed from the queue when sent this way! This way they
+can be sent to multiple addresses for testing.
+
+## config.yaml
+
+Used for both tools to find their database, and for `send_email.py` to know how to
+connect to the server. See the `config.yaml.sample` file for example/docs.
diff --git a/tools/email/config.yaml.sample b/tools/email/config.yaml.sample
new file mode 100644
index 00000000..08a3ca90
--- /dev/null
+++ b/tools/email/config.yaml.sample
@@ -0,0 +1,11 @@
+---
+mail:
+  server: smtp.example.com     # your server
+  port: 587
+  user: someone@example.com    # to log in with
+  sender: someone@example.com  # Email to use as envelope sender. Usually, but not always, matches user.
+  password: # Can be either a string here
+    smtp: someone@example.com  # Or a lookup of a key/value pair in the secrets store
+db:                            # List of databases that we can connect to
+  pgweb: dbname=pgweb          # Each in psyocpg2 dsn format
+  pgeu: dbname=postgresqleu
diff --git a/tools/email/parse_email.py b/tools/email/parse_email.py
new file mode 100755
index 00000000..5a1ab270
--- /dev/null
+++ b/tools/email/parse_email.py
@@ -0,0 +1,50 @@
+#!/usr/bin/env python3
+
+import argparse
+import email
+import email.policy
+import psycopg2
+import sys
+import yaml
+
+
+def print_message(msg, level=0):
+    def _out(t):
+        print("{}{}".format('  ' * level, t))
+
+    _out(msg.get_content_type())
+    if msg.is_multipart():
+        for p in msg.iter_parts():
+            print_message(p, level + 1)
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser("Print email MIME structure")
+    parser.add_argument('db', help='Name of database (looked up in config.yaml) t connect to')
+    parser.add_argument('id', type=int, help='ID of email entry to send')
+
+    args = parser.parse_args()
+
+    with open('config.yaml') as f:
+        config = yaml.load(f, Loader=yaml.SafeLoader)
+
+    if args.db not in config['db']:
+        print("Non-existing db specified")
+        sys.exit(1)
+
+    # Connect to db and get message
+    dbconn = psycopg2.connect(config['db'][args.db])
+    curs = dbconn.cursor()
+    curs.execute("SELECT fullmsg FROM mailqueue_queuedmail WHERE id=%(id)s", {
+        'id': args.id,
+    })
+    r = curs.fetchall()
+    dbconn.close()
+
+    if len(r) == 0:
+        print("Email not found")
+        sys.exit(1)
+
+    msg = email.message_from_string(r[0][0], policy=email.policy.default)
+
+    print_message(msg)
diff --git a/tools/email/send_email.py b/tools/email/send_email.py
new file mode 100755
index 00000000..68c02feb
--- /dev/null
+++ b/tools/email/send_email.py
@@ -0,0 +1,68 @@
+#!/usr/bin/env python3
+
+import argparse
+import psycopg2
+import smtplib
+import sys
+import yaml
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser("Email tester")
+    parser.add_argument('db', help='Name of database (looked up in config.yaml) t connect to')
+    parser.add_argument('id', type=int, help='ID of email entry to send')
+    parser.add_argument('recipient', help='Email address of recipient to send to')
+
+    args = parser.parse_args()
+
+    with open('config.yaml') as f:
+        config = yaml.load(f, Loader=yaml.SafeLoader)
+
+    if args.db not in config['db']:
+        print("Non-existing db specified")
+        sys.exit(1)
+
+    if isinstance(config['mail']['password'], str):
+        password = config['mail']['password']
+    elif isinstance(config['mail']['password'], dict):
+        import secretstorage
+        coll = secretstorage.get_default_collection(secretstorage.dbus_init())
+        if coll.is_locked():
+            coll.unlock()
+        r = list(coll.search_items(config['mail']['password']))
+        if len(r) == 0:
+            print("Could not find password in secret storage.")
+            sys.exit(1)
+        elif len(r) > 1:
+            print("Found more than one password, try again.")
+            sys.exit(1)
+        password = r[0].get_secret().decode()
+    else:
+        print("Invalid type for password in configuration")
+        sys.exit(1)
+
+    # Connect to db and get message
+    dbconn = psycopg2.connect(config['db'][args.db])
+    curs = dbconn.cursor()
+    curs.execute("SELECT fullmsg FROM mailqueue_queuedmail WHERE id=%(id)s", {
+        'id': args.id,
+    })
+    r = curs.fetchall()
+    dbconn.close()
+
+    if len(r) == 0:
+        print("Email not found")
+        sys.exit(1)
+
+    msg = r[0][0]
+
+    # Now do it!
+    smtp = smtplib.SMTP(host=config['mail']['server'], port=config['mail']['port'])
+    smtp.starttls()
+    smtp.login(user=config['mail']['user'], password=password)
+
+    smtp.sendmail(config['mail']['sender'], args.recipient, msg)
+
+    smtp.quit()
+
+    print("Sent.")
-- 
2.39.5