scripts/email_domain_setup_checker.py — Email Domain Setup Checker

#!/usr/bin/env python3
# Email Domain Setup Checker 0.1.1
# By Zero Width Lab (zerowidthlab.com)
"""Read-only email domain checks using guarded DNS-over-HTTPS.

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

from __future__ import annotations

import argparse
import json
import os
import re
import sys
import unicodedata
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path
from typing import Any

__author__ = "Zero Width Lab"

SIGNATURE = "By Zero Width Lab (zerowidthlab.com)"
FAKE_PATH = "EMAIL_DNS_FAKE_RESPONSES"
FAKE_ACK = "EMAIL_DNS_FAKE_TRANSPORT_TEST_ACK"
RESOLVERS = {"google": "dns.google", "cloudflare": "cloudflare-dns.com"}
COMMON_SELECTORS = ("default", "selector1", "selector2", "google", "k1", "s1", "s2")
MAX_BYTES = 1_048_576
MAX_DNS_QUERIES = 60


class CheckError(Exception):
    """A request or input failure suitable for a user-facing message."""


class QueryLimitError(CheckError):
    """The run has exhausted its DNS request budget."""


def display_text(value: Any) -> str:
    parts = []
    for character in str(value):
        code = ord(character)
        if unicodedata.category(character).startswith("C") or code in (0x2028, 0x2029):
            parts.append(f"\\u{code:04x}" if code <= 0xFFFF else f"\\U{code:08x}")
        else:
            parts.append(character)
    return "".join(parts)


def domain_name(value: str, *, spf_target: bool = False) -> str:
    value = value.rstrip(".")
    if not value or len(value) > 253 or ".." in value:
        raise CheckError("invalid domain name")
    try:
        ascii_name = value.encode("idna").decode("ascii").lower()
    except (UnicodeError, ValueError) as exc:
        raise CheckError("invalid domain name") from exc
    labels = ascii_name.split(".")
    label_pattern = (r"[a-z0-9_](?:[a-z0-9_-]*[a-z0-9_])?" if spf_target
                     else r"[a-z0-9](?:[a-z0-9-]*[a-z0-9])?")
    if len(labels) < 2 or any(
        len(label) > 63 or not re.fullmatch(label_pattern, label)
        for label in labels
    ) or not re.fullmatch(r"[a-z]{2,63}|xn--[a-z0-9-]{2,59}", labels[-1]):
        raise CheckError("invalid domain name")
    return ascii_name


def selector_name(value: str) -> str:
    if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_-]{0,62}", value):
        raise CheckError("invalid DKIM selector")
    return value.lower()


def guarded_url(url: str, domain: str) -> None:
    parsed = urllib.parse.urlsplit(url)
    try:
        port = parsed.port
    except ValueError as exc:
        raise CheckError("invalid destination port") from exc
    allowed = set(RESOLVERS.values()) | {f"mta-sts.{domain}"}
    if (parsed.scheme != "https" or parsed.hostname not in allowed or port not in (None, 443)
            or parsed.username is not None or parsed.password is not None or parsed.fragment):
        raise CheckError("destination is outside the allowed HTTPS hosts")


class GuardedRedirect(urllib.request.HTTPRedirectHandler):
    def __init__(self, domain: str):
        self.domain = domain

    def redirect_request(self, req: Any, fp: Any, code: int, msg: str,
                         headers: Any, newurl: str) -> Any:
        guarded_url(newurl, self.domain)
        if urllib.parse.urlsplit(req.full_url).hostname != urllib.parse.urlsplit(newurl).hostname:
            raise CheckError("redirect to a different host refused")
        return super().redirect_request(req, fp, code, msg, headers, newurl)


class Client:
    def __init__(self, domain: str, resolver: str):
        self.domain = domain
        self.resolver = resolver
        self.dns_queries = 0
        self.opener = urllib.request.build_opener(GuardedRedirect(domain))
        self.fake: dict[str, Any] | None = None
        fixture = os.environ.get(FAKE_PATH)
        acknowledgement = os.environ.get(FAKE_ACK)
        if fixture or acknowledgement:
            if not fixture or acknowledgement != "1":
                raise CheckError(f"fake transport needs both {FAKE_PATH} and {FAKE_ACK}=1")
            print("warning: Synthetic fixture transport active; no real network requests will be made", file=sys.stderr)
            try:
                loaded = json.loads(Path(fixture).read_text(encoding="utf-8"))
            except (OSError, UnicodeError, json.JSONDecodeError) as exc:
                raise CheckError(f"cannot read fake transport fixture: {display_text(exc)}") from exc
            if not isinstance(loaded, dict):
                raise CheckError("fake transport fixture must be a JSON object")
            if "responses" in loaded:
                raise CheckError("fake transport fixture must map URLs directly to responses")
            self.fake = loaded

    def request(self, url: str, accept: str | None = None) -> bytes:
        guarded_url(url, self.domain)
        hostname = urllib.parse.urlsplit(url).hostname
        if hostname not in {RESOLVERS[self.resolver], f"mta-sts.{self.domain}"}:
            raise CheckError("destination is not the selected resolver or domain policy host")
        if self.fake is not None:
            if url not in self.fake:
                raise CheckError(f"missing fake response for {display_text(url)}")
            item = self.fake[url]
            if isinstance(item, dict) and "error" in item:
                raise CheckError(str(item["error"]))
            if isinstance(item, str):
                data = item.encode("utf-8")
            else:
                data = json.dumps(item).encode("utf-8")
            if len(data) > MAX_BYTES:
                raise CheckError("response exceeds size limit")
            return data
        headers = {"User-Agent": "EmailDomainSetupChecker/0.1.1"}
        if accept is not None:
            headers["Accept"] = accept
        request = urllib.request.Request(url, headers=headers)
        try:
            with self.opener.open(request, timeout=15) as response:
                guarded_url(response.url, self.domain)
                data = response.read(MAX_BYTES + 1)
        except (urllib.error.URLError, TimeoutError, OSError, ValueError) as exc:
            raise CheckError(f"HTTPS lookup failed: {display_text(exc)}") from exc
        if len(data) > MAX_BYTES:
            raise CheckError("response exceeds size limit")
        return data

    def dns(self, name: str, record_type: str) -> list[str]:
        if self.dns_queries >= MAX_DNS_QUERIES:
            raise QueryLimitError(f"DNS query limit reached ({MAX_DNS_QUERIES}); remaining checks skipped")
        self.dns_queries += 1
        host = RESOLVERS[self.resolver]
        path = "/resolve" if self.resolver == "google" else "/dns-query"
        query = urllib.parse.urlencode({"name": name, "type": record_type})
        accept = "application/dns-json" if self.resolver == "cloudflare" else None
        raw = self.request(f"https://{host}{path}?{query}", accept)
        try:
            response = json.loads(raw.decode("utf-8"))
            if not isinstance(response, dict) or not isinstance(response.get("Status"), int):
                raise ValueError("invalid DNS response")
            status = response["Status"]
            if status == 3:
                return []
            if status != 0:
                raise CheckError(f"DNS response status {status}")
            answers = response.get("Answer", [])
            if not isinstance(answers, list):
                raise ValueError("invalid DNS answers")
            type_number = {"MX": 15, "TXT": 16}[record_type]
            return [str(item["data"]) for item in answers
                    if isinstance(item, dict) and item.get("type") == type_number and isinstance(item.get("data"), str)]
        except (UnicodeError, json.JSONDecodeError, ValueError, KeyError) as exc:
            raise CheckError(f"invalid DNS response: {display_text(exc)}") from exc


def txt_value(value: str) -> str:
    chunks = re.findall(r'"((?:\\.|[^"\\])*)"', value)
    if chunks:
        return "".join(chunk.replace(r'\"', '"').replace(r"\\", "\\") for chunk in chunks)
    return value


def tags(value: str) -> dict[str, str]:
    parsed = {}
    for item in value.split(";"):
        if "=" in item:
            key, content = item.split("=", 1)
            parsed[key.strip().lower()] = content.strip()
    return parsed


def check_domain(domain: str, selectors: list[str], resolver: str, mta_sts: bool,
                 client: Client) -> dict[str, Any]:
    findings: list[dict[str, str]] = []
    skipped: list[str] = []
    query_limit_reached = False
    def add(check: str, severity: str, message: str) -> None:
        findings.append({"check": check, "severity": severity, "message": message})
    def lookup(name: str, record_type: str, check: str) -> list[str] | None:
        nonlocal query_limit_reached
        if query_limit_reached:
            skipped.append(f"{check}: {name} {record_type} (DNS query limit reached ({MAX_DNS_QUERIES}))")
            return None
        try:
            return client.dns(name, record_type)
        except QueryLimitError as exc:
            query_limit_reached = True
            skipped.append(f"{check}: {name} {record_type} ({exc})")
            return None
        except CheckError as exc:
            skipped.append(f"{check}: {name} {record_type} ({exc})")
            return None
    mx = lookup(domain, "MX", "MX")
    if mx is not None:
        add("MX", "ok" if mx else "error", f"{len(mx)} MX record(s) found" if mx else "No MX records found")
        for record in mx:
            add("MX", "info", f"MX target: {record}")
    root_txt = lookup(domain, "TXT", "SPF")
    if root_txt is not None:
        spf = [txt_value(x) for x in root_txt if txt_value(x).lower().startswith("v=spf1")]
        if len(spf) != 1:
            add("SPF", "error", f"Expected exactly one SPF record; found {len(spf)}")
        else:
            count = 0
            visited: set[str] = set()
            def walk(current: str, policy: str, depth: int) -> bool:
                nonlocal count
                if depth > 20 or current in visited:
                    add("SPF", "error", f"SPF include or redirect loop at {current}")
                    return False
                visited.add(current)
                try:
                    # RFC 7208: left-to-right mechanisms stop at all (5.1);
                    # redirect applies last and is ignored if any all is present (6.1).
                    # Count include, a, mx, ptr, exists and redirect recursively (4.6.4).
                    # Static inventory only: no sender-IP matching or macro evaluation.
                    tokens = policy.split()[1:]
                    has_all = any(token.lstrip("+~-?").lower() == "all" for token in tokens)
                    mechanisms = [token for token in tokens if "=" not in token]
                    redirects = [token for token in tokens if token.lower().startswith("redirect=")]
                    if len(redirects) > 1:
                        message = "duplicate redirect modifier (RFC 7208 section 6)"
                        add("SPF", "warning", message)
                        skipped.append(f"SPF: {current}: {message}; redirect targets not followed (static count is partial)")
                    ordered = mechanisms + ([] if has_all or len(redirects) > 1 else redirects)
                    for token in ordered:
                        mechanism = token.lstrip("+~-?")
                        lower = mechanism.lower()
                        if lower == "all" and (token.startswith(("+", "?")) or token.lower() == "all"):
                            add("SPF", "warning", f"SPF uses {token}; review whether it permits spoofing")
                        if lower == "all":
                            break
                        elif lower == "ptr" or lower.startswith("ptr:"):
                            add("SPF", "warning", "SPF uses the discouraged ptr mechanism")
                        if (re.match(r"^(?:a|mx|ptr)(?:[:/]|$)", lower)
                                or lower.startswith(("exists:", "include:", "redirect="))):
                            count += 1
                            if count > 10:
                                return True
                        target = None
                        if lower.startswith("include:"):
                            target = mechanism.split(":", 1)[1]
                        elif lower.startswith("redirect="):
                            target = mechanism.split("=", 1)[1]
                        if target:
                            if "%{" in target:
                                message = "macro target not followed (static count is partial)"
                                add("SPF", "warning", message)
                                skipped.append(f"SPF: {current}: {target}: {message}")
                                continue
                            try:
                                target = domain_name(target, spf_target=True)
                            except CheckError:
                                add("SPF", "error", f"Invalid SPF target {target}")
                                continue
                            nested = lookup(target, "TXT", "SPF")
                            if nested is not None:
                                policies = [txt_value(x) for x in nested if txt_value(x).lower().startswith("v=spf1")]
                                if len(policies) != 1:
                                    add("SPF", "warning", f"Expected one SPF record at {target}; found {len(policies)}")
                                elif walk(target, policies[0], depth + 1):
                                    return True
                    return False
                finally:
                    visited.remove(current)
            exceeded = walk(domain, spf[0], 0)
            if exceeded:
                add("SPF", "warning", f"static lookup count is {count}, above the 10-lookup limit; evaluation can stop earlier for a matching sender")
            else:
                add("SPF", "ok", f"Static SPF lookup count: {count} of 10 (RFC 7208 processing order; not a sender evaluation)")
    dmarc = lookup(f"_dmarc.{domain}", "TXT", "DMARC")
    if dmarc is not None:
        records = [txt_value(x) for x in dmarc if txt_value(x).lower().startswith("v=dmarc1")]
        if len(records) != 1:
            add("DMARC", "error", f"Expected one DMARC record; found {len(records)}")
        else:
            values = tags(records[0])
            policy = values.get("p", "missing")
            add("DMARC", "warning" if policy == "none" else "ok" if policy in ("quarantine", "reject") else "error", f"DMARC policy: {policy}")
            for key, label in (("pct", "percentage"), ("rua", "aggregate reports"), ("ruf", "forensic reports"), ("adkim", "DKIM alignment"), ("aspf", "SPF alignment"), ("sp", "subdomain policy")):
                add("DMARC", "info", f"DMARC {key} ({label}): {values.get(key, 'not specified')}")
    requested = list(dict.fromkeys(selector_name(x) for x in selectors))
    for selector in list(dict.fromkeys(requested + list(COMMON_SELECTORS))):
        dkim = lookup(f"{selector}._domainkey.{domain}", "TXT", "DKIM")
        if dkim is not None:
            found = any(txt_value(x).lower().startswith("v=dkim1") or "p=" in txt_value(x).lower() for x in dkim)
            origin = "requested" if selector in requested else "probed"
            add("DKIM", "ok" if found else "warning" if selector in requested else "info", f"DKIM selector {selector} ({origin}): {'record found' if found else 'no record found'}")
    for check, name, prefix, severity in (("TLS-RPT", f"_smtp._tls.{domain}", "v=tlsrptv1", "warning"),
                                          ("MTA-STS", f"_mta-sts.{domain}", "v=stsv1", "warning"),
                                          ("BIMI", f"default._bimi.{domain}", "v=bimi1", "info")):
        values = lookup(name, "TXT", check)
        if values is not None:
            found = any(txt_value(x).lower().startswith(prefix) for x in values)
            add(check, "ok" if found else severity, f"{check} TXT record {'found' if found else 'not found'}")
    if mta_sts and query_limit_reached:
        skipped.append("MTA-STS HTTPS policy: DNS query limit reached; remaining checks skipped")
    elif mta_sts:
        url = f"https://mta-sts.{domain}/.well-known/mta-sts.txt"
        try:
            policy_text = client.request(url, "text/plain").decode("utf-8")
            policy: dict[str, list[str]] = {}
            for line in policy_text.splitlines():
                if ":" in line:
                    key, value = line.split(":", 1)
                    policy.setdefault(key.strip().lower(), []).append(value.strip())
            mode = policy.get("mode", ["missing"])[0].lower()
            version = policy.get("version", ["missing"])[0].lower()
            max_age = policy.get("max_age", ["missing"])[0]
            mx_values = policy.get("mx", [])
            valid = (version == "stsv1" and mode in ("enforce", "testing", "none")
                     and max_age.isdigit() and bool(mx_values)
                     and all(len(policy.get(key, [])) == 1 for key in ("version", "mode", "max_age")))
            add("MTA-STS policy", "ok" if valid else "warning", f"HTTPS policy: version {version}, mode {mode}, max_age {max_age}, MX {'present' if mx_values else 'missing'}")
        except (CheckError, UnicodeError) as exc:
            skipped.append(f"MTA-STS HTTPS policy: {exc}")
    return {"domain": domain, "resolver": resolver, "incomplete": bool(skipped), "skipped": skipped, "findings": findings}


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Check public email-domain DNS setup", epilog=SIGNATURE)
    sub = parser.add_subparsers(dest="command", required=True)
    check = sub.add_parser("check", help="check one domain", epilog=SIGNATURE)
    check.add_argument("domain")
    check.add_argument("--dkim-selector", action="append", default=[])
    check.add_argument("--resolver", choices=tuple(RESOLVERS), default="google")
    check.add_argument("--json", action="store_true")
    check.add_argument("--mta-sts", action="store_true")
    check.add_argument("--allow-partial", action="store_true")
    args = parser.parse_args(argv)
    try:
        domain = domain_name(args.domain)
        selectors = [selector_name(x) for x in args.dkim_selector]
        client = Client(domain, args.resolver)
        result = check_domain(domain, selectors, args.resolver, args.mta_sts, client)
    except CheckError as exc:
        print(f"error: {display_text(exc)}", file=sys.stderr)
        return 2
    if args.json:
        print(json.dumps(result, ensure_ascii=True, indent=2))
    else:
        print(f"Email Domain Setup Checker: {display_text(result['domain'])}")
        for item in result["findings"]:
            print(f"[{item['severity']}] {item['check']}: {display_text(item['message'])}")
        if result["incomplete"]:
            print("incomplete: some checks were skipped")
            for item in result["skipped"]:
                print(f"skipped: {display_text(item)}")
    if result["incomplete"] and not args.allow_partial:
        return 2
    return 3 if any(x["severity"] == "error" for x in result["findings"]) else 0


if __name__ == "__main__":
    raise SystemExit(main())

License: MIT (read it).

Back to the listing