diff --git a/contrib/testing/README.md b/contrib/testing/README.md index 556b595..a817dec 100644 --- a/contrib/testing/README.md +++ b/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`. diff --git a/contrib/testing/mail-dns-live-check.py b/contrib/testing/mail-dns-live-check.py new file mode 100755 index 0000000..6725e36 --- /dev/null +++ b/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") diff --git a/src/apps/config/tests/test_security_material.py b/src/apps/config/tests/test_security_material.py index 32e4e9f..35a656a 100644 --- a/src/apps/config/tests/test_security_material.py +++ b/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()