Building a DMARC Report Parser and Dashboard in Python

DMARC aggregate reports land in your mailbox every day as gzip-compressed XML attachments. Most teams configure DMARC once and forget it exists. That's a problem: those reports tell you who's sending email on behalf of your domain, what fraction is passing authentication, and whether someone is actively spoofing you.

This article walks through building a parser and terminal dashboard in Python — no third-party libraries beyond the standard library.

What's Inside a DMARC Aggregate Report

DMARC aggregate reports follow RFC 7489. Each report is a ZIP or gzip archive wrapping an XML file. The relevant structure:

<feedback>
  <report_metadata>
    <org_name>Google</org_name>
    <date_range>
      <begin>1720742400</begin>
      <end>1720828799</end>
    </date_range>
  </report_metadata>
  <policy_published>
    <domain>example.com</domain>
    <p>reject</p>
  </policy_published>
  <record>
    <row>
      <source_ip>209.85.220.41</source_ip>
      <count>47</count>
      <policy_evaluated>
        <disposition>none</disposition>
        <dkim>pass</dkim>
        <spf>pass</spf>
      </policy_evaluated>
    </row>
    <identifiers>
      <header_from>example.com</header_from>
    </identifiers>
  </record>
</feedback>

Each <record> represents a batch of messages from a single source IP. You want IP, count, SPF result, DKIM result, and disposition for every record.

Building the Parser

Start with a parser that handles both ZIP and gzip formats — Google, Microsoft, and Yahoo all use slightly different packaging.

import gzip
import zipfile
import xml.etree.ElementTree as ET
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional
import socket
from datetime import datetime


@dataclass
class DmarcRecord:
    source_ip: str
    count: int
    disposition: str
    dkim: str
    spf: str
    header_from: str
    hostname: Optional[str] = None


@dataclass
class DmarcReport:
    org_name: str
    domain: str
    begin: datetime
    end: datetime
    policy: str
    records: list[DmarcRecord] = field(default_factory=list)


def _extract_xml(path: Path) -> str:
    if path.suffix == ".zip":
        with zipfile.ZipFile(path) as zf:
            xml_name = next(n for n in zf.namelist() if n.endswith(".xml"))
            return zf.read(xml_name).decode("utf-8")
    elif path.suffix in (".gz", ".gzip"):
        with gzip.open(path, "rb") as f:
            return f.read().decode("utf-8")
    else:
        return path.read_text()


def parse_report(path: Path, resolve_hostnames: bool = False) -> DmarcReport:
    xml_content = _extract_xml(path)
    root = ET.fromstring(xml_content)

    meta = root.find("report_metadata")
    policy = root.find("policy_published")

    report = DmarcReport(
        org_name=meta.findtext("org_name", "unknown"),
        domain=policy.findtext("domain", "unknown"),
        begin=datetime.utcfromtimestamp(int(meta.findtext("date_range/begin", "0"))),
        end=datetime.utcfromtimestamp(int(meta.findtext("date_range/end", "0"))),
        policy=policy.findtext("p", "none"),
    )

    for record in root.findall("record"):
        row = record.find("row")
        idents = record.find("identifiers")
        evaluated = row.find("policy_evaluated")

        ip = row.findtext("source_ip", "")
        hostname = None
        if resolve_hostnames and ip:
            try:
                hostname = socket.gethostbyaddr(ip)[0]
            except socket.herror:
                hostname = None

        report.records.append(
            DmarcRecord(
                source_ip=ip,
                count=int(row.findtext("count", "0")),
                disposition=evaluated.findtext("disposition", "none"),
                dkim=evaluated.findtext("dkim", "unknown"),
                spf=evaluated.findtext("spf", "unknown"),
                header_from=idents.findtext("header_from", ""),
                hostname=hostname,
            )
        )

    return report

The resolve_hostnames flag is off by default. Reverse DNS adds latency when processing hundreds of reports in a batch.

Aggregating Reports and Building the Dashboard

A single report covers one day and one reporting organization. You'll receive several per domain per day. This function aggregates them into a unified pass/fail summary:

from collections import defaultdict


