WebMail

DMARC Processing

• Purpose: Scan DMARC mails and process accordingly:
  • If something needs to be done, sends a mail to admin with a notification.
  • If nothing critical, just dumps it from .DMARC to trash?

admin
├── .DMARC
│   ├── cur
│   ├── new
│   └── tmp
├── cur
├── new
├── tmp
...

Roundcube Sieve:

Filter name: DMARC
Filter enabled
matching any of the following rules
Subject contains Report Domain: eawf.com
From contains DMARC
Actions
Move message to DMARC
Stop evaluating rules

Root crontab:

-rwxr-xr-x 1 root root 4401 Apr 18 22:57 /usr/local/sbin/dmarc_report_scan.py

Root Crontab Configuration

0 8,20 * * * /usr/local/sbin/dmarc_report_scan.py --maildir /var/vmail/eawf.com/admin/.DMARC --trusted-ip 50.28.104.213 --report-file /var/log/dmarc-scan-report.txt --delete-processed >/dev/null 2>&1

Python Code

#!/usr/bin/env python3

import argparse
import os
import zipfile
import xml.etree.ElementTree as ET
from email import policy
from email.parser import BytesParser
import hashlib

def parse_args():
    p = argparse.ArgumentParser(description="Scan DMARC aggregate reports from a Maildir.")
    p.add_argument("--maildir", required=True)
    p.add_argument("--trusted-ip", action="append", default=[])
    p.add_argument("--state-dir", default="/var/lib/dmarc-report-scan")
    p.add_argument("--report-file")
    p.add_argument("--stdout", action="store_true")
    p.add_argument("--delete-processed", action="store_true")
    p.add_argument("--reset-seen", action="store_true")
    return p.parse_args()

def load_seen(state_dir):
    os.makedirs(state_dir, exist_ok=True)
    seen_file = os.path.join(state_dir, "seen.txt")
    if not os.path.exists(seen_file):
        return set()
    with open(seen_file) as f:
        return set(line.strip() for line in f)

def save_seen(state_dir, seen):
    seen_file = os.path.join(state_dir, "seen.txt")
    with open(seen_file, "w") as f:
        for s in seen:
            f.write(s + "\n")

def hash_msg(path):
    return hashlib.sha256(path.encode()).hexdigest()

def is_trusted(ip, trusted_list):
    return ip in trusted_list

def parse_xml(data):
    root = ET.fromstring(data)
    results = []
    for rec in root.findall(".//record"):
        ip = rec.findtext(".//source_ip")
        count = int(rec.findtext(".//count"))
        spf = rec.findtext(".//policy_evaluated/spf")
        dkim = rec.findtext(".//policy_evaluated/dkim")
        results.append((ip, count, spf, dkim))
    return results

def process_mail(path, trusted_ips):
    alerts = []
    noise = []
    with open(path, "rb") as f:
        msg = BytesParser(policy=policy.default).parse(f)

    found = False

    for part in msg.walk():
        if part.get_content_type() in ("application/zip", "application/gzip", "application/octet-stream"):
            payload = part.get_payload(decode=True)
            if not payload:
                continue
            try:
                with zipfile.ZipFile(io.BytesIO(payload)) as z:
                    for name in z.namelist():
                        xml_data = z.read(name)
                        for ip, count, spf, dkim in parse_xml(xml_data):
                            found = True
                            if is_trusted(ip, trusted_ips):
                                if spf != "pass" or dkim != "pass":
                                    alerts.append(f"ALERT: trusted sender failed {ip} ({count})")
                            else:
                                if spf == "pass" or dkim == "pass":
                                    alerts.append(f"ALERT: untrusted sender passed {ip} ({count})")
                                else:
                                    noise.append(f"Noise: untrusted sender failed {ip} ({count})")
            except:
                pass

    return found, alerts, noise

def main():
    args = parse_args()

    seen = set() if args.reset_seen else load_seen(args.state_dir)

    report_lines = []
    alerts_found = False

    for sub in ("cur", "new"):
        d = os.path.join(args.maildir, sub)
        if not os.path.isdir(d):
            continue

        for fname in os.listdir(d):
            fpath = os.path.join(d, fname)
            h = hash_msg(fpath)

            if h in seen:
                continue

            found, alerts, noise = process_mail(fpath, args.trusted_ip)

            seen.add(h)

            if found and args.delete_processed:
                try:
                    os.remove(fpath)
                except:
                    pass

            for a in alerts:
                report_lines.append(a)
                alerts_found = True

    save_seen(args.state_dir, seen)

    if not report_lines:
        report_lines.append("No issues detected.")

    report = "\n".join(report_lines)

    if args.report_file:
        with open(args.report_file, "w") as f:
            f.write(report + "\n")

    if args.stdout:
        print(report)

    if alerts_found:
        os.system(f'''(
echo "From: dmarc-monitor@eawf.com"
echo "To: admin@eawf.com"
echo "Subject: DMARC ALERT: action needed"
echo "X-EAWF-Notice: DMARC-SCAN"
echo
cat {args.report_file}
) | /usr/sbin/sendmail -t''')

if __name__ == "__main__":
    import io
    main()