Cover mail DNS setup integration

contrib/testing/README.md

@@ -32,6 +32,9 @@ one-off files in `/tmp`:
   `store-diagnostics.py` preserve the remaining interactive and diagnostic
   bring-up tests. Destructive live helpers require an explicit opt-in
   environment variable.
+- `mail-dns-live-check.py` logs in through the real Web and IMAPS interfaces
+  and verifies that setup produced both an actionable SPF/DKIM/DMARC task and
+  a forwardable administrator message without private-key material.
 
 The complete mapping from the original one-off `/tmp` names to maintained
 scripts is recorded in `tmp-script-map.md`.

contrib/testing/mail-dns-live-check.py

@@ -0,0 +1,84 @@
+#!/usr/bin/env python3
+"""Verify the setup-created mail-DNS task and administrator message."""
+
+from __future__ import annotations
+
+import http.cookiejar
+import imaplib
+import json
+import os
+import ssl
+import urllib.request
+from email import policy
+from email.parser import BytesParser
+from urllib.parse import urlsplit
+
+
+base_url = os.environ.get("BONGO_TEST_WEB_URL", "http://127.0.0.1:8080").rstrip("/")
+user = os.environ.get("BONGO_TEST_USER", "admin")
+password = os.environ.get("BONGO_TEST_PASSWORD", "")
+domain = os.environ.get("BONGO_TEST_DOMAIN", "bongo.test")
+timeout = int(os.environ.get("BONGO_TEST_TIMEOUT", "90"))
+if not password:
+    raise SystemExit("set BONGO_TEST_PASSWORD for the disposable administrator")
+
+cookies = http.cookiejar.CookieJar()
+opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cookies))
+request = urllib.request.Request(
+    base_url + "/api/session",
+    data=json.dumps({"user": user, "password": password}).encode("utf-8"),
+    headers={"Content-Type": "application/json"}, method="POST")
+with opener.open(request, timeout=timeout) as response:
+    if response.status != 201:
+        raise SystemExit(f"Web login returned HTTP {response.status}")
+
+with opener.open(base_url + "/api/tasks?include_closed=1",
+                 timeout=timeout) as response:
+    tasks = json.load(response).get("tasks", [])
+matching = [task for task in tasks
+            if task.get("type") == "mail-dns" and task.get("key") == domain]
+if len(matching) != 1:
+    raise SystemExit(f"expected one mail-dns task for {domain}, got {len(matching)}")
+task = matching[0]
+if task.get("severity") != "warning" or not task.get("dismissible"):
+    raise SystemExit("mail-dns task does not have the expected actionable policy")
+details = task.get("details", {})
+if details.get("domain") != domain or not details.get("actions"):
+    raise SystemExit("mail-dns task has no actionable DNS details")
+if set(details.get("suggested", {})) != {"spf", "dkim", "dmarc"}:
+    raise SystemExit("mail-dns task lacks an SPF, DKIM, or DMARC suggestion")
+
+host = os.environ.get("BONGO_TEST_IMAP_HOST") or urlsplit(base_url).hostname
+context = ssl.create_default_context()
+if os.environ.get("BONGO_TEST_INSECURE_TLS") == "1":
+    context.check_hostname = False
+    context.verify_mode = ssl.CERT_NONE
+with imaplib.IMAP4_SSL(
+        host, int(os.environ.get("BONGO_TEST_IMAPS_PORT", "993")),
+        ssl_context=context, timeout=timeout) as client:
+    client.login(user, password)
+    status, _count = client.select("INBOX", readonly=True)
+    if status != "OK":
+        raise SystemExit("cannot select administrator INBOX")
+    status, identifiers = client.search(
+        None, "HEADER", "X-Bongo-Notification", "mail-dns")
+    if status != "OK" or not identifiers or not identifiers[0].split():
+        raise SystemExit("administrator mail-DNS notification was not found")
+    identifier = identifiers[0].split()[-1]
+    status, data = client.fetch(identifier, "(BODY.PEEK[])")
+    if status != "OK":
+        raise SystemExit("cannot read administrator mail-DNS notification")
+    payload = next((item[1] for item in data
+                    if isinstance(item, tuple) and isinstance(item[1], bytes)), b"")
+
+message = BytesParser(policy=policy.default).parsebytes(payload)
+if message.get("X-Bongo-Notification") != "mail-dns":
+    raise SystemExit("administrator message has the wrong notification marker")
+if domain not in str(message.get("To", "")):
+    raise SystemExit("administrator message has the wrong recipient domain")
+upper_payload = payload.upper()
+if (b"-----BEGIN PRIVATE KEY-----" in upper_payload or
+        b"-----BEGIN RSA PRIVATE KEY-----" in upper_payload or
+        b"-----BEGIN ENCRYPTED PRIVATE KEY-----" in upper_payload):
+    raise SystemExit("administrator message contains private-key material")
+print("PASS actionable mail-DNS task and administrator notification")