def aggregate_reports(paths: list[Path]) -> dict:
    totals: dict = defaultdict(lambda: {"pass": 0, "fail": 0, "ips": defaultdict(int)})

    for path in paths:
        try:
            report = parse_report(path)
        except Exception as e:
            print(f"[WARN] Could not parse {path.name}: {e}")
            continue

        for rec in report.records:
            domain = report.domain
            both_pass = rec.dkim == "pass" and rec.spf == "pass"

            if both_pass:
                totals[domain]["pass"] += rec.count
            else:
                totals[domain]["fail"] += rec.count

            totals[domain]["ips"][rec.source_ip] += rec.count

    return dict(totals)


def print_dashboard(aggregated: dict) -> None:
    print(f"\n{'Domain':<30} {'Pass':>8} {'Fail':>8} {'Pass%':>8}  Top Sender IP")
    print("-" * 75)
    for domain, stats in sorted(aggregated.items()):
        total = stats["pass"] + stats["fail"]
        pct = (stats["pass"] / total * 100) if total else 0.0
        top_ip = max(stats["ips"], key=stats["ips"].get, default="—")
        flag = "  ⚠ low pass rate" if pct < 95 else ""
        print(f"{domain:<30} {stats['pass']:>8} {stats['fail']:>8} {pct:>7.1f}%  {top_ip}{flag}")
    print()

The ⚠ threshold at 95% is a starting point. Adjust to your domain's established baseline once you have a few weeks of data.

Pulling Reports Directly from Your Inbox

DMARC reports arrive as email attachments. Point them to a dedicated folder in your mailbox via an email filter, then pull them with this IMAP snippet:

import imaplib
import email
import tempfile
import os

IMAP_HOST = os.environ["IMAP_HOST"]
IMAP_USER = os.environ["IMAP_USER"]
IMAP_PASS = os.environ["IMAP_PASS"]
IMAP_FOLDER = os.environ.get("IMAP_FOLDER", "DMARC")


def fetch_attachments(dest_dir: Path) -> list[Path]:
    saved: list[Path] = []
    with imaplib.IMAP4_SSL(IMAP_HOST) as imap:
        imap.login(IMAP_USER, IMAP_PASS)
        imap.select(IMAP_FOLDER)
        _, data = imap.search(None, "UNSEEN")
        for uid in data[0].split():
            _, msg_data = imap.fetch(uid, "(RFC822)")
            msg = email.message_from_bytes(msg_data[0][1])
            for part in msg.walk():
                fn = part.get_filename()
                if fn and (fn.endswith(".zip") or fn.endswith(".gz")):
                    dest = dest_dir / fn
                    dest.write_bytes(part.get_payload(decode=True))
                    saved.append(dest)
    return saved


if __name__ == "__main__":
    with tempfile.TemporaryDirectory() as tmp:
        tmpdir = Path(tmp)
        attachments = fetch_attachments(tmpdir)
        print(f"Fetched {len(attachments)} report(s).")
        aggregated = aggregate_reports(attachments)
        print_dashboard(aggregated)

Four environment variables, one cron job, and you have nightly DMARC visibility with zero external dependencies.

What Patterns Actually Matter

Once the parser is running, focus on these signals:

Sudden fail spike from a known IP. A legitimate email service provider that previously passed DKIM and SPF starts failing. This almost always means a key rotation or header misconfiguration on their side — your DMARC report catches it before customers see quarantined mail.

Unknown source IPs at volume. An IP you don't recognize is sending 500+ messages per day claiming your domain as sender. That's either shadow IT using an unlisted ESP, or active spoofing at scale. Either way, you need to know.

Disposition mismatch. Your policy is p=reject but messages with both DKIM and SPF failures show disposition none. This usually means the reporting organization applied a local override (forwarding scenarios are common). Understand the override before assuming your policy is enforced end-to-end.

Gradual policy ramp-up. Most production deployments start at p=none, move to p=quarantine, then p=reject over 60–90 days. The dashboard lets you track the fail rate trend and move confidently between stages.

For organizations setting up email authentication from scratch, the email and domain security hardening checklist from AYI NEDJIMI Consultants covers SPF flattening, DKIM key sizes, DMARC ramp-up timelines, and a monitoring cadence you can adapt to this parser.

The Takeaway

DMARC aggregate reports are free, daily, and machine-readable. Most teams configure the DNS record and never look at the data again. A 250-line Python script built entirely on the standard library turns that ignored stream into a daily authentication health check: who's sending on your behalf, what's failing, and whether someone is spoofing your domain.

Run it as a cron job. Pipe the output to a Slack webhook or write the fail counts to a time-series metric. The infrastructure footprint is essentially zero.

---

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.