Nautilus/DMARC-Sentinel
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add TLS reports capabilities

Browse Source
This commit is contained in:
Beda Schmid
2026-06-23 11:49:58 -03:00
parent 5c1dc95c36
commit f4ae9f8532
20 changed files with 1158 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/attachment_extractor.py
+14 -13
View File
@@ -22,6 +22,7 @@ class ExtractedReport:
ARCHIVE_SUFFIXES = (".zip", ".gz")
XML_MIME_HINTS = {"text/xml", "application/xml", "application/dmarc+xml"}
JSON_MIME_HINTS = {"application/json", "application/tlsrpt+json"}
GZIP_MIME_HINTS = {"application/gzip", "application/x-gzip"}
ZIP_MIME_HINTS = {"application/zip", "application/x-zip-compressed"}
@@ -68,25 +69,25 @@ def _extract_zip(filename: str, payload: bytes, max_mb: int, max_reports: int, m
lower = info.filename.lower()
if lower.endswith(ARCHIVE_SUFFIXES):
raise AttachmentExtractionError(f"{filename} contains nested archive {info.filename}")
if not lower.endswith(".xml"):
if not lower.endswith((".xml", ".json")):
continue
if len(reports) >= max_reports:
raise AttachmentExtractionError(f"{filename} exceeds archive XML report limit of {max_reports}")
raise AttachmentExtractionError(f"{filename} exceeds archive report limit of {max_reports}")
with archive.open(info) as handle:
xml = handle.read(_max_bytes(max_mb) + 1)
_ensure_size(xml, max_mb, info.filename)
_ensure_ratio(info.compress_size, len(xml), max_ratio, info.filename)
reports.append(ExtractedReport(info.filename, xml, _sha(xml)))
report = handle.read(_max_bytes(max_mb) + 1)
_ensure_size(report, max_mb, info.filename)
_ensure_ratio(info.compress_size, len(report), max_ratio, info.filename)
reports.append(ExtractedReport(info.filename, report, _sha(report)))
return reports
def _extract_gzip(filename: str, payload: bytes, max_mb: int, max_ratio: int) -> list[ExtractedReport]:
with gzip.GzipFile(fileobj=io.BytesIO(payload)) as gz:
xml = gz.read(_max_bytes(max_mb) + 1)
_ensure_size(xml, max_mb, filename)
_ensure_ratio(len(payload), len(xml), max_ratio, filename)
report = gz.read(_max_bytes(max_mb) + 1)
_ensure_size(report, max_mb, filename)
_ensure_ratio(len(payload), len(report), max_ratio, filename)
out_name = filename[:-3] if filename.lower().endswith(".gz") else f"{filename}.xml"
return [ExtractedReport(out_name, xml, _sha(xml))]
return [ExtractedReport(out_name, report, _sha(report))]
def extract_payload(
@@ -106,7 +107,7 @@ def extract_payload(
return _extract_zip(filename, payload, max_mb, max_reports_per_archive, max_compression_ratio)
if lower.endswith(".gz") or mime in GZIP_MIME_HINTS:
return _extract_gzip(filename, payload, max_mb, max_compression_ratio)
if lower.endswith(".xml") or mime in XML_MIME_HINTS:
if lower.endswith((".xml", ".json")) or mime in XML_MIME_HINTS | JSON_MIME_HINTS:
_ensure_size(payload, max_mb, filename)
return [ExtractedReport(filename, payload, _sha(payload))]
return []
@@ -117,9 +118,9 @@ def message_has_candidate_attachment(message: Message) -> bool:
filename = part.get_filename() or ""
content_type = (part.get_content_type() or "").lower()
lower = filename.lower()
if lower.endswith((".xml", ".xml.gz", ".gz", ".zip")):
if lower.endswith((".xml", ".json", ".xml.gz", ".json.gz", ".gz", ".zip")):
return True
if content_type in XML_MIME_HINTS | GZIP_MIME_HINTS | ZIP_MIME_HINTS:
if content_type in XML_MIME_HINTS | JSON_MIME_HINTS | GZIP_MIME_HINTS | ZIP_MIME_HINTS:
return True
return False