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

feat: add mail import detail events

Browse Source
This commit is contained in:
Christian Krakau-Louis
2026-05-22 20:06:51 +02:00
parent 5f416d042f
commit c1b81d3e06
13 changed files with 520 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/services/gmail_client.py
+69 -3
View File
@@ -214,6 +214,7 @@ class GmailClient:
"new_domains": [],
"errors": [],
"new_ingested_ids": [],
"details": [],
}
try:
@@ -232,6 +233,12 @@ class GmailClient:
for msg_id in message_ids:
if msg_id in self.already_ingested_ids:
self._append_detail(
stats,
status="skipped",
reason="already_ingested_message",
message_id=msg_id,
)
continue
stats["processed"] += 1
@@ -292,6 +299,11 @@ class GmailClient:
return ids
@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})
def _process_message(self, service, msg_id: str, stats: dict) -> int:
"""
Download a Gmail message and process any DMARC-report attachments.
@@ -305,11 +317,17 @@ class GmailClient:
except HttpError as exc:
logger.error("Gmail API: failed to fetch message %s: %s", msg_id, exc)
stats["errors"].append(f"Failed to fetch message {msg_id}")
self._append_detail(
stats,
status="error",
reason="message_fetch_failed",
message_id=msg_id,
)
return 0
raw_bytes = base64.urlsafe_b64decode(msg_data.get("raw", ""))
msg = email.message_from_bytes(raw_bytes)
return self._process_attachments(msg, stats)
return self._process_attachments(msg, stats, message_id=msg_id)
@staticmethod
def _decode_part_filename(part: email.message.Message) -> str:
@@ -352,33 +370,81 @@ class GmailClient:
self.report_store.add_report(report)
return True
def _process_attachments(self, msg: email.message.Message, stats: dict) -> int:
def _process_attachments(
self,
msg: email.message.Message,
stats: dict,
message_id: Optional[str] = None,
) -> int:
"""Walk a parsed email message and extract DMARC report attachments."""
reports_found = 0
for part in msg.walk():
filename = self._decode_part_filename(part)
if not filename or not self._is_dmarc_attachment(filename):
if not filename:
continue
disposition = part.get_content_disposition()
if disposition not in ("attachment", None):
continue
if not self._is_dmarc_attachment(filename):
self._append_detail(
stats,
status="skipped",
reason="unsupported_attachment",
message_id=message_id,
filename=filename,
)
continue
content = part.get_payload(decode=True)
if not content:
self._append_detail(
stats,
status="skipped",
reason="empty_attachment",
message_id=message_id,
filename=filename,
)
continue
try:
report = DMARCParser.parse_file(content, filename)
domain = str(report.get("domain", "unknown"))
report_id = str(report.get("report_id", ""))
if self._store_report_if_new(report):
stats["reports_found"] += 1
reports_found += 1
self._append_detail(
stats,
status="imported",
message_id=message_id,
filename=filename,
domain=domain,
report_id=report_id,
)
else:
stats["duplicate_reports"] = stats.get("duplicate_reports", 0) + 1
self._append_detail(
stats,
status="duplicate",
message_id=message_id,
filename=filename,
domain=domain,
report_id=report_id,
)
except Exception as exc: # pylint: disable=broad-exception-caught
logger.error("Failed to parse DMARC attachment %s: %s", filename, exc)
stats["errors"].append(f"Failed to parse {filename}: {exc}")
self._append_detail(
stats,
status="error",
reason="parse_failed",
message_id=message_id,
filename=filename,
error=str(exc),
)
return reports_found
backend/app/services/imap_client.py
+148 -57
View File
@@ -140,17 +140,24 @@ class IMAPClient:
def _process_single_email(self, mail, email_id: bytes, stats: dict) -> None:
"""Fetch, parse, and store DMARC attachments from one email message."""
message_id = email_id.decode("utf-8", errors="replace")
try:
status, msg_data = mail.fetch(email_id, "(RFC822)")
if status != "OK":
logger.error("Error fetching email ID %s", email_id)
self._append_detail(
stats,
status="error",
reason="message_fetch_failed",
message_id=message_id,
)
return
raw_email = msg_data[0][1]
msg = email.message_from_bytes(raw_email)
if self._is_dmarc_report_email(msg):
reports_found = self._process_attachments(msg, stats)
reports_found = self._process_attachments(msg, stats, message_id=message_id)
stats["reports_found"] += reports_found
# Mark email as read (and optionally delete)
@@ -163,6 +170,13 @@ class IMAPClient:
error_msg = f"Error processing email ID {email_id}: {str(e)}"
logger.error(error_msg)
stats["errors"].append(error_msg)
self._append_detail(
stats,
status="error",
reason="message_processing_failed",
message_id=message_id,
error=str(e),
)
def fetch_reports(self, days: int = 7) -> Dict[str, Any]:
"""
@@ -185,6 +199,7 @@ class IMAPClient:
"duplicate_reports": 0,
"new_domains": [],
"errors": [],
"details": [],
}
try:
@@ -334,12 +349,7 @@ class IMAPClient:
filename = self._decode_email_header(filename)
# Check file extension
if (
filename.lower().endswith(".xml")
or filename.lower().endswith(".zip")
or filename.lower().endswith(".gz")
or filename.lower().endswith(".gzip")
):
if self._is_dmarc_filename(filename):
return True
# Check content type
@@ -355,7 +365,115 @@ class IMAPClient:
return False
def _process_attachments(self, msg: email.message.Message, stats: dict | None = None) -> int:
@staticmethod
def _is_dmarc_filename(filename: str) -> bool:
lower = filename.lower()
return (
lower.endswith(".xml")
or lower.endswith(".zip")
or lower.endswith(".gz")
or lower.endswith(".gzip")
)
@staticmethod
def _append_detail(stats: dict | None, **detail: str) -> None:
"""Append a compact attachment/message outcome to the import stats."""
if stats is None:
return
stats.setdefault("details", []).append(
{key: value for key, value in detail.items() if value}
)
def _store_report_if_new(
self,
report: Dict[str, Any],
*,
filename: str,
stats: dict | None,
message_id: str | None,
) -> bool:
domain = report.get("domain", "unknown")
report_id = report.get("report_id", "")
if report_id and (
self.report_store.has_report(domain, report_id)
or (self.db is not None and report_exists(self.db, domain, report_id))
):
logger.info("Skipping duplicate DMARC report %s for %s", report_id, domain)
if stats is not None:
stats["duplicate_reports"] = stats.get("duplicate_reports", 0) + 1
self._append_detail(
stats,
status="duplicate",
message_id=message_id,
filename=filename,
domain=str(domain),
report_id=str(report_id),
)
return False
if self.db is not None:
save_parsed_report(self.db, report)
self.report_store.add_report(report)
self._append_detail(
stats,
status="imported",
message_id=message_id,
filename=filename,
domain=str(domain),
report_id=str(report_id),
)
return True
def _process_dmarc_attachment(
self,
part: email.message.Message,
*,
filename: str,
stats: dict | None,
message_id: str | None,
) -> bool:
try:
content = part.get_payload(decode=True)
if not content:
self._append_detail(
stats,
status="skipped",
reason="empty_attachment",
message_id=message_id,
filename=filename,
)
return False
report = DMARCParser.parse_file(content, filename)
stored = self._store_report_if_new(
report,
filename=filename,
stats=stats,
message_id=message_id,
)
if stored:
logger.info("Successfully processed DMARC report: %s", filename)
return stored
except Exception as exc: # pylint: disable=broad-exception-caught
logger.error("Error processing attachment %s: %s", filename, str(exc))
if stats is not None:
stats.setdefault("errors", []).append(f"Failed to parse {filename}: {exc}")
self._append_detail(
stats,
status="error",
reason="parse_failed",
message_id=message_id,
filename=filename,
error=str(exc),
)
return False
def _process_attachments(
self,
msg: email.message.Message,
stats: dict | None = None,
message_id: str | None = None,
) -> int:
"""
Process email attachments that might be DMARC reports
@@ -368,57 +486,30 @@ class IMAPClient:
reports_found = 0
for part in msg.walk():
content_disposition = part.get_content_disposition()
if part.get_content_disposition() != "attachment":
continue
if content_disposition == "attachment":
filename = part.get_filename()
if filename:
# Decode filename if needed
filename = self._decode_email_header(filename)
filename = part.get_filename()
if not filename:
continue
# Check if it's a likely DMARC report file
if (
filename.lower().endswith(".xml")
or filename.lower().endswith(".zip")
or filename.lower().endswith(".gz")
or filename.lower().endswith(".gzip")
):
filename = self._decode_email_header(filename)
if not self._is_dmarc_filename(filename):
self._append_detail(
stats,
status="skipped",
reason="unsupported_attachment",
message_id=message_id,
filename=filename,
)
continue
try:
# Get attachment content
content = part.get_payload(decode=True)
# 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)
or (
self.db is not None
and report_exists(self.db, domain, report_id)
)
):
logger.info(
"Skipping duplicate DMARC report %s for %s",
report_id,
domain,
)
if stats is not None:
stats["duplicate_reports"] = (
stats.get("duplicate_reports", 0) + 1
)
continue
# Add the report to the store
if self.db is not None:
save_parsed_report(self.db, report)
self.report_store.add_report(report)
reports_found += 1
logger.info("Successfully processed DMARC report: %s", filename)
except Exception as e: # pylint: disable=broad-exception-caught
logger.error("Error processing attachment %s: %s", filename, str(e))
if self._process_dmarc_attachment(
part,
filename=filename,
stats=stats,
message_id=message_id,
):
reports_found += 1
return reports_found
backend/app/services/import_history.py
+34 -1
View File
@@ -9,13 +9,24 @@ from app.models.mail_source_import import MailSourceImport
MAX_STORED_ERRORS = 10
MAX_ERROR_LENGTH = 500
MAX_STORED_DETAILS = 50
MAX_DETAIL_VALUE_LENGTH = 300
DETAIL_FIELDS = {
"status",
"reason",
"message_id",
"filename",
"domain",
"report_id",
"error",
}
def _sanitize_error(value: object) -> str:
"""Return a compact, log-safe error string for storage and UI display."""
text = str(value).replace("\r", "").replace("\n", " ").strip()
if len(text) > MAX_ERROR_LENGTH:
return text[: MAX_ERROR_LENGTH - 1] + "..."
return text[: MAX_ERROR_LENGTH - 3] + "..."
return text
@@ -23,6 +34,27 @@ def _json_list(values: Optional[Iterable[Any]]) -> str:
return json.dumps([str(value) for value in values or []])
def _sanitize_detail_value(value: object) -> str:
text = _sanitize_error(value)
if len(text) > MAX_DETAIL_VALUE_LENGTH:
return text[: MAX_DETAIL_VALUE_LENGTH - 3] + "..."
return text
def _json_details(values: Optional[Iterable[Any]]) -> str:
details = []
for value in list(values or [])[:MAX_STORED_DETAILS]:
if not isinstance(value, dict):
continue
entry = {}
for key in DETAIL_FIELDS:
if key in value and value[key] not in (None, ""):
entry[key] = _sanitize_detail_value(value[key])
if entry:
details.append(entry)
return json.dumps(details)
def record_import_attempt(
db: Session,
source: MailSource,
@@ -47,6 +79,7 @@ def record_import_attempt(
error_count=len(result_errors),
new_domains=_json_list(results.get("new_domains", [])),
errors=json.dumps(errors),
details=_json_details(results.get("details", [])),
started_at=started_at,
finished_at=datetime.utcnow(),
)