#!/usr/bin/env python3 """Checks saved raw message sources for authentication, alignment and one-click unsubscribe. Usage: python3 scripts/check_message_headers.py SOURCE.txt [MORE.txt ...] [--class marketing|transactional] [--volume bulk|low] [--microsoft-volume bulk|low] [--json] Each file is the raw RFC 5322 source of one received message, saved from a seed inbox the team controls (the receiving mailbox, not the sent folder). Any file extension works. Python 3.9+, standard library only. Reads files, writes nothing, and never prints To, Cc or Delivered-To addresses. Checks (thresholds and reasons: references/deliverability.md): DL05 dmarc the receiver's Authentication-Results show dmarc=pass DL06 microsoft-both-pass spf=pass and dkim=pass on the same message DL04 dkim a passing DKIM signature aligns with the From domain; a marketing message with no DKIM-Signature at all is FAIL DL07 one-click-and-body-link HTTPS List-Unsubscribe, List-Unsubscribe-Post, both headers in a passing signature's h= tag, and a visible unsubscribe link in the body (marketing class only) DL08 transport-and-dns a Received header shows TLS The receiver's results are the topmost Authentication-Results header plus the next ones from the same receiver (same organisational domain in the authserv-id), stopping at the first header from another server. Some receivers write one header per method. Microsoft writes no authserv-id, so a topmost header with none is used alone. Lower headers from other servers are listed and ignored. --volume bulk (the default) applies the mailbox providers' bulk sender rules, so DL05 and the DL07 header checks are FAIL when they break. Use --volume low only for a domain below every provider's bulk threshold (Gmail: close to 5,000 a day to personal Gmail accounts, counted on the primary domain and permanent once reached); those checks then report WARN. --microsoft-volume (default bulk) does the same for DL06: pass low when the From domain sends under 5,000 a day to Microsoft consumer mailboxes. A headers-only file (no body) reports the DL07 body check as NOT CHECKED. The DL07 body check reads the HTML parts when there are any (what Gmail and Yahoo render), otherwise the plain text parts, and ignores words inside URLs. PASS: unsubscribe wording on a link or within 80 characters of one in the same block, in English, German, French, Spanish, Dutch, Italian or Portuguese, or a body link that reuses the HTTPS List-Unsubscribe URL. WARN: a preferences link only, unsubscribe wording away from any link, or links whose wording the check does not recognise (read the body and rate it by hand). FAIL, at any volume: no link in the body at all. Alignment here is an approximation (same domain or same organisational domain by label count). The receiver's dmarc= result is the authority. DKIM key length cannot be read from headers; check it in DNS (references/deliverability.md). Exit status: 1 when any FAIL, 2 when a file cannot be read, 0 otherwise. """ import argparse import email import html import json import re import sys from email import policy from email.utils import parseaddr SECOND_LEVEL = {"co", "com", "org", "net", "ac", "gov", "edu", "ne", "or", "gr", "ltd", "plc"} TLS_PATTERN = re.compile(r"ESMTPSA?|TLSv?1[._][0-3]|version=TLS", re.I) UNSUBSCRIBE_WORDS = re.compile( r"unsubscri|opt[ -]?out" # English r"|abmelden|abbestellen" # German r"|d[e\u00e9]sabonn|d[e\u00e9]sinscri" # French r"|d(?:ar|ate|arse) de baja|desuscri" # Spanish r"|uitschrijven|afmelden" # Dutch r"|disiscri" # Italian r"|descadastr", # Portuguese re.I) PREFERENCE_WORDS = re.compile(r"pr[e\u00e9]f[e\u00e9\u00e8\u00ea]r[e\u00e9\u00e8\u00ea]n[cz]", re.I) LINK_PATTERN = re.compile(r"(?:https?|mailto):[^\s<>\"']+", re.I) ANCHOR_HREF = re.compile(r"""]*?\bhref\s*=\s*(?:"([^"]*)"|'([^']*)'|([^\s>]+))""", re.I | re.S) ANCHOR_BLOCK = re.compile(r"(?is)]*\bhref[^>]*>(.*?)") METHOD_FIRST = re.compile(r"^[a-z0-9-]+\s*=", re.I) NEAR_LINK = 80 # characters either side of a link, in the same block, that count as its label BLOCK_TAG = re.compile(r"(?i)]*>|") def unfold(value): return re.sub(r"\s+", " ", str(value or "")).strip() def domain_of(address): _, addr = parseaddr(unfold(address)) return addr.rsplit("@", 1)[1].lower().strip(".>") if "@" in addr else "" def org_domain(domain): labels = [x for x in domain.lower().strip(".").split(".") if x] if len(labels) >= 3 and len(labels[-1]) == 2 and labels[-2] in SECOND_LEVEL: return ".".join(labels[-3:]) return ".".join(labels[-2:]) def aligned(domain, from_domain): return bool(domain and from_domain) and (domain == from_domain or org_domain(domain) == org_domain(from_domain)) def strip_comments(value): text = unfold(value) while True: stripped = re.sub(r"\([^()]*\)", " ", text) if stripped == text: return text text = stripped def parse_auth_results(value): """Returns (authserv_id, results). Microsoft writes no authserv-id, so a first field that already reads method=result is parsed as a result.""" parts = [p.strip() for p in strip_comments(value).split(";")] authserv = "" if parts and parts[0] and not METHOD_FIRST.match(parts[0]): authserv = parts[0].split()[0].lower() parts = parts[1:] results = [] for part in parts: match = re.match(r"([a-z0-9-]+)\s*=\s*([a-z0-9-]+)", part, re.I) if not match: continue props = {k.lower(): v for k, v in re.findall(r"([a-z]+\.[a-z-]+)=([^\s;]+)", part, re.I)} results.append({"method": match.group(1).lower(), "result": match.group(2).lower(), "props": props}) return authserv, results def receiver_results(ar_headers): """Merges the Authentication-Results headers the receiving mailbox added. Returns (block, ignored, merged): the headers used, the lower headers ignored, and per method the authserv-id and results of the first header in the block that reports that method.""" parsed = [parse_auth_results(value) for value in ar_headers] top_id = parsed[0][0] if not top_id: block = parsed[:1] else: top_org = org_domain(top_id) block = [] for authserv, results in parsed: if not authserv or org_domain(authserv) != top_org: break block.append((authserv, results)) ignored = parsed[len(block):] merged = {} for authserv, results in block: seen = {} for result in results: seen.setdefault(result["method"], []).append(result) for method, found in seen.items(): merged.setdefault(method, (authserv, found)) return block, ignored, merged def dkim_domain(result): props = result["props"] return (props.get("header.d", "") or props.get("header.i", "").split("@")[-1]).lower() def has_body(message): if message.is_multipart(): return True return bool(str(message.get_payload() or "").strip()) def parse_dkim_signature(value): tags = {} for chunk in unfold(value).split(";"): if "=" in chunk: key, val = chunk.split("=", 1) tags[key.strip().lower()] = val.strip() headers = [h.strip().lower() for h in re.sub(r"\s+", "", tags.get("h", "")).split(":") if h.strip()] return {"d": tags.get("d", "").lower(), "s": tags.get("s", ""), "h": headers} def signature_passed(signature, passing_domains): """True when a passing result belongs to this signature. Receivers can report the i= tag, which may be a subdomain of d=, as in header.i=@news.example.com for d=example.com.""" domain = signature["d"] return bool(domain) and any(d == domain or d.endswith("." + domain) for d in passing_domains) def visible(html_text): return html.unescape(re.sub(r"<[^>]+>", " ", html_text)) def link_labels(content, is_html): """The words a reader takes as a link's label: its anchor text and the text just before and after it.""" labels = [] for match in (ANCHOR_BLOCK if is_html else LINK_PATTERN).finditer(content): before, after = content[:match.start()], content[match.end():] if is_html: before = visible(BLOCK_TAG.split(before[-1000:])[-1]) after = visible(BLOCK_TAG.split(after[:1000])[0]) labels.append(visible(match.group(1))) labels.extend((LINK_PATTERN.sub(" ", before)[-NEAR_LINK:], LINK_PATTERN.sub(" ", after)[:NEAR_LINK])) return " ".join(labels) def body_view(message): """What the reader sees: the HTML parts when there are any (what Gmail and Yahoo render), otherwise the plain text parts. Returns (kind, links, text, labels): the http, https and mailto links, the visible text with URLs removed, so a word inside a URL never counts as wording, and the words on or beside a link.""" parts = {"html": [], "plain": []} for part in message.walk(): if part.get_content_maintype() != "text" or part.get_content_subtype() not in ("plain", "html"): continue if part.get_content_disposition() == "attachment": continue try: content = part.get_content() except Exception: # undecodable part: fall back to the raw payload content = part.get_payload(decode=True) or b"" content = content.decode("utf-8", "replace") if isinstance(content, bytes) else str(content) parts[part.get_content_subtype()].append(content) kind = "html" if parts["html"] else "plain" links, texts, labels = [], [], [] for content in parts[kind]: if kind == "html": content = re.sub(r"(?is)<(script|style|head)\b.*?", " ", content) for match in ANCHOR_HREF.finditer(content): links.append(html.unescape(next(g for g in match.groups() if g is not None)).strip()) labels.append(link_labels(content, True)) content = visible(content) else: labels.append(link_labels(content, False)) links.extend(LINK_PATTERN.findall(content)) texts.append(LINK_PATTERN.sub(" ", content)) links = [link.rstrip(".,;:)") for link in links if re.match(r"(?:https?|mailto):", link, re.I)] return kind, links, " ".join(texts), " ".join(labels) def check(path, message_class, volume, microsoft_volume): bulk_level = "FAIL" if volume == "bulk" else "WARN" try: with open(path, "rb") as handle: message = email.message_from_binary_file(handle, policy=policy.default) except OSError as exc: print("check_message_headers: cannot read %s: %s" % (path, exc), file=sys.stderr) sys.exit(2) if not message.keys(): print("check_message_headers: %s has no headers; save the raw source of a received message" % path, file=sys.stderr) sys.exit(2) findings, info = [], [] def add(level, rule, name, detail): findings.append({"level": level, "rule": rule, "name": name, "detail": detail}) from_domain = domain_of(message.get("From", "")) return_domain = domain_of(message.get("Return-Path", "")) signatures = [parse_dkim_signature(v) for v in message.get_all("DKIM-Signature", [])] info.append("From domain: %s. Return-Path domain: %s." % (from_domain or "not found", return_domain or "none")) for sig in signatures: info.append("DKIM-Signature: d=%s s=%s (h= lists %d headers)" % (sig["d"], sig["s"], len(sig["h"]))) if not signatures: info.append("DKIM-Signature: none") ar_headers = message.get_all("Authentication-Results", []) passing_dkim = [] if not ar_headers: add("WARN", "DL05", "dmarc", "no Authentication-Results header. Save the message from the receiving" " mailbox, not the sent folder, then rerun.") else: block, ignored, merged = receiver_results(ar_headers) spf = merged["spf"][1][0] if "spf" in merged else None dkims = merged["dkim"][1] if "dkim" in merged else [] dmarc = merged["dmarc"][1][0] if "dmarc" in merged else None for r in dkims: if r["result"] == "pass": passing_dkim.append((dkim_domain(r), r["props"].get("header.s", ""))) dkim_pass = bool(passing_dkim) summary = [] if spf: summary.append("spf=%s (%s)" % (spf["result"], spf["props"].get("smtp.mailfrom", "?").split("@")[-1])) for r in dkims: summary.append("dkim=%s (%s)" % (r["result"], dkim_domain(r) or "?")) if dmarc: summary.append("dmarc=%s" % dmarc["result"]) results_text = ", ".join(summary) or "no results" if not block[0][0]: info.append("Receiver Authentication-Results: authserv-id missing (Microsoft writes none); confirm the" " file was saved from the receiving mailbox: %s" % results_text) elif len(block) == 1: info.append("Receiver Authentication-Results (authserv-id %s; confirm it is the receiving mailbox): %s" % (block[0][0], results_text)) else: info.append("Receiver Authentication-Results: %d headers merged from %s (one receiver; confirm it is" " the receiving mailbox): %s" % (len(block), ", ".join(a for a, _ in block), results_text)) if ignored: info.append("Ignored %d lower Authentication-Results header%s from %s: results added before the" " receiving mailbox prove nothing." % ( len(ignored), "" if len(ignored) == 1 else "s", ", ".join(a or "a server with no authserv-id" for a, _ in ignored))) if spf: spf_domain = spf["props"].get("smtp.mailfrom", "").split("@")[-1].lower() info.append("SPF domain %s aligned with From: %s." % ( spf_domain or "?", "yes" if aligned(spf_domain, from_domain) else "no")) # DL05 if not dmarc: add("WARN", "DL05", "dmarc", "no dmarc result in the receiver's Authentication-Results; check that the" " receiver evaluates DMARC and that a DMARC record exists for %s." % (from_domain or "the From domain")) elif dmarc["result"] == "pass": add("PASS", "DL05", "dmarc", "dmarc=pass for %s." % (from_domain or "the From domain")) elif volume == "bulk": add("FAIL", "DL05", "dmarc", "dmarc=%s for From domain %s. Bulk streams need dmarc=pass on the" " received message." % (dmarc["result"], from_domain or "?")) else: add("WARN", "DL05", "dmarc", "dmarc=%s for From domain %s. Below bulk volume this is not yet a" " provider requirement; fix it before any send day crosses a bulk threshold." % ( dmarc["result"], from_domain or "?")) # DL06 spf_ok = bool(spf and spf["result"] == "pass") if spf_ok and dkim_pass: add("PASS", "DL06", "microsoft-both-pass", "spf=pass and dkim=pass.") else: state = "spf=%s and dkim=%s." % (spf["result"] if spf else "missing", "pass" if dkim_pass else "no passing signature") if microsoft_volume == "bulk": add("FAIL", "DL06", "microsoft-both-pass", state + " Microsoft rejects mail from domains sending" " 5,000 or more a day to its consumer mailboxes unless both pass (550 5.7.515).") else: add("WARN", "DL06", "microsoft-both-pass", state + " Below Microsoft's 5,000 a day threshold (DL02)," " so not yet a rejection; fix it before the domain reaches 5,000 a day.") # DL04 if message_class == "marketing" and not signatures and not passing_dkim: add("FAIL", "DL04", "dkim", "no DKIM-Signature on a marketing message. Bulk sender rules need DKIM," " forwarding then breaks DMARC, and one-click unsubscribe needs a signature over both" " List-Unsubscribe headers.") elif ar_headers: if not signatures and passing_dkim: info.append("The receiver reports dkim=pass but the file has no DKIM-Signature header: save the full raw" " source from the receiving mailbox to check h=.") aligned_dkim = [d for d, _ in passing_dkim if aligned(d, from_domain)] if aligned_dkim: add("PASS", "DL04", "dkim", "passing DKIM signature aligned with %s (d=%s)." % ( from_domain, ", ".join(aligned_dkim))) else: add("WARN", "DL04", "dkim", "no passing DKIM signature aligns with %s (passing d=: %s). Forwarding" " breaks SPF, so DKIM alignment is what keeps DMARC passing. Alignment here is an" " approximation; dmarc= is the authority." % ( from_domain or "?", ", ".join(d for d, _ in passing_dkim) or "none")) # DL07 headers and body if message_class == "marketing": missing = [] lu_values = [unfold(v) for v in message.get_all("List-Unsubscribe", [])] uris = re.findall(r"<([^>]+)>", " ".join(lu_values)) https = [u for u in uris if u.lower().startswith("https://")] if not lu_values: missing.append("no List-Unsubscribe header") elif not https: missing.append("List-Unsubscribe has no HTTPS URI") post = unfold(message.get("List-Unsubscribe-Post", "")) if not post: missing.append("no List-Unsubscribe-Post header") elif post.lower() != "list-unsubscribe=one-click": missing.append("List-Unsubscribe-Post is '%s', not List-Unsubscribe=One-Click" % post) passing_domains = {d for d, _ in passing_dkim} if ar_headers: covering = [s for s in signatures if signature_passed(s, passing_domains)] which = "a passing DKIM signature (passing d=: %s)" % (", ".join(sorted(passing_domains)) or "none") else: covering = signatures which = "any DKIM signature (no Authentication-Results, so pass or fail is unknown)" for header in ("list-unsubscribe", "list-unsubscribe-post"): if not any(header in s["h"] for s in covering): missing.append("%s is not in h= of %s" % (header, which)) if missing: add(bulk_level, "DL07", "one-click-and-body-link", "one-click unsubscribe is not usable: %s. %s" % ( "; ".join(missing), "Bulk marketing streams need all parts (RFC 8058)." if volume == "bulk" else "A low-volume stream can run without it, but it becomes FAIL once the domain is bulk.")) else: add("PASS", "DL07", "one-click-and-body-link", "HTTPS List-Unsubscribe, List-Unsubscribe-Post" " One-Click, both headers signed by a passing DKIM signature. Test the POST without cookies.") if len(https) > 1: add("WARN", "DL07", "one-click-and-body-link", "List-Unsubscribe has %d HTTPS URIs; RFC 8058" " expects exactly one." % len(https)) view = body_view(message) if has_body(message) else None if view is None: info.append("NOT CHECKED DL07 body: no message body (a headers-only save). Save the full raw source" " to check the body link.") else: kind, links, text, labels = view where = "the HTML part" if kind == "html" else "the text part" if not links: add("FAIL", "DL07", "one-click-and-body-link", "no link in %s of the body. Gmail and Yahoo require" " a visible unsubscribe link in the body." % where) elif UNSUBSCRIBE_WORDS.search(labels): add("PASS", "DL07", "one-click-and-body-link", "unsubscribe wording on or beside a link in %s (words" " inside URLs not counted)." % where) elif PREFERENCE_WORDS.search(labels): add("WARN", "DL07", "one-click-and-body-link", "%s has a preferences link but no unsubscribe" " wording. Add a visible unsubscribe link." % where) elif set(links) & set(https): add("PASS", "DL07", "one-click-and-body-link", "a link in %s reuses the HTTPS List-Unsubscribe URL;" " the wording was not recognised, so confirm the label reads as unsubscribe in the message's" " language." % where) elif UNSUBSCRIBE_WORDS.search(text): add("WARN", "DL07", "one-click-and-body-link", "%s has unsubscribe wording, but not on or within %d" " characters of a link in the same block. Read the body: FAIL when no link unsubscribes or opens preferences." % (where, NEAR_LINK)) else: add("WARN", "DL07", "one-click-and-body-link", "no unsubscribe wording recognised in %s. The check" " reads English, German, French, Spanish, Dutch, Italian and Portuguese wording and the" " List-Unsubscribe URL only: read the body and rate the link by hand (FAIL when no link" " unsubscribes or opens preferences)." % where) else: info.append("NOT CHECKED DL07: transactional class. One-click applies to marketing and subscribed" " messages; confirm the content carries no promotion (PC07).") # DL08 received = [unfold(v) for v in message.get_all("Received", [])] tls = [r for r in received if TLS_PATTERN.search(r)] if tls: hint = TLS_PATTERN.search(tls[0]).group(0) add("PASS", "DL08", "transport-and-dns", "a Received header shows TLS (%s)." % hint) else: add("WARN", "DL08", "transport-and-dns", "no Received header shows TLS (ESMTPS or a TLS version).") return {"file": path, "class": message_class, "volume": volume, "microsoft_volume": microsoft_volume, "info": info, "findings": findings} def main(): parser = argparse.ArgumentParser(description="Check saved raw message sources.") parser.add_argument("files", nargs="+") parser.add_argument("--class", dest="message_class", choices=["marketing", "transactional"], default="marketing") parser.add_argument("--volume", choices=["bulk", "low"], default="bulk", help="bulk (default) applies the bulk sender rules; low reports them as WARN") parser.add_argument("--microsoft-volume", choices=["bulk", "low"], default="bulk", help="bulk (default) when the From domain sends 5,000 or more a day to Microsoft consumer" " mailboxes; low reports DL06 as WARN") parser.add_argument("--json", action="store_true") args = parser.parse_args() reports = [check(path, args.message_class, args.volume, args.microsoft_volume) for path in args.files] totals = {"FAIL": 0, "WARN": 0, "PASS": 0} lines = [] for report in reports: counts = {level: sum(1 for f in report["findings"] if f["level"] == level) for level in totals} for level in totals: totals[level] += counts[level] report["counts"] = counts lines.append("== %s (class %s, volume %s, Microsoft volume %s)" % ( report["file"], report["class"], report["volume"], report["microsoft_volume"])) lines.extend(report["info"]) for f in report["findings"]: lines.append("%s %s %s: %s" % (f["level"], f["rule"], f["name"], f["detail"])) lines.append("Result: %d FAIL, %d WARN, %d PASS" % (counts["FAIL"], counts["WARN"], counts["PASS"])) lines.append("") summary = "%d file%s: %d FAIL, %d WARN, %d PASS. Nothing was changed." % ( len(reports), "" if len(reports) == 1 else "s", totals["FAIL"], totals["WARN"], totals["PASS"]) if args.json: print(json.dumps({"files": reports, "totals": totals, "summary": summary}, indent=2)) else: lines.append(summary) print("\n".join(lines)) sys.exit(1 if totals["FAIL"] else 0) if __name__ == "__main__": main()