tests/test_email_domain_setup_checker.py — Email Domain Setup Checker

"""Exercise the read-only DNS checker with synthetic responses.

By Zero Width Lab (zerowidthlab.com)
"""

from __future__ import annotations

import json
import importlib.util
import os
import subprocess
import sys
import tempfile
import unittest
import urllib.request
from pathlib import Path
from unittest import mock
from urllib.parse import urlencode

# By Zero Width Lab (zerowidthlab.com)
__author__ = "Zero Width Lab"

ROOT = Path(__file__).resolve().parent.parent
SCRIPT = ROOT / "scripts" / "email_domain_setup_checker.py"
DOMAIN = "example.com"
SELECTORS = ("default", "selector1", "selector2", "google", "k1", "s1", "s2")


def dns_url(name: str, record_type: str, resolver: str = "google") -> str:
    base = "https://dns.google/resolve" if resolver == "google" else "https://cloudflare-dns.com/dns-query"
    return base + "?" + urlencode({"name": name, "type": record_type})


def dns_answer(name: str, record_type: int, data: str) -> dict[str, object]:
    return {"name": name + ".", "type": record_type, "TTL": 300, "data": data}


def base_responses(resolver: str = "google") -> dict[str, object]:
    records = [
        (DOMAIN, "MX", [dns_answer(DOMAIN, 15, "10 mail.example.com.")]),
        (DOMAIN, "TXT", [dns_answer(DOMAIN, 16, '"v=spf1 mx -all"')]),
        ("_dmarc." + DOMAIN, "TXT", [dns_answer("_dmarc." + DOMAIN, 16, '"v=DMARC1; p=reject; pct=100; rua=mailto:reports@example.com; adkim=s; aspf=r; sp=quarantine"')]),
        ("_smtp._tls." + DOMAIN, "TXT", [dns_answer("_smtp._tls." + DOMAIN, 16, '"v=TLSRPTv1; rua=mailto:reports@example.com"')]),
        ("_mta-sts." + DOMAIN, "TXT", [dns_answer("_mta-sts." + DOMAIN, 16, '"v=STSv1; id=20260930"')]),
        ("default._bimi." + DOMAIN, "TXT", []),
    ]
    records.extend((selector + "._domainkey." + DOMAIN, "TXT", []) for selector in SELECTORS)
    return {dns_url(name, kind, resolver): {"Status": 0, "Answer": answers}
            for name, kind, answers in records}


def set_response(responses: dict[str, object], name: str, kind: str,
                 answers: list[dict[str, object]], resolver: str = "google") -> None:
    url = dns_url(name, kind, resolver)
    responses[url] = {"Status": 0, "Answer": answers}


