From cd6fa25f5d36ec81862023cf556ae521d65a2b78 Mon Sep 17 00:00:00 2001 From: Christian Krakau-Louis Date: Sat, 23 May 2026 13:58:32 +0200 Subject: [PATCH] feat: add DMARC aggregate parser compatibility --- backend/app/services/dmarc_parser.py | 260 ++++++++++++++++++++----- backend/app/tests/test_dmarc_parser.py | 156 +++++++++++++++ docs/milestones.md | 7 +- 3 files changed, 368 insertions(+), 55 deletions(-) diff --git a/backend/app/services/dmarc_parser.py b/backend/app/services/dmarc_parser.py index df8327a..b3672c7 100644 --- a/backend/app/services/dmarc_parser.py +++ b/backend/app/services/dmarc_parser.py @@ -3,7 +3,7 @@ import io import logging import zipfile from datetime import datetime -from typing import Any, Dict, Optional +from typing import Any, Dict, List, Optional import defusedxml.ElementTree as ET @@ -136,76 +136,231 @@ class DMARCParser: for child in el: DMARCParser._strip_namespace(child) + @staticmethod + def _namespace(tag: str) -> str: + """Return the XML namespace from an ElementTree tag.""" + return tag[1:].split("}", 1)[0] if tag.startswith("{") and "}" in tag else "" + + @staticmethod + def _safe_int(value: Any, default: int = 0) -> int: + """Parse integer fields without failing the entire report on bad optional data.""" + try: + return int(value) + except (TypeError, ValueError): + return default + + @staticmethod + def _text(parent, name: str, default: str = "") -> str: + """Return stripped child text for a parsed XML element.""" + return (parent.findtext(name, default) or default).strip() + + @staticmethod + def _parse_text_list(parent, name: str) -> List[str]: + """Return all non-empty child text values for repeated simple elements.""" + return [ + text + for text in (DMARCParser._text(child, ".") for child in parent.findall(name)) + if text + ] + + @staticmethod + def _collect_extension_values(parent) -> Dict[str, Any]: + """Capture namespaced extension values without coupling to vendor-specific schemas.""" + values: Dict[str, Any] = {} + for child in list(parent): + key = child.tag + if len(child): + values[key] = DMARCParser._collect_extension_values(child) + else: + values[key] = (child.text or "").strip() + return values + + @staticmethod + def _extension_value(element) -> Any: + """Return a scalar or nested mapping for a vendor extension element.""" + if len(element): + return DMARCParser._collect_extension_values(element) + return (element.text or "").strip() + + @staticmethod + def _detect_variant(root, xml_namespace: str) -> dict: + """Identify the aggregate report format variant for debugging/import history.""" + version = DMARCParser._text(root, "version", "1.0") + has_rfc9990_fields = any( + root.find(path) is not None + for path in ( + "report_metadata/generator", + "policy_published/discovery_method", + "policy_published/np", + "policy_published/testing", + "record/identifiers/envelope_to", + ) + ) + if xml_namespace or has_rfc9990_fields: + variant = "rfc9990" + else: + variant = "rfc7489-compatible" + return { + "variant": variant, + "schema_version": version, + "xml_namespace": xml_namespace, + } + @staticmethod def _parse_metadata(root) -> dict: """Parse the report_metadata section of a DMARC XML report.""" report: dict = {} metadata = root.find("report_metadata") if metadata is not None: - report["report_id"] = metadata.findtext("report_id", "") - report["org_name"] = metadata.findtext("org_name", "") - report["email"] = metadata.findtext("email", "") + report["report_id"] = DMARCParser._text(metadata, "report_id") + report["org_name"] = DMARCParser._text(metadata, "org_name") + report["email"] = DMARCParser._text(metadata, "email") + report["extra_contact_info"] = DMARCParser._text(metadata, "extra_contact_info") + report["generator"] = DMARCParser._text(metadata, "generator") + errors = DMARCParser._parse_text_list(metadata, "error") + if errors: + report["errors"] = errors date_range = metadata.find("date_range") if date_range is not None: - begin_ts = int(date_range.findtext("begin", 0)) - end_ts = int(date_range.findtext("end", 0)) + begin_ts = DMARCParser._safe_int(date_range.findtext("begin", 0)) + end_ts = DMARCParser._safe_int(date_range.findtext("end", 0)) report["begin_date"] = datetime.fromtimestamp(begin_ts).isoformat() report["end_date"] = datetime.fromtimestamp(end_ts).isoformat() report["begin_timestamp"] = begin_ts report["end_timestamp"] = end_ts return report + @staticmethod + def _parse_policy(root) -> dict: + """Parse policy_published, including RFC 9990 optional fields.""" + policy = root.find("policy_published") + if policy is None: + return {} + parsed = { + "domain": DMARCParser._text(policy, "domain"), + "policy": { + "p": DMARCParser._text(policy, "p", "none"), + "sp": DMARCParser._text(policy, "sp"), + "pct": DMARCParser._text(policy, "pct", "100"), + "np": DMARCParser._text(policy, "np"), + "fo": DMARCParser._text(policy, "fo"), + "adkim": DMARCParser._text(policy, "adkim"), + "aspf": DMARCParser._text(policy, "aspf"), + "testing": DMARCParser._text(policy, "testing"), + "discovery_method": DMARCParser._text(policy, "discovery_method"), + }, + } + parsed["policy"] = {key: value for key, value in parsed["policy"].items() if value} + parsed["policy"].setdefault("pct", "100") + return parsed + + @staticmethod + def _parse_policy_reasons(policy_evaluated) -> List[dict]: + """Parse policy_evaluated/reason override data.""" + reasons = [] + for reason in policy_evaluated.findall("reason"): + parsed = { + "type": DMARCParser._text(reason, "type"), + "comment": DMARCParser._text(reason, "comment"), + } + if parsed["type"] or parsed["comment"]: + reasons.append(parsed) + return reasons + + @staticmethod + def _parse_row(record_elem) -> dict: + """Parse the record row and policy_evaluated section.""" + parsed: dict = {} + row = record_elem.find("row") + if row is None: + return parsed + + parsed["source_ip"] = DMARCParser._text(row, "source_ip") + parsed["count"] = DMARCParser._safe_int(row.findtext("count", 0)) + policy_evaluated = row.find("policy_evaluated") + if policy_evaluated is None: + return parsed + + parsed["disposition"] = DMARCParser._text(policy_evaluated, "disposition", "none") + parsed["dkim_result"] = DMARCParser._text(policy_evaluated, "dkim").lower() + parsed["spf_result"] = DMARCParser._text(policy_evaluated, "spf").lower() + reasons = DMARCParser._parse_policy_reasons(policy_evaluated) + if reasons: + parsed["policy_override_reasons"] = reasons + return parsed + + @staticmethod + def _parse_identifiers(record_elem) -> dict: + """Parse identifier fields used for aggregate policy evaluation.""" + parsed: dict = {} + identifiers = record_elem.find("identifiers") + if identifiers is None: + return parsed + parsed["header_from"] = DMARCParser._text(identifiers, "header_from") + parsed["envelope_from"] = DMARCParser._text(identifiers, "envelope_from") + parsed["envelope_to"] = DMARCParser._text(identifiers, "envelope_to") + return parsed + + @staticmethod + def _parse_auth_results(record_elem) -> dict: + """Parse uninterpreted DKIM/SPF authentication results.""" + parsed: dict = {} + auth_results = record_elem.find("auth_results") + if auth_results is None: + return parsed + + spf_entries = [ + { + "domain": DMARCParser._text(spf, "domain"), + "scope": DMARCParser._text(spf, "scope"), + "result": DMARCParser._text(spf, "result").lower(), + "human_result": DMARCParser._text(spf, "human_result"), + } + for spf in auth_results.findall("spf") + ] + if spf_entries: + parsed["spf"] = spf_entries + + dkim_entries = [ + { + "domain": DMARCParser._text(dkim, "domain"), + "result": DMARCParser._text(dkim, "result").lower(), + "selector": DMARCParser._text(dkim, "selector"), + "human_result": DMARCParser._text(dkim, "human_result"), + } + for dkim in auth_results.findall("dkim") + ] + if dkim_entries: + parsed["dkim"] = dkim_entries + return parsed + + @staticmethod + def _parse_record_extensions(record_elem) -> dict: + """Parse record-level extension elements.""" + extension_values = {} + for child in record_elem: + if child.tag not in {"row", "identifiers", "auth_results"}: + extension_values[child.tag] = DMARCParser._extension_value(child) + return {"extensions": extension_values} if extension_values else {} + @staticmethod def _parse_record(record_elem) -> dict: """Parse a single element into a dictionary.""" record: dict = {} - - row = record_elem.find("row") - if row is not None: - record["source_ip"] = row.findtext("source_ip", "") - record["count"] = int(row.findtext("count", 0)) - policy_evaluated = row.find("policy_evaluated") - if policy_evaluated is not None: - record["disposition"] = policy_evaluated.findtext("disposition", "none") - record["dkim_result"] = policy_evaluated.findtext("dkim", "").lower() - record["spf_result"] = policy_evaluated.findtext("spf", "").lower() - - identifiers = record_elem.find("identifiers") - if identifiers is not None: - record["header_from"] = identifiers.findtext("header_from", "") - - auth_results = record_elem.find("auth_results") - if auth_results is not None: - spf_entries = [ - { - "domain": spf.findtext("domain", ""), - "result": spf.findtext("result", "").lower(), - } - for spf in auth_results.findall("spf") - ] - if spf_entries: - record["spf"] = spf_entries - - dkim_entries = [ - { - "domain": dkim.findtext("domain", ""), - "result": dkim.findtext("result", "").lower(), - "selector": dkim.findtext("selector", ""), - } - for dkim in auth_results.findall("dkim") - ] - if dkim_entries: - record["dkim"] = dkim_entries + record.update(DMARCParser._parse_row(record_elem)) + record.update(DMARCParser._parse_identifiers(record_elem)) + record.update(DMARCParser._parse_auth_results(record_elem)) + record.update(DMARCParser._parse_record_extensions(record_elem)) return record @staticmethod def _compute_summary(records: list) -> dict: """Compute aggregate pass/fail statistics for a list of records.""" - total_count = sum(r["count"] for r in records) + total_count = sum(r.get("count", 0) for r in records) passed_count = sum( - r["count"] + r.get("count", 0) for r in records if r.get("spf_result") == "pass" or r.get("dkim_result") == "pass" ) @@ -224,19 +379,18 @@ class DMARCParser: """ try: root = ET.fromstring(xml_content) + xml_namespace = DMARCParser._namespace(root.tag) DMARCParser._strip_namespace(root) report = DMARCParser._parse_metadata(root) + report.update(DMARCParser._detect_variant(root, xml_namespace)) # Parse policy published - policy = root.find("policy_published") - if policy is not None: - report["domain"] = policy.findtext("domain", "") - report["policy"] = { - "p": policy.findtext("p", "none"), - "sp": policy.findtext("sp", ""), - "pct": policy.findtext("pct", "100"), - } + report.update(DMARCParser._parse_policy(root)) + + extension = root.find("extension") + if extension is not None: + report["extensions"] = DMARCParser._collect_extension_values(extension) # Parse records records = [DMARCParser._parse_record(elem) for elem in root.findall("record")] diff --git a/backend/app/tests/test_dmarc_parser.py b/backend/app/tests/test_dmarc_parser.py index 446d3f4..440e523 100644 --- a/backend/app/tests/test_dmarc_parser.py +++ b/backend/app/tests/test_dmarc_parser.py @@ -129,3 +129,159 @@ class TestDMARCParser: assert result["domain"] == "example.com" assert result["records"] == [] assert result["summary"]["total_count"] == 0 + + def test_parse_rfc9990_style_report_variant(self): + """RFC 9990-era namespaces and optional fields should parse without breaking legacy shape.""" + xml = b""" + + 1.0 + + Example Receiver + dmarc@example.test + https://example.test/dmarc + 2026-05-23-example.org + + 1779494400 + 1779580799 + + Multiple DMARC records were ignored before treewalk. + ExampleRUA 2.0 + + + example.org + treewalk +

