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

feat: add forensic report parsing storage

Browse Source
This commit is contained in:
Christian Krakau-Louis
2026-05-23 12:52:54 +02:00
parent ca9e0bfd4a
commit 1f954c9e08
18 changed files with 1005 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/services/forensic_parser.py
+197
View File
@@ -0,0 +1,197 @@
import hashlib
import json
import re
from datetime import datetime
from email import message_from_bytes
from email.message import Message
from email.parser import Parser
from email.utils import getaddresses, parsedate_to_datetime
from typing import Any, Dict, Optional
MAX_FORENSIC_REPORT_SIZE = 10 * 1024 * 1024
_EMAIL_RE = re.compile(r"\b([A-Z0-9._%+-]{1,64})@([A-Z0-9.-]+\.[A-Z]{2,})\b", re.IGNORECASE)
_LONG_TOKEN_RE = re.compile(r"\b[A-Za-z0-9_./+=-]{28,}\b")
def _coerce_text(value: Any) -> str:
if value is None:
return ""
if isinstance(value, bytes):
return value.decode("utf-8", errors="replace")
return str(value)
def _clean(value: Any, *, redact: bool = True) -> str:
text = " ".join(_coerce_text(value).replace("\r", " ").replace("\n", " ").split())
return redact_text(text) if redact else text
def redact_text(value: str) -> str:
"""Redact email local-parts and long opaque tokens from forensic metadata."""
def _redact_email(match: re.Match[str]) -> str:
local = match.group(1)
domain = match.group(2)
prefix = local[:2] if len(local) > 2 else local[:1]
return f"{prefix}***@{domain.lower()}"
redacted = _EMAIL_RE.sub(_redact_email, value)
return _LONG_TOKEN_RE.sub("[redacted-token]", redacted)
def _header(msg: Optional[Message], name: str, *, redact: bool = True) -> str:
return _clean(msg.get(name, "") if msg is not None else "", redact=redact)
def _payload_text(part: Message) -> str:
payload = part.get_payload(decode=True)
if payload is not None:
charset = part.get_content_charset() or "utf-8"
return payload.decode(charset, errors="replace")
payload_value = part.get_payload()
if isinstance(payload_value, list):
return ""
return _coerce_text(payload_value)
def _message_part_payload(part: Message) -> Optional[Message]:
payload = part.get_payload()
if isinstance(payload, list) and payload:
return payload[0]
return None
def _parse_feedback_headers(text: str) -> Message:
return Parser().parsestr(text or "")
def _domain_from_address(value: str) -> str:
addresses = getaddresses([value])
for _, addr in addresses:
if "@" in addr:
return addr.rsplit("@", 1)[-1].lower()
return ""
def _parse_datetime(value: str) -> Optional[datetime]:
if not value:
return None
try:
parsed = parsedate_to_datetime(value)
except (TypeError, ValueError, IndexError):
return None
if parsed is None:
return None
return parsed.replace(tzinfo=None)
def _message_id_hash(value: str) -> str:
cleaned = _clean(value, redact=False)
if not cleaned:
return ""
return hashlib.sha256(cleaned.encode("utf-8")).hexdigest()[:24]
class ForensicParser:
"""Parse DMARC forensic/failure report emails without retaining message bodies."""
@staticmethod
def is_forensic_report(msg: Message) -> bool:
if msg.get_content_type() == "multipart/report":
report_type = (msg.get_param("report-type") or "").lower()
if report_type == "feedback-report":
return True
for part in msg.walk():
content_type = part.get_content_type().lower()
if content_type == "message/feedback-report":
return True
if content_type == "text/rfc822-headers" and "dmarc" in _payload_text(part).lower():
return True
subject = _header(msg, "Subject", redact=False).lower()
return "dmarc" in subject and any(
term in subject for term in ("failure", "forensic", "ruf")
)
@classmethod
def parse_bytes(
cls, content: bytes, *, message_id_hint: Optional[str] = None
) -> Dict[str, Any]:
if len(content) > MAX_FORENSIC_REPORT_SIZE:
raise ValueError("Forensic report is too large")
if not content:
raise ValueError("Forensic report is empty")
msg = message_from_bytes(content)
if not cls.is_forensic_report(msg):
raise ValueError("Email is not a DMARC forensic report")
feedback = None
original_headers = None
for part in msg.walk():
content_type = part.get_content_type().lower()
if content_type == "message/feedback-report":
feedback = _message_part_payload(part) or _parse_feedback_headers(
_payload_text(part)
)
elif content_type == "text/rfc822-headers":
original_headers = _parse_feedback_headers(_payload_text(part))
elif content_type == "message/rfc822" and original_headers is None:
original_headers = _message_part_payload(part)
feedback = feedback or msg
reported_domain = (
_header(feedback, "Reported-Domain", redact=False)
or _header(feedback, "DKIM-Domain", redact=False)
or _domain_from_address(_header(feedback, "Original-Mail-From", redact=False))
or _domain_from_address(_header(original_headers, "From", redact=False))
).lower()
source_ip = _header(feedback, "Source-IP", redact=False)
auth_failure = _header(feedback, "Auth-Failure", redact=False)
original_message_id = _header(original_headers, "Message-ID", redact=False)
top_message_id = _header(msg, "Message-ID", redact=False)
report_id = (
_clean(message_id_hint, redact=False)
or _message_id_hash(top_message_id)
or _message_id_hash(original_message_id)
or hashlib.sha256(content).hexdigest()[:24]
)
if not report_id.startswith("ruf-"):
report_id = f"ruf-{report_id}"
source_email = _header(msg, "From")
arrival_date = _parse_datetime(_header(feedback, "Arrival-Date", redact=False))
details = {
"identity_alignment": _header(feedback, "Identity-Alignment", redact=False),
"dkim_domain": _header(feedback, "DKIM-Domain", redact=False),
"spf_dns": _header(feedback, "SPF-DNS", redact=False),
"reported_uri": _header(feedback, "Reported-URI"),
}
details = {key: value for key, value in details.items() if value}
return {
"report_id": report_id,
"source_email": source_email,
"feedback_type": _header(feedback, "Feedback-Type", redact=False) or "auth-failure",
"user_agent": _header(feedback, "User-Agent"),
"version": _header(feedback, "Version", redact=False),
"reported_domain": reported_domain,
"source_ip": source_ip,
"auth_failure": auth_failure,
"delivery_result": _header(feedback, "Delivery-Result", redact=False),
"arrival_date": arrival_date,
"authentication_results": _header(feedback, "Authentication-Results"),
"original_mail_from": _header(feedback, "Original-Mail-From"),
"original_from": _header(original_headers, "From"),
"original_to": _header(original_headers, "To"),
"original_subject": _header(original_headers, "Subject"),
"original_message_id": _message_id_hash(original_message_id),
"original_date": _header(original_headers, "Date", redact=False),
"feedback_headers": json.dumps(details, sort_keys=True) if details else None,
}
backend/app/services/forensic_persistence.py
+103
View File
@@ -0,0 +1,103 @@
import json
from typing import Any, Dict, Optional
from sqlalchemy.orm import Session
from app.models.domain import Domain
from app.models.report import ForensicReport
from app.utils.domain_validator import DomainValidationError, validate_domain
def forensic_report_exists(db: Session, report_id: str) -> bool:
"""Return True when a forensic report ID is already persisted."""
if not report_id:
return False
return (
db.query(ForensicReport.id).filter(ForensicReport.report_id == report_id).first()
is not None
)
def _domain_for_report(db: Session, domain_name: Optional[str]) -> Optional[Domain]:
if not domain_name:
return None
normalized = domain_name.lower().strip(".")
is_valid, _, error_code = validate_domain(normalized, check_dns=False)
if not is_valid and error_code != DomainValidationError.DNS_RESOLUTION_FAILED:
return None
domain = db.query(Domain).filter(Domain.name == normalized).first()
if domain is None:
domain = Domain(name=normalized)
db.add(domain)
db.flush()
return domain
def save_forensic_report(db: Session, report: Dict[str, Any]) -> tuple[ForensicReport, bool]:
"""Persist a parsed forensic report.
Returns ``(row, created)``. The caller owns the transaction and should
commit after related work has completed.
"""
report_id = str(report.get("report_id") or "")
existing = db.query(ForensicReport).filter(ForensicReport.report_id == report_id).first()
if existing is not None:
return existing, False
domain = _domain_for_report(db, report.get("reported_domain"))
feedback_headers = report.get("feedback_headers")
if isinstance(feedback_headers, dict):
feedback_headers = json.dumps(feedback_headers, sort_keys=True)
row = ForensicReport(
domain_id=domain.id if domain else None,
report_id=report_id,
source_email=report.get("source_email"),
feedback_type=report.get("feedback_type"),
user_agent=report.get("user_agent"),
version=report.get("version"),
reported_domain=report.get("reported_domain"),
source_ip=report.get("source_ip"),
auth_failure=report.get("auth_failure"),
delivery_result=report.get("delivery_result"),
arrival_date=report.get("arrival_date"),
authentication_results=report.get("authentication_results"),
original_mail_from=report.get("original_mail_from"),
original_from=report.get("original_from"),
original_to=report.get("original_to"),
original_subject=report.get("original_subject"),
original_message_id=report.get("original_message_id"),
original_date=report.get("original_date"),
feedback_headers=feedback_headers,
)
db.add(row)
db.flush()
return row, True
def forensic_report_to_dict(row: ForensicReport) -> Dict[str, Any]:
"""Convert a forensic report row to an API-safe dictionary."""
return {
"id": row.id,
"report_id": row.report_id,
"domain": row.domain.name if row.domain else row.reported_domain,
"reported_domain": row.reported_domain,
"source_email": row.source_email,
"feedback_type": row.feedback_type,
"user_agent": row.user_agent,
"version": row.version,
"source_ip": row.source_ip,
"auth_failure": row.auth_failure,
"delivery_result": row.delivery_result,
"arrival_date": row.arrival_date.isoformat() if row.arrival_date else None,
"authentication_results": row.authentication_results,
"original_mail_from": row.original_mail_from,
"original_from": row.original_from,
"original_to": row.original_to,
"original_subject": row.original_subject,
"original_message_id": row.original_message_id,
"original_date": row.original_date,
"feedback_headers": json.loads(row.feedback_headers) if row.feedback_headers else {},
"processed_at": row.processed_at.isoformat() if row.processed_at else None,
}
backend/app/services/gmail_client.py
+81 -3
View File
@@ -21,6 +21,8 @@ from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from app.services.dmarc_parser import DMARCParser
from app.services.forensic_parser import ForensicParser
from app.services.forensic_persistence import forensic_report_exists, save_forensic_report
from app.services.report_persistence import report_exists, save_parsed_report
from app.services.report_store import ReportStore
@@ -47,8 +49,8 @@ GMAIL_SCOPES = [
# ---------------------------------------------------------------------------
DMARC_GMAIL_QUERY = (
"has:attachment "
"(filename:zip OR filename:gz OR filename:xml) "
"((has:attachment (filename:zip OR filename:gz OR filename:xml)) "
'OR subject:"DMARC failure" OR subject:"failure report" OR subject:forensic OR subject:ruf) '
"(subject:dmarc OR subject:report OR subject:rua OR subject:submitter "
'OR subject:"aggregate report" OR subject:"domain report" '
'OR subject:"report domain" OR from:dmarc OR from:dmarc-noreply '
@@ -210,7 +212,9 @@ class GmailClient:
"success": True,
"processed": 0,
"reports_found": 0,
"forensic_reports_found": 0,
"duplicate_reports": 0,
"duplicate_forensic_reports": 0,
"new_domains": [],
"errors": [],
"new_ingested_ids": [],
@@ -302,7 +306,9 @@ class GmailClient:
@staticmethod
def _append_detail(stats: dict, **detail: str) -> None:
"""Append a compact attachment/message outcome to the import stats."""
stats.setdefault("details", []).append({key: value for key, value in detail.items() if value})
stats.setdefault("details", []).append(
{key: value for key, value in detail.items() if value}
)
def _process_message(self, service, msg_id: str, stats: dict) -> int:
"""
@@ -327,6 +333,8 @@ class GmailClient:
raw_bytes = base64.urlsafe_b64decode(msg_data.get("raw", ""))
msg = email.message_from_bytes(raw_bytes)
if ForensicParser.is_forensic_report(msg):
return 1 if self._process_forensic_message(raw_bytes, stats, message_id=msg_id) else 0
return self._process_attachments(msg, stats, message_id=msg_id)
@staticmethod
@@ -370,6 +378,76 @@ class GmailClient:
self.report_store.add_report(report)
return True
def _process_forensic_message(
self,
raw_bytes: bytes,
stats: dict,
message_id: Optional[str] = None,
) -> bool:
"""Parse and persist one DMARC forensic report message."""
try:
report = ForensicParser.parse_bytes(raw_bytes, message_id_hint=message_id)
report_id = str(report.get("report_id", ""))
domain = str(report.get("reported_domain") or "unknown")
if self.db is None:
self._append_detail(
stats,
status="skipped",
reason="forensic_report_requires_database",
message_id=message_id,
domain=domain,
report_id=report_id,
)
return False
if forensic_report_exists(self.db, report_id):
stats["duplicate_forensic_reports"] = stats.get("duplicate_forensic_reports", 0) + 1
self._append_detail(
stats,
status="duplicate",
reason="duplicate_forensic_report",
message_id=message_id,
domain=domain,
report_id=report_id,
)
return False
_row, created = save_forensic_report(self.db, report)
if not created:
stats["duplicate_forensic_reports"] = stats.get("duplicate_forensic_reports", 0) + 1
self._append_detail(
stats,
status="duplicate",
reason="duplicate_forensic_report",
message_id=message_id,
domain=domain,
report_id=report_id,
)
return False
stats["forensic_reports_found"] = stats.get("forensic_reports_found", 0) + 1
self._append_detail(
stats,
status="imported",
reason="forensic_report",
message_id=message_id,
domain=domain,
report_id=report_id,
)
return True
except Exception as exc: # pylint: disable=broad-exception-caught
logger.error("Failed to parse Gmail forensic report %s: %s", message_id, exc)
stats["errors"].append(f"Failed to parse forensic report {message_id}: {exc}")
self._append_detail(
stats,
status="error",
reason="forensic_parse_failed",
message_id=message_id,
error=str(exc),
)
return False
def _process_attachments(
self,
msg: email.message.Message,
backend/app/services/imap_client.py
+95
View File
@@ -7,6 +7,8 @@ from typing import Any, Dict, Optional, Tuple
from app.core.config import get_settings
from app.services.dmarc_parser import DMARCParser
from app.services.forensic_parser import ForensicParser
from app.services.forensic_persistence import forensic_report_exists, save_forensic_report
from app.services.report_persistence import report_exists, save_parsed_report
from app.services.report_store import ReportStore
@@ -202,6 +204,19 @@ class IMAPClient:
raw_email = msg_data[0][1]
msg = email.message_from_bytes(raw_email)
if ForensicParser.is_forensic_report(msg):
imported = self._process_forensic_email(
raw_email,
stats=stats,
message_id=message_id,
)
mail.store(email_id, "+FLAGS", "\\Seen")
if self.delete_emails and imported:
mail.store(email_id, "+FLAGS", "\\Deleted")
stats["deleted"] = stats.get("deleted", 0) + 1
stats["processed"] += 1
return
if self._is_dmarc_report_email(msg):
reports_found = self._process_attachments(msg, stats, message_id=message_id)
stats["reports_found"] += reports_found
@@ -243,8 +258,10 @@ class IMAPClient:
"success": True,
"processed": 0,
"reports_found": 0,
"forensic_reports_found": 0,
"deleted": 0,
"duplicate_reports": 0,
"duplicate_forensic_reports": 0,
"new_domains": [],
"errors": [],
"details": [],
@@ -472,6 +489,84 @@ class IMAPClient:
)
return True
def _process_forensic_email(
self,
raw_email: bytes,
*,
stats: Optional[Dict[str, Any]],
message_id: Optional[str],
) -> bool:
try:
report = ForensicParser.parse_bytes(raw_email)
report_id = str(report.get("report_id", ""))
domain = str(report.get("reported_domain") or "unknown")
if self.db is not None and forensic_report_exists(self.db, report_id):
if stats is not None:
stats["duplicate_forensic_reports"] = (
stats.get("duplicate_forensic_reports", 0) + 1
)
self._append_detail(
stats,
status="duplicate",
reason="duplicate_forensic_report",
message_id=message_id,
domain=domain,
report_id=report_id,
)
return False
if self.db is None:
self._append_detail(
stats,
status="skipped",
reason="forensic_report_requires_database",
message_id=message_id,
domain=domain,
report_id=report_id,
)
return False
_row, created = save_forensic_report(self.db, report)
if created:
if stats is not None:
stats["forensic_reports_found"] = stats.get("forensic_reports_found", 0) + 1
self._append_detail(
stats,
status="imported",
reason="forensic_report",
message_id=message_id,
domain=domain,
report_id=report_id,
)
return True
if stats is not None:
stats["duplicate_forensic_reports"] = stats.get("duplicate_forensic_reports", 0) + 1
self._append_detail(
stats,
status="duplicate",
reason="duplicate_forensic_report",
message_id=message_id,
domain=domain,
report_id=report_id,
)
return False
except Exception as exc: # pylint: disable=broad-exception-caught
logger.error("Error processing forensic report email %s: %s", message_id, exc)
if stats is not None:
stats.setdefault("errors", []).append(
f"Failed to parse forensic report {message_id}: {exc}"
)
self._append_detail(
stats,
status="error",
reason="forensic_parse_failed",
message_id=message_id,
error=str(exc),
)
return False
def _process_dmarc_attachment(
self,
part: email.message.Message,