class EmailDomainSetupCheckerTests(unittest.TestCase):
    def run_cli(self, *arguments: str, responses: dict[str, object] | None = None,
                fixture_path: Path | None = None, acknowledge: bool = True) -> subprocess.CompletedProcess[str]:
        with tempfile.TemporaryDirectory() as temporary:
            fixture = fixture_path or Path(temporary) / "responses.json"
            if fixture_path is None:
                fixture.write_text(json.dumps(responses if responses is not None else base_responses()), encoding="utf-8")
            environment = os.environ.copy()
            environment.pop("EMAIL_DNS_FAKE_RESPONSES", None)
            environment.pop("EMAIL_DNS_FAKE_TRANSPORT_TEST_ACK", None)
            environment["EMAIL_DNS_FAKE_RESPONSES"] = str(fixture)
            if acknowledge:
                environment["EMAIL_DNS_FAKE_TRANSPORT_TEST_ACK"] = "1"
            return subprocess.run([sys.executable, str(SCRIPT), *arguments], cwd=ROOT,
                                  capture_output=True, text=True, check=False, env=environment)

    def test_complete_check_and_json_output(self) -> None:
        result = self.run_cli("check", DOMAIN, "--json")
        self.assertEqual(result.returncode, 0, result.stderr)
        payload = json.loads(result.stdout)
        self.assertEqual(payload["domain"], DOMAIN)
        self.assertFalse(payload["incomplete"])
        self.assertEqual(payload["skipped"], [])
        self.assertIn("MX", result.stdout)
        self.assertIn("SPF", result.stdout)
        self.assertIn("DMARC", result.stdout)
        self.assertIn("DKIM", result.stdout)
        self.assertIn("Synthetic fixture", result.stderr)

    def test_common_dkim_selectors_are_probes_not_errors(self) -> None:
        result = self.run_cli("check", DOMAIN, "--json")
        self.assertEqual(result.returncode, 0, result.stderr)
        serialized = result.stdout.lower()
        for selector in SELECTORS:
            self.assertIn(selector, serialized)
        self.assertIn("probed", serialized)

    def test_user_selector_is_checked(self) -> None:
        responses = base_responses()
        set_response(responses, "custom._domainkey." + DOMAIN, "TXT",
                     [dns_answer("custom._domainkey." + DOMAIN, 16, '"v=DKIM1; k=rsa; p=AAAA"')])
        result = self.run_cli("check", DOMAIN, "--dkim-selector", "custom", "--json", responses=responses)
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertIn("custom", result.stdout)

    def test_spf_recursive_lookup_limit(self) -> None:
        responses = base_responses()
        set_response(responses, DOMAIN, "TXT", [dns_answer(DOMAIN, 16, '"v=spf1 include:level1.example.com -all"')])
        for index in range(1, 12):
            name = f"level{index}.example.com"
            following = f"level{index + 1}.example.com"
            set_response(responses, name, "TXT", [dns_answer(name, 16, f'"v=spf1 include:{following} -all"')])
        # The eleventh include must not be fetched, even when its fixture is absent.
        result = self.run_cli("check", DOMAIN, "--json", responses=responses)
        self.assertEqual(result.returncode, 0, result.stderr)
        payload = json.loads(result.stdout)
        self.assertFalse(payload["incomplete"])
        self.assertTrue(any(item["message"] == "static lookup count is 11, above the 10-lookup limit; evaluation can stop earlier for a matching sender"
                            for item in payload["findings"]))

    def test_spf_counts_all_querying_mechanisms_and_redirect_across_includes(self) -> None:
        responses = base_responses()
        set_response(responses, DOMAIN, "TXT", [dns_answer(
            DOMAIN, 16, '"v=spf1 a mx ptr exists:one.example.com include:child.example.com -all"')])
        child = "child.example.com"
        set_response(responses, child, "TXT", [dns_answer(
            child, 16, '"v=spf1 a mx ptr exists:two.example.com redirect=next.example.com"')])
        next_domain = "next.example.com"
        set_response(responses, next_domain, "TXT", [dns_answer(
            next_domain, 16, '"v=spf1 include:never.example.com -all"')])
        result = self.run_cli("check", DOMAIN, "--json", responses=responses)
        self.assertEqual(result.returncode, 0, result.stderr)
        payload = json.loads(result.stdout)
        self.assertFalse(payload["incomplete"])
        self.assertTrue(any(item["message"] == "static lookup count is 11, above the 10-lookup limit; evaluation can stop earlier for a matching sender"
                            for item in payload["findings"]))

    def static_spf(self, policies: dict[str, str]):
        specification = importlib.util.spec_from_file_location("static_checker", SCRIPT)
        module = importlib.util.module_from_spec(specification)
        specification.loader.exec_module(module)
        class SyntheticClient:
            def __init__(self):
                self.names = []
            def dns(self, name, kind):
                self.names.append((name, kind))
                if kind == "MX":
                    return ["10 mail.example.com."]
                return [policies[name]] if name in policies else []
        client = SyntheticClient()
        result = module.check_domain(DOMAIN, [], "google", False, client)
        return [item for item in result["findings"] if item["check"] == "SPF"], client.names

    def test_spf_all_stops_counting_and_target_queries(self) -> None:
        for qualifier in ("", "+", "-", "~", "?"):
            with self.subTest(qualifier=qualifier):
                findings, names = self.static_spf({DOMAIN:
                    f"v=spf1 a {qualifier}all include:ignored.example.org mx ptr exists:x.example.org"})
                self.assertTrue(any("count: 1 of 10" in item["message"] for item in findings))
                self.assertNotIn(("ignored.example.org", "TXT"), names)
                self.assertFalse(any("discouraged ptr" in item["message"] for item in findings))

    def test_spf_redirect_ignored_if_all_present_anywhere(self) -> None:
        for policy in ("v=spf1 redirect=ignored.example.org a -all",
                       "v=spf1 a -all redirect=ignored.example.org"):
            with self.subTest(policy=policy):
                findings, names = self.static_spf({DOMAIN: policy})
                self.assertTrue(any("count: 1 of 10" in item["message"] for item in findings))
                self.assertNotIn(("ignored.example.org", "TXT"), names)

    def test_spf_redirect_followed_last_without_all(self) -> None:
        findings, names = self.static_spf({
            DOMAIN: "v=spf1 redirect=unrelated.example.org include:child.example.net a",
            "child.example.net": "v=spf1 mx -all",
            "unrelated.example.org": "v=spf1 exists:x.example.org -all",
        })
        self.assertTrue(any("count: 5 of 10" in item["message"] for item in findings))
        self.assertLess(names.index(("child.example.net", "TXT")),
                        names.index(("unrelated.example.org", "TXT")))

    def test_spf_nested_includes_count_each_occurrence(self) -> None:
        findings, names = self.static_spf({
            DOMAIN: "v=spf1 include:child.example.net include:child.example.net -all",
            "child.example.net": "v=spf1 include:nested.example.org -all",
            "nested.example.org": "v=spf1 mx -all a",
        })
        self.assertTrue(any("count: 6 of 10" in item["message"] for item in findings))
        self.assertEqual(names.count(("nested.example.org", "TXT")), 2)
        self.assertFalse(any("loop" in item["message"] for item in findings))

    def test_spf_underscore_target_names_reach_resolver(self) -> None:
        findings, names = self.static_spf({DOMAIN: "v=spf1 include:_spf.example.org -all",
                                         "_spf.example.org": "v=spf1 mx -all"})
        self.assertIn(("_spf.example.org", "TXT"), names)
        self.assertTrue(any("count: 2 of 10" in item["message"] for item in findings))
        self.assertFalse(any("Invalid SPF target" in item["message"] for item in findings))

    def test_spf_ten_lookup_boundary_and_static_wording(self) -> None:
        for count in (10, 11):
            with self.subTest(count=count):
                findings, _ = self.static_spf({DOMAIN: "v=spf1 " + "a " * count + "-all"})
                if count == 10:
                    self.assertTrue(any("count: 10 of 10" in item["message"] for item in findings))
                    self.assertFalse(any(item["severity"] == "error" for item in findings))
                else:
                    self.assertTrue(any("static lookup count is 11, above the 10-lookup limit" in item["message"] and item["severity"] == "warning" for item in findings))
                self.assertFalse(any("permerror" in item["message"] for item in findings))
                self.assertTrue(any("not a sender evaluation" in item["message"] or "evaluation can stop earlier for a matching sender" in item["message"] for item in findings))

    def test_spf_child_all_does_not_stop_parent_and_exp_is_not_counted(self) -> None:
        findings, names = self.static_spf({
            DOMAIN: "v=SpF1 InClUdE:Child.Example.Net MX ExP=ignored.example.org -AlL",
            "child.example.net": "v=SPF1 A -ALL include:ignored.example.org",
        })
        self.assertTrue(any("count: 3 of 10" in item["message"] for item in findings))
        self.assertNotIn(("ignored.example.org", "TXT"), names)
        self.assertIn(("child.example.net", "TXT"), names)

    def test_spf_duplicate_redirect_and_macro_targets_are_partial_warnings(self) -> None:
        for policy, message, count in (
            ("v=spf1 a redirect=first.example.org ReDiReCt=second.example.org",
             "duplicate redirect modifier (RFC 7208 section 6)", 1),
            ("v=spf1 a -all redirect=first.example.org redirect=second.example.org",
             "duplicate redirect modifier (RFC 7208 section 6)", 1),
            ("v=spf1 include:%{d}.example.org -all",
             "macro target not followed (static count is partial)", 1),
            ("v=spf1 ReDiReCt=%{d}.example.org",
             "macro target not followed (static count is partial)", 1),
        ):
            with self.subTest(policy=policy):
                responses = base_responses()
                set_response(responses, DOMAIN, "TXT", [dns_answer(DOMAIN, 16, policy)])
                for allow_partial, exit_code in ((False, 2), (True, 0)):
                    arguments = ["check", DOMAIN, "--json"] + (["--allow-partial"] if allow_partial else [])
                    result = self.run_cli(*arguments, responses=responses)
                    self.assertEqual(result.returncode, exit_code, result.stderr)
                    payload = json.loads(result.stdout)
                    self.assertTrue(payload["incomplete"])
                    self.assertTrue(any(message in item for item in payload["skipped"]))
                    self.assertTrue(any(item["message"] == message and item["severity"] == "warning"
                                        for item in payload["findings"]))
                    self.assertFalse(any(item["severity"] == "error" for item in payload["findings"]))
                findings, names = self.static_spf({DOMAIN: policy})
                self.assertTrue(any(f"count: {count} of 10" in item["message"] for item in findings))
                self.assertEqual([name for name, kind in names if kind == "TXT" and name in
                                  ("first.example.org", "second.example.org", "%{d}.example.org")], [])

    def test_spf_redirect_loop_is_bounded(self) -> None:
        findings, names = self.static_spf({DOMAIN: "v=spf1 redirect=loop.example.org",
                                         "loop.example.org": "v=spf1 redirect=example.com"})
        self.assertTrue(any("loop" in item["message"] for item in findings))
        self.assertLess(len(names), 20)

    def test_total_dns_query_ceiling_marks_remaining_checks_incomplete(self) -> None:
        responses = base_responses()
        selectors = [f"custom{index}" for index in range(59)]
        for selector in selectors[:58]:
            set_response(responses, f"{selector}._domainkey.{DOMAIN}", "TXT", [])
        arguments = ["check", DOMAIN, "--json", "--mta-sts"]
        for selector in selectors:
            arguments.extend(("--dkim-selector", selector))
        result = self.run_cli(*arguments, responses=responses)
        self.assertEqual(result.returncode, 2, result.stderr)
        payload = json.loads(result.stdout)
        self.assertTrue(payload["incomplete"])
        self.assertTrue(any("DNS query limit reached (60)" in item for item in payload["skipped"]))
        self.assertFalse(any("missing fake response" in item for item in payload["skipped"]))
        self.assertFalse(any(item["check"] == "TLS-RPT" for item in payload["findings"]))
        self.assertTrue(any("MTA-STS HTTPS policy" in item for item in payload["skipped"]))

    def test_resolver_accept_header_is_cloudflare_only(self) -> None:
        specification = importlib.util.spec_from_file_location("checker_module", SCRIPT)
        self.assertIsNotNone(specification)
        self.assertIsNotNone(specification.loader)
        module = importlib.util.module_from_spec(specification)
        specification.loader.exec_module(module)
        class Response:
            url = "https://dns.google/resolve"
            def __enter__(self):
                return self
            def __exit__(self, *_arguments):
                return False
            def read(self, _size):
                return b'{"Status": 0, "Answer": []}'
        for resolver in ("google", "cloudflare"):
            with self.subTest(resolver=resolver), mock.patch.dict(os.environ, {
                "EMAIL_DNS_FAKE_RESPONSES": "", "EMAIL_DNS_FAKE_TRANSPORT_TEST_ACK": ""
            }):
                client = module.Client(DOMAIN, resolver)
                response = Response()
                response.url = dns_url(DOMAIN, "TXT", resolver)
                client.opener.open = mock.Mock(return_value=response)
                self.assertEqual(client.dns(DOMAIN, "TXT"), [])
                request = client.opener.open.call_args.args[0]
                self.assertEqual(request.get_header("Accept"),
                                 "application/dns-json" if resolver == "cloudflare" else None)

    def test_alternate_fixture_list_is_rejected(self) -> None:
        result = self.run_cli("check", DOMAIN, responses={"responses": []})
        self.assertEqual(result.returncode, 2)
        self.assertIn("map URLs directly", result.stderr)

    def test_spf_include_loop_is_bounded(self) -> None:
        responses = base_responses()
        set_response(responses, DOMAIN, "TXT", [dns_answer(DOMAIN, 16, '"v=spf1 include:loop.example.com -all"')])
        set_response(responses, "loop.example.com", "TXT",
                     [dns_answer("loop.example.com", 16, '"v=spf1 include:example.com -all"')])
        result = self.run_cli("check", DOMAIN, "--json", responses=responses)
        self.assertEqual(result.returncode, 3, result.stderr)
        self.assertIn("loop", result.stdout.lower())

    def test_dmarc_policy_and_tags_are_reported(self) -> None:
        result = self.run_cli("check", DOMAIN, "--json")
        self.assertEqual(result.returncode, 0, result.stderr)
        for term in ("reject", "quarantine", "pct", "rua", "ruf", "adkim", "aspf"):
            self.assertIn(term, result.stdout.lower())

    def test_partial_lookup_exits_two_and_allow_partial_returns_result_status(self) -> None:
        responses = base_responses()
        responses.pop(dns_url("_smtp._tls." + DOMAIN, "TXT"))
        refused = self.run_cli("check", DOMAIN, "--json", responses=responses)
        allowed = self.run_cli("check", DOMAIN, "--json", "--allow-partial", responses=responses)
        self.assertEqual(refused.returncode, 2, refused.stderr)
        self.assertEqual(allowed.returncode, 0, allowed.stderr)
        for result in (refused, allowed):
            payload = json.loads(result.stdout)
            self.assertTrue(payload["incomplete"])
            self.assertTrue(payload["skipped"])
            self.assertIn("_smtp._tls", result.stdout)

    def test_missing_fixture_fails_before_request(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            missing = Path(temporary) / "missing.json"
            result = self.run_cli("check", DOMAIN, fixture_path=missing)
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("fixture", result.stderr.lower())

    def test_fake_transport_requires_both_environment_variables(self) -> None:
        result = self.run_cli("check", DOMAIN, acknowledge=False)
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("ack", result.stderr.lower())

    def test_invalid_domain_rejected(self) -> None:
        for domain in ("bad..example", "-bad.example", "https://example.com", "127.0.0.1"):
            with self.subTest(domain=domain):
                result = self.run_cli("check", domain)
                self.assertNotEqual(result.returncode, 0)
                self.assertIn("domain", result.stderr.lower())

    def test_cloudflare_resolver_uses_declared_host(self) -> None:
        result = self.run_cli("check", DOMAIN, "--resolver", "cloudflare", "--json",
                              responses=base_responses("cloudflare"))
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(json.loads(result.stdout)["resolver"], "cloudflare")

    def test_mta_sts_policy_is_fetched_only_when_requested(self) -> None:
        responses = base_responses()
        policy_url = "https://mta-sts.example.com/.well-known/mta-sts.txt"
        responses[policy_url] = "version: STSv1\nmode: enforce\nmx: mail.example.com\nmax_age: 604800\n"
        normal = self.run_cli("check", DOMAIN, "--json", responses=base_responses())
        selected = self.run_cli("check", DOMAIN, "--mta-sts", "--json", responses=responses)
        self.assertEqual(normal.returncode, 0, normal.stderr)
        self.assertEqual(selected.returncode, 0, selected.stderr)
        self.assertNotIn("MTA-STS policy", normal.stdout)
        self.assertIn("MTA-STS policy", selected.stdout)
        self.assertIn("enforce", selected.stdout)

    def test_redirect_guard_rechecks_exact_https_hosts(self) -> None:
        specification = importlib.util.spec_from_file_location("checker_module", SCRIPT)
        self.assertIsNotNone(specification)
        self.assertIsNotNone(specification.loader)
        module = importlib.util.module_from_spec(specification)
        specification.loader.exec_module(module)
        for url in ("http://dns.google/resolve", "https://dns.google.evil.example/resolve",
                    "https://mta-sts.other.example/.well-known/mta-sts.txt",
                    "https://dns.google:444/resolve"):
            with self.subTest(url=url), self.assertRaises(module.CheckError):
                module.guarded_url(url, DOMAIN)
        module.guarded_url("https://dns.google/resolve", DOMAIN)
        module.guarded_url("https://cloudflare-dns.com/dns-query", DOMAIN)
        module.guarded_url("https://mta-sts.example.com/.well-known/mta-sts.txt", DOMAIN)
        redirect = module.GuardedRedirect(DOMAIN)
        original = urllib.request.Request("https://dns.google/resolve")
        with self.assertRaises(module.CheckError):
            redirect.redirect_request(original, None, 302, "Found", {}, "https://outside.example/steal")

    def test_untrusted_dns_data_cannot_forge_text_lines(self) -> None:
        responses = base_responses()
        hostile = "10 mail.example.com.\nFORGED\x1b[31m\u2028"
        set_response(responses, DOMAIN, "MX", [dns_answer(DOMAIN, 15, hostile)])
        result = self.run_cli("check", DOMAIN, responses=responses)
        self.assertNotIn("\x1b", result.stdout)
        self.assertNotIn("\u2028", result.stdout)
        self.assertNotIn("\nFORGED", result.stdout)
        self.assertIn("FORGED", result.stdout)


if __name__ == "__main__":
    unittest.main()

License: MIT (read it).

Back to the listing