quarantine

+ reject + none + 1 + s + r + y + + + mx1.example.test + + + + 2001:db8::1 + 5 + + quarantine + fail + pass + + local_policy + trusted relay + + + + + news.example.org + bounce.example.org + customer.example.net + + + + example.net + selector1 + fail + body hash did not verify + + + bounce.example.org + mfrom + pass + sender authorized + + + mail-platform + + + """ + + result = DMARCParser.parse_file(xml, "report.xml") + + assert result["variant"] == "rfc9990" + assert result["schema_version"] == "1.0" + assert result["xml_namespace"] == "urn:ietf:params:xml:ns:dmarc-2.0" + assert result["report_id"] == "2026-05-23-example.org" + assert result["generator"] == "ExampleRUA 2.0" + assert result["errors"] == ["Multiple DMARC records were ignored before treewalk."] + assert result["extensions"] == {"receiver": "mx1.example.test"} + assert result["policy"] == { + "p": "quarantine", + "sp": "reject", + "pct": "100", + "np": "none", + "fo": "1", + "adkim": "s", + "aspf": "r", + "testing": "y", + "discovery_method": "treewalk", + } + record = result["records"][0] + assert record["source_ip"] == "2001:db8::1" + assert record["count"] == 5 + assert record["envelope_from"] == "bounce.example.org" + assert record["envelope_to"] == "customer.example.net" + assert record["policy_override_reasons"] == [ + {"type": "local_policy", "comment": "trusted relay"} + ] + assert record["dkim"] == [ + { + "domain": "example.net", + "result": "fail", + "selector": "selector1", + "human_result": "body hash did not verify", + } + ] + assert record["spf"] == [ + { + "domain": "bounce.example.org", + "scope": "mfrom", + "result": "pass", + "human_result": "sender authorized", + } + ] + assert record["extensions"] == {"source": "mail-platform"} + assert result["summary"]["total_count"] == 5 + assert result["summary"]["passed_count"] == 5 + + def test_parse_report_with_bad_optional_numbers_uses_safe_defaults(self): + """Real reports with malformed counts or timestamps should not crash parsing.""" + xml = b""" + + + Example Receiver + dmarc@example.test + bad-numbers + + not-a-timestamp + also-bad + + + + example.com +

