DMARC Analyser

import os
import zipfile
import gzip
import shutil
import tempfile
from lxml import etree
from collections import Counter

# Directory containing your DMARC reports
input_dir = "path/to/your/dmar/records"

temp_dir = tempfile.mkdtemp()
xml_files = []

def extract_files():
    for root, _, files in os.walk(input_dir):
        for file in files:
            file_path = os.path.join(root, file)

            if file.endswith('.zip'):
                with zipfile.ZipFile(file_path, 'r') as zip_ref:
                    zip_ref.extractall(temp_dir)

            elif file.endswith('.gz'):
                extracted_path = os.path.join(temp_dir, os.path.splitext(file)[0])
                with gzip.open(file_path, 'rb') as f_in:
                    with open(extracted_path, 'wb') as f_out:
                        shutil.copyfileobj(f_in, f_out)

            elif file.endswith('.xml'):
                shutil.copy(file_path, temp_dir)

def parse_and_aggregate():
    alignment_counter = Counter()
    total_count = 0

    for fname in os.listdir(temp_dir):
        if fname.endswith('.xml'):
            path = os.path.join(temp_dir, fname)
            try:
                tree = etree.parse(path)
                records = tree.findall(".//record")
                for record in records:
                    count = int(record.findtext("row/count", default="1"))
                    spf_result = record.findtext("row/policy_evaluated/spf")
                    spf_aligned = record.findtext("row/policy_evaluated/spf") == "pass"
                    dkim_result = record.findtext("row/policy_evaluated/dkim")
                    dkim_aligned = record.findtext("row/policy_evaluated/dkim") == "pass"

                    total_count += count

                    if spf_aligned and dkim_aligned:
                        alignment_counter["Fully aligned"] += count
                    elif spf_aligned or dkim_aligned:
                        alignment_counter["Partially aligned"] += count
                    else:
                        alignment_counter["Misaligned"] += count

            except Exception as e:
                print(f"Error parsing {fname}: {e}")

    return alignment_counter, total_count

# Run
extract_files()
results, total = parse_and_aggregate()

# Summary
print("\nDMARC Alignment Summary:")
for category in ["Fully aligned", "Partially aligned", "Misaligned"]:
    count = results.get(category, 0)
    percent = (count / total) * 100 if total else 0
    print(f"{category}: {count} messages ({percent:.2f}%)")

print(f"Total messages analyzed: {total}")

def get_misaligned_details():
    details = []
    for fname in os.listdir(temp_dir):
        if fname.endswith('.xml'):
            path = os.path.join(temp_dir, fname)
            try:
                tree = etree.parse(path)
                records = tree.findall(".//record")
                for record in records:
                    count = int(record.findtext("row/count", default="1"))

                    spf_result = record.findtext("row/policy_evaluated/spf")
                    dkim_result = record.findtext("row/policy_evaluated/dkim")

                    if spf_result != "pass" and dkim_result != "pass":
                        source_ip = record.findtext("row/source_ip")
                        disposition = record.findtext("row/policy_evaluated/disposition")
                        envelope_from = record.findtext("identifiers/envelope_from")
                        header_from = record.findtext("identifiers/header_from")
                        org_name = tree.findtext(".//report_metadata/org_name")
                        begin = int(tree.findtext(".//report_metadata/date_range/begin"))
                        end = int(tree.findtext(".//report_metadata/date_range/end"))

                        details.append({
                            "Count": count,
                            "Source IP": source_ip,
                            "Header From": header_from,
                            "Envelope From": envelope_from,
                            "SPF Result": spf_result,
                            "DKIM Result": dkim_result,
                            "DMARC Disposition": disposition,
                            "Sending Org": org_name,
                            "Date Range": f"{begin} to {end}",
                            "File": fname
                        })
            except Exception as e:
                print(f"Error parsing {fname}: {e}")
    return details


misaligned_messages = get_misaligned_details()

# Example: Print summary
if misaligned_messages:
    for msg in misaligned_messages:
        print(f"{msg['Date Range']} | {msg['Source IP']} | {msg['Header From']} | "
              f"SPF: {msg['SPF Result']} | DKIM: {msg['DKIM Result']} | Count: {msg['Count']}")
else:
    print("\nNo misaligned messages!")

How to use

• launch jupyter in your venv of choice
• update input_dirto point to the location of your files
• execute the code ans see results

Example output

DMARC Alignment Summary:
Fully aligned: 1352 messages (97.83%)
Partially aligned: 20 messages (1.45%)
Misaligned: 10 messages (0.72%)
Total messages analyzed: 1382
1760486400 to 1760572800 | 52.212.19.177 | yourdomain.com | SPF: fail | DKIM: fail | Count: 2
1760486400 to 1760572800 | 52.212.19.177 | yourdomain.com | SPF: fail | DKIM: fail | Count: 8

Technical notes

Prerequisites

• Python installed locally
• jupyter or equivalent installed to run the code (or convert to standalone script)
• Create a email group for your domain, e.g. dmarc-reports@yourdomain.com
• Add TXT recod to DNS with Name: _dmarc, Value: v=DMARC1; p=quarantine; pct=100; rua=mailto:dmarc-reports@yourdomain.com
• wait for a few days to compile a list of emails with the reports
• download the .gz/.zip files to a local directory - e.g. ~/Desktop/dmarc