Validate and Auto-Correct DMARC Records Across Domains

Before you run this

This guide builds a Python tool that looks up the _dmarc.<domain> TXT record for one or more domains, parses it, and reports where the DMARC policy is missing, malformed, or weaker than the baseline you set. In its default form it only reads DNS — it changes nothing. The optional second half pushes a corrected record through a DNS provider API (I use Cloudflare as the concrete example).

• Privileges: The validator needs no root. Installing the Python packages into a virtualenv needs no root either. The correction step needs a scoped DNS API token for your provider — not root, not your account password.
• This is a change to mail handling, not a cosmetic tweak. Publishing p=reject while your SPF and DKIM are not fully aligned will cause receiving servers to reject legitimate mail. DNS also caches: a bad record lingers for the TTL. Treat every write as production-affecting.
• Test first. Read the script before running it. Run the validator against a throwaway or staging domain first, and if you use the correction step, point it at one non-critical domain with --apply off (dry-run) before you ever let it write.
• Save what you overwrite. The correction step prints the existing record before changing anything so you can restore it. DMARC records are not "irreversible", but a wrong p= value silently discards mail until you notice and the TTL expires — so treat it as if it were.

Assumptions for this guide: Ubuntu 22.04, Python 3.10+, packages installed in a virtualenv with pip. On RHEL/Alma the only difference is python3 -m venv needing the python3 package group; the Python code is identical.

What a valid DMARC record looks like

A DMARC record is a single DNS TXT record at _dmarc.<domain>. The pieces that matter for validation:

• v=DMARC1 — required, must be first.
• p= — required policy: none, quarantine, or reject.
• rua= — where aggregate reports go (a mailto: URI). Optional in the spec, but you almost always want it.
• pct= — integer 0–100.
• sp=, adkim=, aspf=, fo=, ruf=, ri= — optional.

I'm deliberately validating the tags I'm certain of. For the full tag list and exact semantics, see the DMARC specification (RFC 7489). Don't let the script "correct" tags whose meaning you haven't confirmed.

Install the dependencies

python3 -m venv ~/dmarc-venv
source ~/dmarc-venv/bin/activate
pip install dnspython requests   # dnspython for lookups, requests for the API step

The validator

Save this as dmarc_check.py. It reads DNS and reports; it writes nothing.

#!/usr/bin/env python3
"""Validate DMARC records for one or more domains and suggest corrections.

Read-only: this queries DNS and prints findings. It does not change anything.
"""

import sys
import argparse
import dns.resolver   # from the 'dnspython' package

# --- Your baseline. Edit these to the policy every domain should meet. ---
REQUIRED_POLICY = "reject"                    # the p= value you want
REQUIRED_RUA    = "mailto:dmarc@example.com"  # aggregate report address
VALID_POLICIES  = {"none", "quarantine", "reject"}


def fetch_dmarc(domain):
    """Return the DMARC TXT string for a domain, or None if absent."""
    name = f"_dmarc.{domain}"
    try:
        answers = dns.resolver.resolve(name, "TXT")
    except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer):
        return None
    for rdata in answers:
        # A TXT record may arrive as several quoted chunks; join them.
        txt = "".join(s.decode() for s in rdata.strings)
        if txt.lower().startswith("v=dmarc1"):
            return txt
    return None


def parse_tags(record):
    """Turn 'v=DMARC1; p=none; ...' into a dict of tag -> value."""
    tags = {}
    for part in record.split(";"):
        part = part.strip()
        if "=" not in part:
            continue
        key, _, value = part.partition("=")
        tags[key.strip().lower()] = value.strip()
    return tags


def validate(record):
    """Return (list_of_problems, suggested_record)."""
    problems = []
    if record is None:
        return (["no DMARC record found"],
                f"v=DMARC1; p={REQUIRED_POLICY}; rua={REQUIRED_RUA}")

    tags = parse_tags(record)

    if tags.get("v") != "DMARC1":
        problems.append("record does not start with v=DMARC1")

    p = tags.get("p")
    if p is None:
        problems.append("missing required p= tag")
    elif p not in VALID_POLICIES:
        problems.append(f"invalid p= value: {p!r}")
    elif p != REQUIRED_POLICY:
        problems.append(f"p={p}, expected p={REQUIRED_POLICY}")

    if "rua" not in tags:
        problems.append("no rua= aggregate report address")

    pct = tags.get("pct")
    if pct is not None and (not pct.isdigit() or not 0 <= int(pct) <= 100):
        problems.append(f"pct out of range: {pct!r}")

    # Build a corrected record from existing tags, forcing v/p/rua.
    fixed = dict(tags)
    fixed["v"] = "DMARC1"
    fixed["p"] = REQUIRED_POLICY
    fixed.setdefault("rua", REQUIRED_RUA)
    ordered = ["v", "p"] + [k for k in fixed if k not in ("v", "p")]
    suggested = "; ".join(f"{k}={fixed[k]}" for k in ordered)
    return problems, suggested


