gitea-admin/gh-christianlouis-dmarq
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: harden DMARC mail import

Browse Source
This commit is contained in:
Christian Krakau-Louis
2026-05-22 19:17:32 +02:00
parent 9e9df108c8
commit 07b2ea400e
6 changed files with 293 additions and 518 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/services/gmail_client.py
+20 -9
View File
@@ -48,9 +48,10 @@ GMAIL_SCOPES = [
DMARC_GMAIL_QUERY = (
"has:attachment "
"(filename:zip OR filename:gz OR filename:xml) "
"(subject:dmarc OR subject:report OR subject:rua "
"(subject:dmarc OR subject:report OR subject:rua OR subject:submitter "
'OR subject:"aggregate report" OR subject:"domain report" '
"OR from:dmarc OR from:dmarc-noreply OR from:reports OR from:postmaster)"
'OR subject:"report domain" OR from:dmarc OR from:dmarc-noreply '
"OR from:noreply-dmarc-support OR from:reports OR from:postmaster)"
)
# How many message results to fetch per API page
@@ -331,16 +332,28 @@ class GmailClient:
or lower.endswith(".gzip")
)
def _store_report_if_new(self, report: Dict[str, Any]) -> bool:
"""Store a parsed report unless that domain/report ID is already present."""
domain = report.get("domain", "unknown")
report_id = report.get("report_id", "")
if report_id and self.report_store.has_report(domain, report_id):
logger.info("Skipping duplicate DMARC report %s for %s", report_id, domain)
return False
self.report_store.add_report(report)
return True
def _process_attachments(self, msg: email.message.Message, stats: dict) -> int:
"""Walk a parsed email message and extract DMARC report attachments."""
reports_found = 0
for part in msg.walk():
if part.get_content_disposition() != "attachment":
filename = self._decode_part_filename(part)
if not filename or not self._is_dmarc_attachment(filename):
continue
filename = self._decode_part_filename(part)
if not self._is_dmarc_attachment(filename):
disposition = part.get_content_disposition()
if disposition not in ("attachment", None):
continue
content = part.get_payload(decode=True)
@@ -348,10 +361,8 @@ class GmailClient:
continue
try:
parser = DMARCParser()
reports = parser.parse(content, filename)
for report in reports:
self.report_store.add_report(report)
report = DMARCParser.parse_file(content, filename)
if self._store_report_if_new(report):
stats["reports_found"] += 1
reports_found += 1
except Exception as exc: # pylint: disable=broad-exception-caught
backend/app/services/imap_client.py
+10
View File
@@ -386,6 +386,16 @@ class IMAPClient:
# Parse the DMARC report
report = DMARCParser.parse_file(content, filename)
domain = report.get("domain", "unknown")
report_id = report.get("report_id", "")
if report_id and self.report_store.has_report(domain, report_id):
logger.info(
"Skipping duplicate DMARC report %s for %s",
report_id,
domain,
)
continue
# Add the report to the store
self.report_store.add_report(report)
backend/app/tests/test_gmail_client.py
+49 -15
View File
@@ -8,16 +8,20 @@ tests never make real network calls.
import base64
import email as email_mod
import json
import zipfile
from email import encoders as email_encoders
from email.mime.base import MIMEBase
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from io import BytesIO
from typing import Optional
from unittest.mock import MagicMock, patch
import pytest
from app.services.gmail_client import GmailClient
from app.services.report_store import ReportStore
from app.tests.test_data import SAMPLE_XML
# ---------------------------------------------------------------------------
# Helpers
@@ -82,6 +86,14 @@ def _b64_raw(raw_bytes: bytes) -> str:
return base64.urlsafe_b64encode(raw_bytes).decode()
def _zip_xml(xml: str = SAMPLE_XML, name: str = "report.xml") -> bytes:
"""Create a small DMARC ZIP attachment."""
buf = BytesIO()
with zipfile.ZipFile(buf, "w") as zf:
zf.writestr(name, xml.encode("utf-8"))
return buf.getvalue()
# ===========================================================================
# __init__ / basic construction
# ===========================================================================
@@ -432,6 +444,9 @@ class TestProcessMessage:
class TestProcessAttachments:
def setup_method(self):
ReportStore.get_instance().clear()
def test_no_attachments_returns_zero(self):
client = _make_client()
msg = email_mod.message_from_bytes(b"From: a@b.com\r\nTo: c@d.com\r\n\r\nHello")
@@ -449,24 +464,45 @@ class TestProcessAttachments:
assert count == 0
assert stats["reports_found"] == 0
def test_dmarc_xml_attachment_is_parsed(self):
"""A .xml attachment is parsed via DMARCParser and counts as a report."""
def test_google_style_zip_attachment_is_parsed(self):
"""A Google DMARC ZIP attachment is parsed and counted."""
client = _make_client()
raw = _make_raw_email([{"filename": "report.xml", "content": b"<xml_content/>"}])
raw = _make_raw_email(
[
{
"filename": "google.com!example.com!1597449600!1597535999.zip",
"content": _zip_xml(),
}
]
)
msg = email_mod.message_from_bytes(raw)
stats = {"reports_found": 0, "errors": []}
mock_report = {"domain": "example.com", "records": []}
with patch("app.services.gmail_client.DMARCParser") as mock_parser_class:
mock_parser = MagicMock()
mock_parser.parse.return_value = [mock_report]
mock_parser_class.return_value = mock_parser
# Also mock report_store.add_report to avoid real persistence
with patch.object(client.report_store, "add_report"):
count = client._process_attachments(msg, stats)
count = client._process_attachments(msg, stats)
assert count == 1
assert stats["reports_found"] == 1
assert "example.com" in client.report_store.get_domains()
def test_duplicate_report_is_skipped(self):
"""Repeated imports of the same domain/report ID should not inflate totals."""
client = _make_client()
raw = _make_raw_email(
[
{
"filename": "google.com!example.com!1597449600!1597535999.zip",
"content": _zip_xml(),
}
]
)
msg = email_mod.message_from_bytes(raw)
first_stats = {"reports_found": 0, "errors": []}
second_stats = {"reports_found": 0, "errors": []}
assert client._process_attachments(msg, first_stats) == 1
assert client._process_attachments(msg, second_stats) == 0
assert client.report_store.get_domain_summary("example.com")["reports_processed"] == 1
def test_dmarc_attachment_with_empty_content_skipped(self):
"""A DMARC-named attachment with truly empty payload is skipped gracefully."""
@@ -500,12 +536,10 @@ class TestProcessAttachments:
call_count += 1
if call_count == 1:
raise ValueError("bad xml")
return [good_report]
return good_report
with patch("app.services.gmail_client.DMARCParser") as mock_parser_class:
mock_parser = MagicMock()
mock_parser.parse.side_effect = parse_side_effect
mock_parser_class.return_value = mock_parser
mock_parser_class.parse_file.side_effect = parse_side_effect
with patch.object(client.report_store, "add_report"):
count = client._process_attachments(msg, stats)