none

+ + + + 203.0.113.10 + not-a-count + + none + pass + pass + + + + + """ + + result = DMARCParser.parse_file(xml, "report.xml") + + assert result["begin_timestamp"] == 0 + assert result["end_timestamp"] == 0 + assert result["records"][0]["count"] == 0 + assert result["summary"]["total_count"] == 0 diff --git a/docs/milestones.md b/docs/milestones.md index 6623e62..a717df0 100644 --- a/docs/milestones.md +++ b/docs/milestones.md @@ -182,12 +182,15 @@ Exit criteria: ## Milestone 11: DMARC Format Compatibility (DMARCbis) and Standards Alignment -Status: Planned +Status: In Progress Goal: keep DMARQ compatible with evolving DMARC report formats and nomenclature without breaking existing imports. +Delivered: +- Add parser compatibility for RFC 9990-style aggregate report namespaces, version detection, policy metadata, identifiers, override reasons, auth-result details, and namespaced extensions. +- Keep legacy RFC 7489-style reports backward compatible through fixture coverage. + Planned: -- Add parser compatibility for newer aggregate report schemas/namespaces. - Store newly introduced fields with safe defaults. - Update CSV export and domain/source reporting to include new metadata where it improves operator actionability. - Add fixture-driven tests for representative real-world DMARCbis-style reports.