src/apps/config/tests/test_security_material.py

@@ -32,14 +32,18 @@ from unittest import mock
 
 SOURCE = Path(__file__).resolve().parents[4]
 sys.path.insert(0, str(SOURCE / "src" / "libs" / "python"))
+sys.path.insert(0, str(SOURCE / "src" / "www"))
 
 from bongo.configuration.model import Scenario  # noqa: E402
 from bongo.configuration.security_material import (  # noqa: E402
     DomainSecurityResult,
+    _private_key_data,
     dns_notification_message,
     inspect_mail_dns,
+    install_dns_tasks,
     provision_security_material,
 )
+from bongo.configuration.storage import ConfigurationError  # noqa: E402
 
 
 class MailSecurityMaterialTest(unittest.TestCase):
@@ -60,7 +64,8 @@ class MailSecurityMaterialTest(unittest.TestCase):
         self.assertIn(b"X-Bongo-Notification: mail-dns", message)
         self.assertIn(b"bongo._domainkey.example.test", message)
         self.assertIn(b"v=spf1 mx -all", message)
-        self.assertNotIn(b"PRIVATE KEY", message)
+        self.assertNotIn(b"-----BEGIN PRIVATE KEY-----", message)
+        self.assertNotIn(b"-----BEGIN RSA PRIVATE KEY-----", message)
 
     def test_dns_inspection_selects_only_mail_authentication_records(self):
         responses = [
@@ -79,6 +84,45 @@ class MailSecurityMaterialTest(unittest.TestCase):
         self.assertEqual(result["example.test"]["dmarc"],
                          ["v=DMARC1; p=none"])
 
+    def test_notification_is_empty_when_dns_has_no_remaining_work(self):
+        result = DomainSecurityResult(
+            "example.test", "bongo", {},
+            {"spf": "v=spf1 mx -all",
+             "dkim": "v=DKIM1; k=rsa; p=PUBLIC",
+             "dmarc": "v=DMARC1; p=none"}, ())
+        self.assertEqual(dns_notification_message((result,)), b"")
+
+    def test_private_key_import_refuses_symlinks_and_writable_keys(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            key = root / "dkim.key"
+            key.write_text("private")
+            key.chmod(0o620)
+            with self.assertRaisesRegex(ConfigurationError, "untrusted users"):
+                _private_key_data(key)
+            key.chmod(0o600)
+            link = root / "dkim-link.key"
+            link.symlink_to(key)
+            with self.assertRaisesRegex(ConfigurationError,
+                                        "not a regular file"):
+                _private_key_data(link)
+
+    def test_private_key_import_requires_rsa_2048_or_stronger(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            key = Path(temporary) / "dkim.key"
+            key.write_text("private")
+            key.chmod(0o600)
+            completed = SimpleNamespace(
+                stdout=b"Public Key Algorithm: EC\nMedium (256 bits)\n")
+            with mock.patch(
+                    "bongo.configuration.security_material.shutil.which",
+                    return_value="/usr/bin/certtool"), mock.patch(
+                    "bongo.configuration.security_material._run",
+                    return_value=completed):
+                with self.assertRaisesRegex(ConfigurationError,
+                                            "RSA key of at least"):
+                    _private_key_data(key)
+
     def test_provision_writes_private_key_dns_hint_and_dmarc_state(self):
         with tempfile.TemporaryDirectory() as temporary:
             root = Path(temporary)
@@ -126,6 +170,77 @@ class MailSecurityMaterialTest(unittest.TestCase):
             self.assertEqual(repeated[0].actions, results[0].actions)
             generate.assert_called_once_with("example.test", "bongo")
 
+    def test_matching_published_records_need_no_action(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            paths = SimpleNamespace(
+                ssl_dir=root / "ssl.d", dkim_dir=root / "dkim.d",
+                dmarc_dir=root / "dmarc.d")
+            scenario = Scenario(
+                hostname="mail.example.test", bind_address="192.0.2.5",
+                domains=["example.test"], admin_password="0123456789ab",
+                dkim_selector="bongo",
+                dkim_key_directory=str(paths.dkim_dir),
+                mail_dns={"example.test": {
+                    "spf": ["v=spf1 mx -all"],
+                    "dkim": ["v=DKIM1; k=rsa; p=PUBLIC"],
+                    "dmarc": ["v=DMARC1; p=reject; "
+                              "rua=mailto:dmarc-reports@example.test"],
+                    "errors": [],
+                }},
+            )
+            with mock.patch(
+                    "bongo.configuration.security_material.os.geteuid",
+                    return_value=0), mock.patch(
+                    "bongo.configuration.security_material.os.chown"), mock.patch(
+                    "bongo.configuration.security_material.os.fchown"), \
+                    mock.patch(
+                    "bongo.configuration.security_material._bongo_ids",
+                    return_value=(os.getuid(), os.getgid())), mock.patch(
+                    "bongo.configuration.security_material._generate_private_key",
+                    return_value=(b"PRIVATE KEY\n", "PUBLIC")):
+                results = provision_security_material(paths, scenario)
+            self.assertEqual(results[0].actions, ())
+            self.assertEqual(dns_notification_message(results), b"")
+
+    def test_dns_task_is_actionable_idempotent_and_private(self):
+        result = DomainSecurityResult(
+            "example.test", "bongo", {},
+            {"spf": "v=spf1 mx -all",
+             "dkim": "v=DKIM1; k=rsa; p=PUBLIC",
+             "dmarc": "v=DMARC1; p=none"},
+            ({"action": "publish", "name": "example.test",
+              "type": "TXT", "value": "v=spf1 mx -all"},))
+        with tempfile.TemporaryDirectory() as temporary:
+            database = Path(temporary) / "tasks" / "tasks.sqlite3"
+            configurations = {"web": {"task_database": str(database)}}
+            with mock.patch(
+                    "bongo.configuration.security_material._bongo_ids",
+                    return_value=(os.getuid(), os.getgid())):
+                install_dns_tasks(configurations, (result,))
+                install_dns_tasks(configurations, (result,))
+            from bongo_web.tasks import TaskStore
+            tasks = TaskStore(database).list("admin")
+            self.assertEqual(len(tasks), 1)
+            self.assertEqual(tasks[0]["type"], "mail-dns")
+            self.assertEqual(tasks[0]["key"], "example.test")
+            self.assertEqual(tasks[0]["severity"], "warning")
+            self.assertTrue(tasks[0]["dismissible"])
+            self.assertEqual(tasks[0]["details"]["actions"][0]["value"],
+                             "v=spf1 mx -all")
+            self.assertEqual(database.stat().st_mode & 0o777, 0o600)
+            self.assertEqual(database.parent.stat().st_mode & 0o777, 0o750)
+
+    def test_no_dns_task_database_is_created_without_actions(self):
+        result = DomainSecurityResult(
+            "example.test", "bongo", {},
+            {"spf": "", "dkim": "", "dmarc": ""}, ())
+        with tempfile.TemporaryDirectory() as temporary:
+            database = Path(temporary) / "tasks" / "tasks.sqlite3"
+            install_dns_tasks({"web": {"task_database": str(database)}},
+                              (result,))
+            self.assertFalse(database.exists())
+
 
 if __name__ == "__main__":
     unittest.main()