def main():
    ap = argparse.ArgumentParser(description="Validate DMARC records.")
    ap.add_argument("domains", nargs="+", help="domains to check")
    args = ap.parse_args()

    exit_code = 0
    for domain in args.domains:
        record = fetch_dmarc(domain)
        problems, suggested = validate(record)
        if problems:
            exit_code = 1
            print(f"[FAIL] {domain}")
            print(f"       current:   {record}")
            for prob in problems:
                print(f"       - {prob}")
            print(f"       suggested: {suggested}")
        else:
            print(f"[OK]   {domain}: {record}")
    sys.exit(exit_code)


if __name__ == "__main__":
    main()

Run it against your domains — replace these with your own:

python3 dmarc_check.py example.com example.org

To check a list from a file:

xargs python3 dmarc_check.py < domains.txt

It exits non-zero if any domain fails, so it drops straight into a cron job or CI pipeline as a monitor. That alone — validate and alert — is the safe, high-value use. The correction step below is optional and needs more care.

Auto-correcting through your DNS provider

There is no universal "write DNS" API — corrections go through whichever provider hosts each zone. Below is the concrete pattern for Cloudflare. The field names (type, name, content, ttl) and the Bearer-token auth are Cloudflare's documented DNS-records API; confirm them against Cloudflare's current API reference for "DNS Records for a Zone" before you rely on this, and create an API token scoped to DNS Edit on the specific zone, nothing broader.

import requests

CF_TOKEN = "cloudflare-api-token-goes-here"     # DNS:Edit token, not a password
CF_ZONE  = "cloudflare-zone-id-for-example-com" # the zone ID

def cf_set_dmarc(domain, new_record, apply=False):
    base = f"https://api.cloudflare.com/client/v4/zones/{CF_ZONE}/dns_records"
    headers = {"Authorization": f"Bearer {CF_TOKEN}"}
    name = f"_dmarc.{domain}"

    # Find the existing TXT record (so we can update, not duplicate).
    r = requests.get(base, headers=headers,
                     params={"type": "TXT", "name": name})
    r.raise_for_status()
    existing = r.json()["result"]

    if existing:
        print(f"BACKUP  {name} was: {existing[0]['content']}")  # save this!

    if not apply:
        print(f"DRY-RUN would set {name} -> {new_record}")
        return

    body = {"type": "TXT", "name": name, "content": new_record, "ttl": 3600}
    if existing:
        resp = requests.put(f"{base}/{existing[0]['id']}",
                            headers=headers, json=body)
    else:
        resp = requests.post(base, headers=headers, json=body)
    resp.raise_for_status()
    print(f"UPDATED {name}")

Call it in dry-run first (apply=False), read the BACKUP line, and only flip apply=True once you're satisfied. Keep the printed backup somewhere — that's your rollback source. A normal DMARC record is well under the 255-character TXT limit, so chunk-splitting isn't a concern here.

Verify it worked

Query the record directly, bypassing your local cache by asking a public resolver:

dig @1.1.1.1 +short TXT _dmarc.example.com

Then re-run the validator — it should now report [OK]:

python3 dmarc_check.py example.com

Remember the record can take up to its TTL to appear everywhere; a stale answer from one resolver isn't a failed change.

Undo / rollback

To revert, run cf_set_dmarc again with the original string from the BACKUP line as new_record, or restore the previous value in your provider's dashboard. If the domain never had a DMARC record and you want to remove the one you added, delete the TXT record for _dmarc.<domain> in the provider UI (or via the API's delete endpoint — check the docs for the exact call). After any rollback, confirm with the same dig command above.

If you take one thing from this: run the validator freely and often, but let the write step earn your trust on a test domain before it touches anything that carries real mail.