diff --git a/README.md b/README.md index f55f244..c1481f0 100644 --- a/README.md +++ b/README.md @@ -1,14 +1,15 @@ # DMARC Sentinel -DMARC Sentinel is a self-hosted Docker application for monitoring DMARC aggregate reports. It reads report emails from configured IMAP folders, extracts XML reports, stores normalized telemetry in SQLite, runs deterministic security and deliverability analysis, generates alerts, and uses an LLM to turn already-derived facts into human-readable summaries and recommendations. +DMARC Sentinel is a self-hosted Docker application for monitoring DMARC aggregate reports and SMTP TLS aggregate reports. It reads report emails from configured IMAP folders, extracts report attachments, stores normalized telemetry in SQLite, runs deterministic security and deliverability analysis, generates alerts, and uses an LLM to turn already-derived facts into human-readable summaries and recommendations. Detection is deterministic. The LLM is mandatory for explanations, daily summaries, weekly summaries, and dashboard text, but it is never the source of truth for whether an alert exists or how severe it is. ## What It Does - Connects to one or more IMAP inboxes and reads only the configured DMARC folder by default. -- Supports `.xml`, `.xml.gz`, `.gz`, and `.zip` report attachments. +- Supports `.xml`, `.xml.gz`, `.json`, `.json.gz`, `.gz`, and `.zip` report attachments. - Parses DMARC aggregate XML into reports, records, and authentication results. +- Parses TLSRPT aggregate JSON into reports, policies, session counts, and failure details. - Deduplicates reports by raw XML SHA256. - Classifies known senders using CIDR allowlists, DKIM auth domains, SPF auth domains, and aligned DKIM evidence. - Creates deterministic alerts for unknown failures, known sender failures, quarantine/reject dispositions, new sources, high failure rates, missing reporters, and first-seen policies/reporters. diff --git a/app/attachment_extractor.py b/app/attachment_extractor.py index f6686b8..fb82d8b 100644 --- a/app/attachment_extractor.py +++ b/app/attachment_extractor.py @@ -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 diff --git a/app/homepage.py b/app/homepage.py index 1639129..7dc9880 100644 --- a/app/homepage.py +++ b/app/homepage.py @@ -5,7 +5,7 @@ from datetime import date, datetime, timedelta, timezone from sqlalchemy import func, select from sqlalchemy.orm import Session -from app.models import Alert, DailyStat, InboxStatus, LLMReport, Record, Report +from app.models import Alert, DailyStat, InboxStatus, LLMReport, Record, Report, TLSReport def _pct(pass_count: int, total: int) -> str: @@ -110,7 +110,9 @@ def homepage_summary( date_to: str | None = None, ) -> dict: start, end, scope_label = resolve_date_range(session, period=period, domain=domain, date_from=date_from, date_to=date_to) - domains = session.scalar(select(func.count(func.distinct(Report.domain)))) or 0 + dmarc_domains = set(session.execute(select(Report.domain).distinct()).scalars().all()) + tls_domains = set(session.execute(select(TLSReport.domain).distinct()).scalars().all()) + domains = len(dmarc_domains | tls_domains) reports_stmt = select(func.count(Report.id)) records_stmt = select(Record).join(Report) unknown_stmt = select(func.count(func.distinct(Record.source_ip))).join(Report).where(Record.is_known_sender.is_(False)) diff --git a/app/llm.py b/app/llm.py index a85ceb6..a63a20e 100644 --- a/app/llm.py +++ b/app/llm.py @@ -15,7 +15,7 @@ from app.models import Alert logger = logging.getLogger(__name__) SYSTEM_PROMPT = ( - "You are an expert email authentication and DMARC operations analyst. You explain deterministic DMARC telemetry " + "You are an expert email authentication, DMARC, and SMTP TLS reporting operations analyst. You explain deterministic telemetry " "to a business owner/admin. You must not invent facts. You must distinguish confirmed facts from likely " "interpretations. You must never claim an account is compromised solely from DMARC aggregate failures. You must " "provide practical next steps. Output only valid JSON matching the requested schema." @@ -32,6 +32,15 @@ ALERT_PROMPT = ( "object with these keys: summary, risk, recommended_action, confidence." ) +TLS_ALERT_PROMPT = ( + "Explain this SMTP TLS aggregate report alert to a business owner/admin. Be precise, do not invent facts, and base " + "the explanation only on the supplied deterministic TLSRPT facts. Explain that TLSRPT covers delivery-session TLS " + "success or failure against policies such as MTA-STS, not DMARC authentication. Mention affected MX hosts, failure " + "counts/rates, and result types when supplied. Recommend concrete checks such as certificate hostname coverage, " + "certificate chain validity, MTA-STS policy contents, MX hostnames, and mail server TLS configuration. Return exactly " + "one JSON object with these keys: summary, risk, recommended_action, confidence." +) + POSTURE_DIGEST_PROMPT = ( "Write a current DMARC posture report for the admin using all supplied deterministic telemetry and all open alerts. " "Base the report on unresolved/open risk, not only one report day. Mention exact counts/rates, important failing or " @@ -148,6 +157,13 @@ def normalize_summary_output(output: dict[str, Any], payload: dict[str, Any]) -> def fallback_alert_explanation(alert: Alert | Any) -> AlertExplanation: + if str(getattr(alert, "type", "")).startswith("tls_"): + return AlertExplanation( + summary=getattr(alert, "summary", "A deterministic SMTP TLS reporting alert was created."), + risk="Review the TLSRPT facts. TLS aggregate reports describe SMTP TLS delivery-session failures and do not by themselves prove message loss or account compromise.", + recommended_action="Check the affected MX hostnames, certificate chain and hostname coverage, MTA-STS policy, and mail server TLS configuration before changing policy.", + confidence="fallback", + ) return AlertExplanation( summary=getattr(alert, "summary", "DMARC Sentinel created a deterministic alert."), risk="Review the deterministic facts. DMARC aggregate data alone does not prove mailbox compromise.", @@ -195,19 +211,20 @@ class LLMClient: raise RuntimeError(f"LLM call failed: {last_error}") def explain_alert(self, alert: Alert) -> AlertExplanation: + is_tls = str(alert.type).startswith("tls_") payload = { - "task": "explain_dmarc_alert", + "task": "explain_tlsrpt_alert" if is_tls else "explain_dmarc_alert", "domain": alert.domain, "severity": alert.severity, "alert_type": alert.type, "facts": json.loads(alert.details_json or "{}"), "required_json_schema": { "summary": "string, one concise sentence based only on the supplied facts", - "risk": "string, business/operational risk without claiming compromise from aggregate data alone", + "risk": "string, business/operational risk without claiming unsupported compromise or data loss", "recommended_action": "string, concrete next step for the admin", "confidence": "low|medium|high", }, - "instruction": self._prompt(self.settings.llm.alert_prompt_path, ALERT_PROMPT), + "instruction": TLS_ALERT_PROMPT if is_tls else self._prompt(self.settings.llm.alert_prompt_path, ALERT_PROMPT), } last_error: Exception | None = None for _ in range(2): diff --git a/app/main.py b/app/main.py index 0c7916c..aab8f37 100644 --- a/app/main.py +++ b/app/main.py @@ -22,7 +22,7 @@ from app.dns_policy import DomainDnsPolicy, collect_domain_dns_policy from app.homepage import domain_homepage_summary, domain_metrics, homepage_summary, latest_summary, resolve_date_range, traffic_distribution from app.inbox_locks import InboxRunLease, inbox_run_locks from app.jobs import import_jobs -from app.models import Alert, DailyStat, DomainDnsSnapshot, InboxStatus, LLMReport, Record, Report, SkippedReportPayload, utcnow +from app.models import Alert, DailyStat, DomainDnsSnapshot, InboxStatus, LLMReport, Record, Report, SkippedReportPayload, TLSPolicy, TLSReport, utcnow from app.scheduler import generate_open_posture_summaries, scheduler_ok, start_scheduler from app.schemas import BacklogRequest, ProcessNowRequest from app.message_processor import process_inbox @@ -100,7 +100,7 @@ def health(): @app.get("/", response_class=HTMLResponse, dependencies=[Depends(require_dashboard_auth)]) def index(request: Request, session: Session = Depends(get_db)): data = homepage_summary(session, period="all") - domains = session.execute(select(Report.domain).distinct().order_by(Report.domain)).scalars().all() + domains = _domain_list(session) alerts_total = session.scalar(select(func.count(Alert.id)).where(Alert.status == "open")) or 0 alerts = _alert_views(session.execute(select(Alert).where(Alert.status == "open")).scalars().all(), session)[:5] traffic = traffic_distribution(session, period="all") @@ -209,8 +209,12 @@ def _alert_view(alert: Alert, session: Session | None = None) -> SimpleNamespace override_type = receiver_action.get("override_type") report_db_id = details.get("report_db_id") report_db_ids = details.get("report_db_ids") if isinstance(details.get("report_db_ids"), list) else [] + tls_report_db_id = report_db_id if details.get("report_type") == "tlsrpt" else None if not report_db_id and isinstance(details.get("report_db_ids"), list) and details["report_db_ids"]: report_db_id = details["report_db_ids"][-1] + if details.get("report_type") == "tlsrpt": + report_db_id = None + report_db_ids = [] if alert.type in {"sudden_unknown_failure_spike"}: report_db_id = None report_time = ( @@ -242,6 +246,8 @@ def _alert_view(alert: Alert, session: Session | None = None) -> SimpleNamespace report_time=report_time, report_db_id=report_db_id, report_db_ids=report_db_ids, + tls_report_db_id=tls_report_db_id, + source_report_url=f"/tls-reports/{tls_report_db_id}" if tls_report_db_id else (f"/reports/{report_db_id}" if report_db_id else None), source_ip=details.get("source_ip"), published_policy=published_policy, receiver_action=receiver_action, @@ -327,6 +333,49 @@ def _domain_sources(session: Session, domain: str) -> list[SimpleNamespace]: ] +def _report_stamp(report: Report | TLSReport) -> datetime: + value = report.date_end or report.date_begin or report.created_at + return _parse_dt(value.isoformat() if hasattr(value, "isoformat") else str(value)) or datetime.now(timezone.utc) + + +def _domain_report_rows(session: Session, domain: str, page: int, page_size: int) -> tuple[list[SimpleNamespace], int]: + dmarc_reports = session.execute(select(Report).where(Report.domain == domain)).scalars().all() + tls_reports = session.execute(select(TLSReport).where(TLSReport.domain == domain)).scalars().all() + rows = [ + SimpleNamespace( + id=report.id, + kind="dmarc", + label="DMARC", + icon="description", + url=f"/reports/{report.id}", + org_name=report.org_name, + report_id=report.report_id, + date_begin=report.date_begin, + date_end=report.date_end, + stamp=_report_stamp(report), + ) + for report in dmarc_reports + ] + rows.extend( + SimpleNamespace( + id=report.id, + kind="tlsrpt", + label="TLS", + icon="lock", + url=f"/tls-reports/{report.id}", + org_name=report.org_name, + report_id=report.report_id, + date_begin=report.date_begin, + date_end=report.date_end, + stamp=_report_stamp(report), + ) + for report in tls_reports + ) + rows.sort(key=lambda item: item.stamp, reverse=True) + total = len(rows) + return rows[(page - 1) * page_size : page * page_size], total + + def _source_history(session: Session, domain: str, source_ip: str | None, alert_type: str, report_db_id: int | None) -> str | None: if not source_ip: return None @@ -400,6 +449,12 @@ def _json_list(value: str | None) -> list: return data if isinstance(data, list) else [] +def _domain_list(session: Session) -> list[str]: + dmarc_domains = set(session.execute(select(Report.domain).distinct()).scalars().all()) + tls_domains = set(session.execute(select(TLSReport.domain).distinct()).scalars().all()) + return sorted(dmarc_domains | tls_domains) + + def _snapshot_model(domain: str, policy: DomainDnsPolicy) -> DomainDnsSnapshot: return DomainDnsSnapshot( domain=domain, @@ -472,14 +527,7 @@ def domain_page(domain: str, request: Request, source_page: int = 1, alert_page: alerts = all_alerts[(alert_page - 1) * alert_page_size : alert_page * alert_page_size] report_page_size = 20 report_page = max(1, report_page) - report_total = session.scalar(select(func.count(Report.id)).where(Report.domain == domain)) or 0 - reports = session.execute( - select(Report) - .where(Report.domain == domain) - .order_by(desc(func.coalesce(Report.date_end, Report.date_begin, Report.created_at))) - .offset((report_page - 1) * report_page_size) - .limit(report_page_size) - ).scalars().all() + reports, report_total = _domain_report_rows(session, domain, report_page, report_page_size) reporters = session.execute( select(Report.org_name, func.count(Report.id)).where(Report.domain == domain).group_by(Report.org_name).order_by(desc(func.count(Report.id))).limit(10) ).all() @@ -545,6 +593,33 @@ def report_page(report_id: int, request: Request, session: Session = Depends(get return templates.TemplateResponse("report.html", {"request": request, "report": report, "alerts": related_alerts}) +@app.get("/tls-reports/{report_id}", response_class=HTMLResponse, dependencies=[Depends(require_dashboard_auth)]) +def tls_report_page(report_id: int, request: Request, session: Session = Depends(get_db)): + report = session.scalar( + select(TLSReport) + .options(selectinload(TLSReport.policies).selectinload(TLSPolicy.failures)) + .where(TLSReport.id == report_id) + ) + if not report: + raise HTTPException(status_code=404) + for policy in report.policies: + policy.policy_strings = _json_list(policy.policy_strings_json) + policy.mx_hosts = _json_list(policy.mx_hosts_json) + policy.total_sessions = policy.total_successful_session_count + policy.total_failure_session_count + policy.failure_rate = ( + policy.total_failure_session_count / policy.total_sessions * 100 + if policy.total_sessions + else 0 + ) + related_alerts = [] + domain_alerts = session.execute(select(Alert).where(Alert.domain == report.domain)).scalars().all() + for alert in domain_alerts: + details = _alert_details(alert) + if details.get("report_type") == "tlsrpt" and details.get("report_db_id") == report.id: + related_alerts.append(_alert_view(alert, session)) + return templates.TemplateResponse("tls_report.html", {"request": request, "report": report, "alerts": related_alerts}) + + @app.get("/alerts", response_class=HTMLResponse, dependencies=[Depends(require_dashboard_auth)]) def alerts_page( request: Request, @@ -729,7 +804,7 @@ def api_homepage_domain(domain: str, session: Session = Depends(get_db)): @app.get("/api/domains", dependencies=[Depends(require_dashboard_auth)]) def api_domains(session: Session = Depends(get_db)): - return {"domains": session.execute(select(Report.domain).distinct()).scalars().all()} + return {"domains": _domain_list(session)} def _overview_payload(session: Session, period: str = "all", domain: str | None = None, date_from: str | None = None, date_to: str | None = None) -> dict: @@ -757,6 +832,9 @@ def api_traffic(period: str = "all", domain: str | None = None, date_from: str | def _latest_report_day(session: Session) -> date | None: latest = session.scalar(select(func.max(func.coalesce(Report.date_end, Report.date_begin, Report.created_at)))) + latest_tls = session.scalar(select(func.max(func.coalesce(TLSReport.date_end, TLSReport.date_begin, TLSReport.created_at)))) + if latest_tls and (not latest or latest_tls > latest): + latest = latest_tls if isinstance(latest, str): latest = _parse_dt(latest) return latest.date() if latest else None diff --git a/app/message_processor.py b/app/message_processor.py index ac3f1ed..a87974e 100644 --- a/app/message_processor.py +++ b/app/message_processor.py @@ -20,7 +20,9 @@ from app.dmarc_parser import DMARCParseError, parse_dmarc_xml from app.imap_client import IMAPClient from app.known_senders import classify_record from app.llm import LLMClient -from app.models import Alert, AuthResult, InboxStatus, MailMessage, Record, Report, SkippedReportPayload, utcnow +from app.models import Alert, AuthResult, InboxStatus, MailMessage, Record, Report, SkippedReportPayload, TLSFailure, TLSPolicy, TLSReport, utcnow +from app.tlsrpt_analyzer import analyze_tls_report +from app.tlsrpt_parser import TLSRPTParseError, parse_tlsrpt_json logger = logging.getLogger(__name__) @@ -79,6 +81,8 @@ def is_candidate_message(message: Message, inbox: InboxConfig) -> bool: return ( inbox.recipient.lower() in recipients or "dmarc" in subject.lower() + or "tls" in subject.lower() + or "tlsrpt" in subject.lower() or inbox.domain.lower() in subject.lower() or "report domain" in subject.lower() or message_has_candidate_attachment(message) @@ -164,7 +168,7 @@ def _record_ingestion_rejection( domain=inbox.domain, severity="warning", type="ingestion_rejected", - title=f"DMARC payload rejected for {inbox.label}", + title=f"Report payload rejected for {inbox.label}", summary=f"A message in {inbox.folder} was rejected during {stage}: {reason}", details_json=json.dumps(details, sort_keys=True, default=str), first_seen_at=now, @@ -186,6 +190,17 @@ def _duplicate_report_sample(existing: Report, mail: MailMessage) -> dict[str, s } +def _duplicate_tls_report_sample(existing: TLSReport, mail: MailMessage) -> dict[str, str | int | None]: + return { + "existing_report_db_id": None, + "existing_report_id": existing.report_id, + "reporting_org": existing.org_name, + "report_date": (existing.date_end or existing.date_begin).date().isoformat() if (existing.date_end or existing.date_begin) else None, + "duplicate_message_uid": mail.imap_uid, + "duplicate_message_id": mail.message_id, + } + + def _record_duplicate_report_payload(session: Session, inbox: InboxConfig, mail: MailMessage, existing: Report, sha256: str) -> None: skipped = session.scalar( select(SkippedReportPayload).where( @@ -216,6 +231,35 @@ def _record_duplicate_report_payload(session: Session, inbox: InboxConfig, mail: skipped.report_date = report_date +def _record_duplicate_tls_report_payload(session: Session, inbox: InboxConfig, mail: MailMessage, existing: TLSReport, sha256: str) -> None: + skipped = session.scalar( + select(SkippedReportPayload).where( + SkippedReportPayload.inbox_id == inbox.id, + SkippedReportPayload.folder == mail.folder, + SkippedReportPayload.imap_uid == mail.imap_uid, + SkippedReportPayload.raw_xml_sha256 == sha256, + SkippedReportPayload.reason == "duplicate_tls_report_payload", + ) + ) + report_date = (existing.date_end or existing.date_begin).date() if (existing.date_end or existing.date_begin) else None + if not skipped: + skipped = SkippedReportPayload( + inbox_id=inbox.id, + folder=mail.folder, + imap_uid=mail.imap_uid, + message_id=mail.message_id, + mail_message_id=mail.id, + reason="duplicate_tls_report_payload", + raw_xml_sha256=sha256, + ) + session.add(skipped) + skipped.message_id = mail.message_id + skipped.mail_message_id = mail.id + skipped.report_identifier = existing.report_id + skipped.reporting_org = existing.org_name + skipped.report_date = report_date + + def _store_report(session: Session, settings: Settings, inbox: InboxConfig, mail: MailMessage, extracted) -> tuple[Report | None, Report | None]: existing = session.scalar(select(Report).where(Report.raw_xml_sha256 == extracted.sha256)) if existing: @@ -286,6 +330,75 @@ def _store_report(session: Session, settings: Settings, inbox: InboxConfig, mail return report, None +def _store_tls_report(session: Session, settings: Settings, inbox: InboxConfig, mail: MailMessage, extracted) -> tuple[TLSReport | None, TLSReport | None]: + existing = session.scalar(select(TLSReport).where(TLSReport.raw_json_sha256 == extracted.sha256)) + if existing: + return None, existing + parsed = parse_tlsrpt_json( + extracted.payload, + max_policies=settings.app.max_xml_records_per_report, + max_failures_per_policy=settings.app.max_xml_records_per_report, + max_session_count=settings.app.max_record_count, + max_future_days=settings.app.max_report_future_days, + max_past_days=settings.app.max_report_past_days, + ) + domains = {policy.policy_domain for policy in parsed.policies} + mismatches = sorted(domain for domain in domains if not _domain_equal(domain, inbox.domain)) + if mismatches: + raise TLSRPTParseError(f"TLS report domain {', '.join(mismatches)} does not match inbox domain {inbox.domain}") + report = TLSReport( + inbox_id=inbox.id, + mail_message_id=mail.id, + raw_json_sha256=extracted.sha256, + report_id=parsed.report_id, + org_name=parsed.org_name, + contact_info=parsed.contact_info, + domain=parsed.domain, + date_begin=parsed.date_begin, + date_end=parsed.date_end, + ) + session.add(report) + session.flush() + for parsed_policy in parsed.policies: + policy = TLSPolicy( + tls_report=report, + policy_type=parsed_policy.policy_type, + policy_domain=parsed_policy.policy_domain, + policy_strings_json=json.dumps(parsed_policy.policy_strings, sort_keys=True), + mx_hosts_json=json.dumps(parsed_policy.mx_hosts, sort_keys=True), + total_successful_session_count=parsed_policy.total_successful_session_count, + total_failure_session_count=parsed_policy.total_failure_session_count, + ) + session.add(policy) + session.flush() + for parsed_failure in parsed_policy.failures: + session.add( + TLSFailure( + policy=policy, + result_type=parsed_failure.result_type, + sending_mta_ip=parsed_failure.sending_mta_ip, + receiving_mx_hostname=parsed_failure.receiving_mx_hostname, + receiving_mx_helo=parsed_failure.receiving_mx_helo, + receiving_ip=parsed_failure.receiving_ip, + failed_session_count=parsed_failure.failed_session_count, + additional_information=parsed_failure.additional_information, + failure_reason_code=parsed_failure.failure_reason_code, + ) + ) + session.flush() + return report, None + + +def _detect_report_kind(filename: str, payload: bytes) -> str: + lower = filename.lower() + stripped = payload.lstrip() + if stripped.startswith(b"<") or lower.endswith(".xml"): + return "dmarc" + if stripped.startswith(b"{") or lower.endswith(".json"): + return "tlsrpt" + raise DMARCParseError("Unsupported report payload: expected DMARC XML or TLSRPT JSON") + + def process_inbox( session: Session, settings: Settings, @@ -338,26 +451,49 @@ def process_inbox( ) imported_any = False for extracted in reports: - report, duplicate_report = _store_report(session, settings, inbox, mail, extracted) - if duplicate_report: - _record_duplicate_report_payload(session, inbox, mail, duplicate_report, extracted.sha256) - summary.duplicate_reports_skipped += 1 - if summary.duplicate_report_samples is None: - summary.duplicate_report_samples = [] - if len(summary.duplicate_report_samples) < 100: - summary.duplicate_report_samples.append(_duplicate_report_sample(duplicate_report, mail)) - continue - if report: - imported_any = True - summary.valid_reports_imported += 1 - summary.records_imported += len(report.records) - alerts = analyze_report(session, settings, report, llm=llm) - new_alerts = [item for item in alerts if item[1]] - summary.alerts_created += len(new_alerts) - summary.llm_explanations_generated += len([item for item in alerts if item[1] and item[0].severity in {"warning", "critical"}]) - for alert, is_new, severity_increased in alerts: - if is_new or severity_increased: - send_alert_email(settings, alert, severity_increased=severity_increased) + kind = _detect_report_kind(extracted.filename, extracted.payload) + if kind == "dmarc": + report, duplicate_report = _store_report(session, settings, inbox, mail, extracted) + if duplicate_report: + _record_duplicate_report_payload(session, inbox, mail, duplicate_report, extracted.sha256) + summary.duplicate_reports_skipped += 1 + if summary.duplicate_report_samples is None: + summary.duplicate_report_samples = [] + if len(summary.duplicate_report_samples) < 100: + summary.duplicate_report_samples.append(_duplicate_report_sample(duplicate_report, mail)) + continue + if report: + imported_any = True + summary.valid_reports_imported += 1 + summary.records_imported += len(report.records) + alerts = analyze_report(session, settings, report, llm=llm) + new_alerts = [item for item in alerts if item[1]] + summary.alerts_created += len(new_alerts) + summary.llm_explanations_generated += len([item for item in alerts if item[1] and item[0].severity in {"warning", "critical"}]) + for alert, is_new, severity_increased in alerts: + if is_new or severity_increased: + send_alert_email(settings, alert, severity_increased=severity_increased) + else: + tls_report, duplicate_tls_report = _store_tls_report(session, settings, inbox, mail, extracted) + if duplicate_tls_report: + _record_duplicate_tls_report_payload(session, inbox, mail, duplicate_tls_report, extracted.sha256) + summary.duplicate_reports_skipped += 1 + if summary.duplicate_report_samples is None: + summary.duplicate_report_samples = [] + if len(summary.duplicate_report_samples) < 100: + summary.duplicate_report_samples.append(_duplicate_tls_report_sample(duplicate_tls_report, mail)) + continue + if tls_report: + imported_any = True + summary.valid_reports_imported += 1 + summary.records_imported += len(tls_report.policies) + alerts = analyze_tls_report(session, settings, tls_report, llm=llm) + new_alerts = [item for item in alerts if item[1]] + summary.alerts_created += len(new_alerts) + summary.llm_explanations_generated += len([item for item in alerts if item[1] and item[0].severity in {"warning", "critical"}]) + for alert, is_new, severity_increased in alerts: + if is_new or severity_increased: + send_alert_email(settings, alert, severity_increased=severity_increased) mail.status = "success" if imported_any else "skipped" mail.error = None mail.processed_at = utcnow() @@ -376,13 +512,19 @@ def process_inbox( mail.status = "failed" mail.error = str(exc) mail.processed_at = utcnow() - if isinstance(exc, (AttachmentExtractionError, DMARCParseError)): + if isinstance(exc, (AttachmentExtractionError, DMARCParseError, TLSRPTParseError)): alert, is_new = _record_ingestion_rejection( session, inbox, mail, str(exc), - stage="attachment extraction" if isinstance(exc, AttachmentExtractionError) else "DMARC XML validation", + stage=( + "attachment extraction" + if isinstance(exc, AttachmentExtractionError) + else "TLSRPT JSON validation" + if isinstance(exc, TLSRPTParseError) + else "report payload validation" + ), ) summary.rejected_messages += 1 if is_new: diff --git a/app/models.py b/app/models.py index 0f602b0..19de294 100644 --- a/app/models.py +++ b/app/models.py @@ -52,6 +52,7 @@ class MailMessage(Base): created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) reports: Mapped[list["Report"]] = relationship(back_populates="mail_message") + tls_reports: Mapped[list["TLSReport"]] = relationship(back_populates="mail_message") class Report(Base): @@ -124,6 +125,60 @@ class AuthResult(Base): record: Mapped[Record] = relationship(back_populates="auth_results") +class TLSReport(Base): + __tablename__ = "tls_reports" + + id: Mapped[int] = mapped_column(primary_key=True) + inbox_id: Mapped[str] = mapped_column(String(120), index=True) + mail_message_id: Mapped[int | None] = mapped_column(ForeignKey("mail_messages.id")) + raw_json_sha256: Mapped[str] = mapped_column(String(64), unique=True, index=True) + report_id: Mapped[str | None] = mapped_column(String(500), index=True) + org_name: Mapped[str | None] = mapped_column(String(255), index=True) + contact_info: Mapped[str | None] = mapped_column(Text) + domain: Mapped[str] = mapped_column(String(255), index=True) + date_begin: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), index=True) + date_end: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), index=True) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) + + mail_message: Mapped[MailMessage | None] = relationship(back_populates="tls_reports") + policies: Mapped[list["TLSPolicy"]] = relationship(back_populates="tls_report", cascade="all, delete-orphan") + + +class TLSPolicy(Base): + __tablename__ = "tls_policies" + + id: Mapped[int] = mapped_column(primary_key=True) + tls_report_id: Mapped[int] = mapped_column(ForeignKey("tls_reports.id"), index=True) + policy_type: Mapped[str | None] = mapped_column(String(80), index=True) + policy_domain: Mapped[str] = mapped_column(String(255), index=True) + policy_strings_json: Mapped[str] = mapped_column(Text, default="[]") + mx_hosts_json: Mapped[str] = mapped_column(Text, default="[]") + total_successful_session_count: Mapped[int] = mapped_column(Integer, default=0) + total_failure_session_count: Mapped[int] = mapped_column(Integer, default=0) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) + + tls_report: Mapped[TLSReport] = relationship(back_populates="policies") + failures: Mapped[list["TLSFailure"]] = relationship(back_populates="policy", cascade="all, delete-orphan") + + +class TLSFailure(Base): + __tablename__ = "tls_failures" + + id: Mapped[int] = mapped_column(primary_key=True) + tls_policy_id: Mapped[int] = mapped_column(ForeignKey("tls_policies.id"), index=True) + result_type: Mapped[str | None] = mapped_column(String(120), index=True) + sending_mta_ip: Mapped[str | None] = mapped_column(String(80), index=True) + receiving_mx_hostname: Mapped[str | None] = mapped_column(String(255), index=True) + receiving_mx_helo: Mapped[str | None] = mapped_column(String(255)) + receiving_ip: Mapped[str | None] = mapped_column(String(80)) + failed_session_count: Mapped[int] = mapped_column(Integer, default=0) + additional_information: Mapped[str | None] = mapped_column(Text) + failure_reason_code: Mapped[str | None] = mapped_column(String(120)) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) + + policy: Mapped[TLSPolicy] = relationship(back_populates="failures") + + class SkippedReportPayload(Base): __tablename__ = "skipped_report_payloads" __table_args__ = ( diff --git a/app/scheduler.py b/app/scheduler.py index 040c6f6..fda66c3 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -15,7 +15,7 @@ from app.db import session_scope from app.inbox_locks import inbox_run_locks from app.llm import LLMClient from app.message_processor import process_inbox -from app.models import Alert, DailyStat, LLMReport, Record, Report, utcnow +from app.models import Alert, DailyStat, LLMReport, Record, Report, TLSPolicy, TLSReport, utcnow logger = logging.getLogger(__name__) @@ -63,6 +63,53 @@ def _domain_records(session: Session, domain: str, start: datetime, end: datetim ).scalars().all() +def _domain_list(session: Session) -> list[str]: + dmarc_domains = set(session.execute(select(Report.domain).distinct()).scalars().all()) + tls_domains = set(session.execute(select(TLSReport.domain).distinct()).scalars().all()) + return sorted(dmarc_domains | tls_domains) + + +def _tls_date_expr(): + return func.coalesce(TLSReport.date_end, TLSReport.date_begin, TLSReport.created_at) + + +def _tls_bounds(session: Session, domain: str | None = None) -> tuple[datetime | None, datetime | None]: + stmt = select(func.min(_tls_date_expr()), func.max(_tls_date_expr())) + if domain: + stmt = stmt.where(TLSReport.domain == domain) + start, end = session.execute(stmt).one() + return _as_utc(start), _as_utc(end) + + +def _merge_bounds(*bounds: tuple[datetime | None, datetime | None]) -> tuple[datetime, datetime]: + starts = [start for start, _ in bounds if start] + ends = [end for _, end in bounds if end] + period_start = min(starts) if starts else datetime.now(timezone.utc) + period_end = max(ends) if ends else period_start + return period_start, period_end + + +def _tls_metrics(session: Session, domain: str, start: datetime | None = None, end: datetime | None = None) -> dict: + stmt = select(TLSPolicy, TLSReport).join(TLSReport).where(TLSReport.domain == domain) + if start: + stmt = stmt.where(_tls_date_expr() >= start) + if end: + stmt = stmt.where(_tls_date_expr() <= end) + rows = session.execute(stmt).all() + successes = sum(policy.total_successful_session_count for policy, _ in rows) + failures = sum(policy.total_failure_session_count for policy, _ in rows) + total = successes + failures + reports = {report.id for _, report in rows} + reporters = sorted({report.org_name for _, report in rows if report.org_name}) + return { + "tls_reports": len(reports), + "tls_successful_sessions": successes, + "tls_failed_sessions": failures, + "tls_failure_rate": round(failures / total * 100, 2) if total else 0, + "tls_reporters": reporters[:10], + } + + def aggregate_daily_stats(session: Session, domain: str, day: date) -> DailyStat: start = datetime.combine(day, time.min, tzinfo=timezone.utc) end = start + timedelta(days=1) @@ -141,6 +188,7 @@ def _summary_payload(session: Session, domain: str, day: date, stat: DailyStat) "unknown_sources": stat.unknown_source_count, "critical_alerts": critical, "warnings": warnings, + **_tls_metrics(session, domain, period_start, period_end), }, "top_sources": json.loads(stat.top_sources_json or "[]"), "reporters": [{"org": org or "unknown", "reports": count} for org, count in reports], @@ -169,8 +217,7 @@ def _posture_payload(session: Session, domain: str) -> tuple[dict, datetime, dat func.max(func.coalesce(Report.date_end, Report.date_begin, Report.created_at)), ).where(Report.domain == domain) ).one() - period_start = _as_utc(bounds[0]) or datetime.now(timezone.utc) - period_end = _as_utc(bounds[1]) or period_start + period_start, period_end = _merge_bounds((_as_utc(bounds[0]), _as_utc(bounds[1])), _tls_bounds(session, domain)) records = session.execute(select(Record).join(Report).where(Report.domain == domain)).scalars().all() reports = session.execute( select(Report.org_name, func.count(Report.id)) @@ -214,6 +261,7 @@ def _posture_payload(session: Session, domain: str) -> tuple[dict, datetime, dat "unknown_failing_sources": len({row.source_ip for row in failing_unknown}), "open_critical_alerts": len([alert for alert in alerts if alert.severity == "critical"]), "open_warnings": len([alert for alert in alerts if alert.severity == "warning"]), + **_tls_metrics(session, domain), }, "top_sources": [ { @@ -250,7 +298,7 @@ def _posture_payload(session: Session, domain: str) -> tuple[dict, datetime, dat def _portfolio_posture_payload(session: Session) -> tuple[dict, datetime, datetime] | None: - domains = session.execute(select(Report.domain).distinct().order_by(Report.domain)).scalars().all() + domains = _domain_list(session) if not domains: return None bounds = session.execute( @@ -259,8 +307,7 @@ def _portfolio_posture_payload(session: Session) -> tuple[dict, datetime, dateti func.max(func.coalesce(Report.date_end, Report.date_begin, Report.created_at)), ) ).one() - period_start = _as_utc(bounds[0]) or datetime.now(timezone.utc) - period_end = _as_utc(bounds[1]) or period_start + period_start, period_end = _merge_bounds((_as_utc(bounds[0]), _as_utc(bounds[1])), _tls_bounds(session)) records = session.execute(select(Record).join(Report)).scalars().all() alerts = session.execute( select(Alert) @@ -286,6 +333,7 @@ def _portfolio_posture_payload(session: Session) -> tuple[dict, datetime, dateti "unknown_sources": len({row.source_ip for row in domain_records if not row.is_known_sender}), "open_critical_alerts": len([alert for alert in domain_alerts if alert.severity == "critical"]), "open_warnings": len([alert for alert in domain_alerts if alert.severity == "warning"]), + **_tls_metrics(session, domain), } ) top_alerts = [ @@ -322,6 +370,9 @@ def _portfolio_posture_payload(session: Session) -> tuple[dict, datetime, dateti "unknown_failing_sources": len({row.source_ip for row in failing_unknown}), "open_critical_alerts": len([alert for alert in alerts if alert.severity == "critical"]), "open_warnings": len([alert for alert in alerts if alert.severity == "warning"]), + "tls_reports": session.scalar(select(func.count(TLSReport.id))) or 0, + "tls_failed_sessions": sum(row["tls_failed_sessions"] for row in domain_rows), + "tls_successful_sessions": sum(row["tls_successful_sessions"] for row in domain_rows), }, "domain_posture": domain_rows, "top_open_alerts": top_alerts, @@ -376,7 +427,7 @@ def generate_open_posture_summaries(settings: Settings, *, force: bool = True) - report.plain_text = plain generated.append(report) send_digest_email(settings, "DMARC Sentinel portfolio posture summary", plain) - domains = session.execute(select(Report.domain).distinct()).scalars().all() + domains = _domain_list(session) for domain in domains: payload, period_start, period_end = _posture_payload(session, domain) existing = session.scalar( @@ -421,7 +472,7 @@ def generate_daily_summaries(settings: Settings, target_day: date | None = None, llm = LLMClient(settings) generated: list[LLMReport] = [] with session_scope() as session: - domains = session.execute(select(Report.domain).distinct()).scalars().all() + domains = _domain_list(session) for domain in domains: stat = aggregate_daily_stats(session, domain, target_day) payload = _summary_payload(session, domain, target_day, stat) @@ -473,7 +524,7 @@ def generate_weekly_summaries(settings: Settings) -> list[LLMReport]: llm = LLMClient(settings) generated: list[LLMReport] = [] with session_scope() as session: - domains = session.execute(select(Report.domain).distinct()).scalars().all() + domains = _domain_list(session) for domain in domains: records = _domain_records(session, domain, start, end) existing = session.scalar( diff --git a/app/templates/alerts.html b/app/templates/alerts.html index 40fd740..be1465a 100644 --- a/app/templates/alerts.html +++ b/app/templates/alerts.html @@ -94,8 +94,8 @@ {% if alert.source_history %} {{ alert.source_history }} {% endif %} - {% if alert.report_db_id %} - Source report + {% if alert.source_report_url %} + Source report {% endif %} {% if alert.published_policy_label or alert.receiver_action_label %} diff --git a/app/templates/domain.html b/app/templates/domain.html index 03dee7c..13cceb3 100644 --- a/app/templates/domain.html +++ b/app/templates/domain.html @@ -167,7 +167,7 @@

Open Alerts ({{ alert_total }})

{% for alert in alerts %} - + {{ alert.severity }} @@ -279,10 +279,10 @@

Recent Reports

{% for report in reports %} - - description + + {{ report.icon }} - {{ report.org_name or "unknown" }} · {{ report.report_id or report.id }} + {{ report.label }} · {{ report.org_name or "unknown" }} · {{ report.report_id or report.id }} {{ report.date_begin | fmt_dt }}{% if report.date_end %} to {{ report.date_end | fmt_dt }}{% endif %} arrow_forward_ios diff --git a/app/templates/tls_report.html b/app/templates/tls_report.html new file mode 100644 index 0000000..56b0d12 --- /dev/null +++ b/app/templates/tls_report.html @@ -0,0 +1,128 @@ +{% extends "base.html" %} +{% block content %} +
+

TLS Report {{ report.id }}

+

{{ report.domain }} · {{ report.org_name or "unknown organization" }}

+
+ +
+
+ Report Org + {{ report.org_name or "unknown" }} +
+
+ Report ID + {{ report.report_id or report.id }} +
+
+ Date Range + {{ report.date_begin | fmt_dt }}
{{ report.date_end | fmt_dt }} +
+
+ Contact + {{ report.contact_info or "not reported" }} +
+
+ +
+

Alerts From This TLS Report

+
+ {% for alert in alerts %} + + + {{ alert.severity }} + + + {{ alert.title }} +

{{ alert.llm_summary or alert.summary }}

+ + {% else %} +
No alerts are linked to this TLS report.
+ {% endfor %} +
+
+ +
+

Policies

+
+
+ + + + + + + + + + + + {% for policy in report.policies %} + + + + + + + + {% else %} + + + + {% endfor %} + +
PolicyMX HostsSuccessful SessionsFailed SessionsFailure Rate
+ {{ policy.policy_type or "unknown" }} + {{ policy.policy_domain }} + + {% for host in policy.mx_hosts %} + {{ host }} + {% else %} + not reported + {% endfor %} + {{ policy.total_successful_session_count }}{{ policy.total_failure_session_count }}{{ "%.1f"|format(policy.failure_rate) }}%
No policies found in this report.
+
+
+
+ +
+

Failure Details

+
+
+ + + + + + + + + + + + + {% set ns = namespace(rows=0) %} + {% for policy in report.policies %} + {% for failure in policy.failures %} + {% set ns.rows = ns.rows + 1 %} + + + + + + + + + {% endfor %} + {% endfor %} + {% if ns.rows == 0 %} + + + + {% endif %} + +
ResultFailed SessionsReceiving MXSending MTA IPReceiving IPReason
{{ failure.result_type or "unknown" }}{{ failure.failed_session_count }}{{ failure.receiving_mx_hostname or failure.receiving_mx_helo or "not reported" }}{{ failure.sending_mta_ip or "not reported" }}{{ failure.receiving_ip or "not reported" }}{{ failure.failure_reason_code or failure.additional_information or "not reported" }}
No failure details were reported.
+
+
+
+{% endblock %} diff --git a/app/tlsrpt_analyzer.py b/app/tlsrpt_analyzer.py new file mode 100644 index 0000000..bca17ec --- /dev/null +++ b/app/tlsrpt_analyzer.py @@ -0,0 +1,119 @@ +from __future__ import annotations + +import json +from typing import Any + +from sqlalchemy.orm import Session + +from app.analyzer import create_or_update_alert +from app.config import Settings +from app.llm import LLMClient +from app.models import Alert, TLSReport + + +def _json_list(value: str | None) -> list: + try: + parsed = json.loads(value or "[]") + except json.JSONDecodeError: + return [] + return parsed if isinstance(parsed, list) else [] + + +def _report_evidence(report: TLSReport) -> dict[str, Any]: + return { + "report_type": "tlsrpt", + "report_db_id": report.id, + "report_id": report.report_id, + "reporting_orgs": [report.org_name] if report.org_name else [], + "contact_info": report.contact_info, + "date_range": { + "begin": report.date_begin.isoformat() if report.date_begin else None, + "end": report.date_end.isoformat() if report.date_end else None, + }, + } + + +def _severity(failures: int, total: int) -> str: + rate = failures / total * 100 if total else 0 + if failures >= 100 or rate >= 20: + return "critical" + return "warning" + + +def analyze_tls_report(session: Session, settings: Settings, report: TLSReport, llm: LLMClient | None = None) -> list[tuple[Alert, bool, bool]]: + created: list[tuple[Alert, bool, bool]] = [] + evidence = _report_evidence(report) + for policy in report.policies: + successes = policy.total_successful_session_count + failures = policy.total_failure_session_count + total = successes + failures + if failures <= 0: + continue + failure_rate = failures / total * 100 if total else 100 + mx_hosts = _json_list(policy.mx_hosts_json) + details = { + **evidence, + "policy_type": policy.policy_type, + "policy_domain": policy.policy_domain, + "policy_strings": _json_list(policy.policy_strings_json), + "mx_hosts": mx_hosts, + "successful_sessions": successes, + "failed_sessions": failures, + "total_sessions": total, + "failure_rate_percent": round(failure_rate, 2), + } + created.append( + create_or_update_alert( + session, + inbox_id=report.inbox_id, + domain=report.domain, + severity=_severity(failures, total), + alert_type="tls_delivery_failures", + key=f"{policy.policy_domain}:{policy.policy_type or 'unknown'}", + title=f"TLS delivery failures reported for {policy.policy_domain}", + summary=( + f"{report.org_name or 'A reporter'} observed {failures} failed SMTP TLS sessions for " + f"{policy.policy_domain} ({failure_rate:.1f}% of {total or failures} reported sessions)." + ), + details=details, + ) + ) + for failure in policy.failures: + if failure.failed_session_count <= 0: + continue + failure_details = { + **details, + "result_type": failure.result_type, + "sending_mta_ip": failure.sending_mta_ip, + "receiving_mx_hostname": failure.receiving_mx_hostname, + "receiving_mx_helo": failure.receiving_mx_helo, + "receiving_ip": failure.receiving_ip, + "failed_session_count": failure.failed_session_count, + "additional_information": failure.additional_information, + "failure_reason_code": failure.failure_reason_code, + } + target = failure.receiving_mx_hostname or ", ".join(mx_hosts) or policy.policy_domain + created.append( + create_or_update_alert( + session, + inbox_id=report.inbox_id, + domain=report.domain, + severity=_severity(failure.failed_session_count, total), + alert_type="tls_failure_detail", + key=f"{policy.policy_domain}:{failure.result_type or 'unknown'}:{target}:{failure.sending_mta_ip or 'unknown'}", + title=f"TLS failure detail for {target}", + summary=( + f"{failure.failed_session_count} sessions failed with " + f"{failure.result_type or 'an unspecified TLS result'} for {target}." + ), + details=failure_details, + ) + ) + if llm and settings.llm.generate_alert_explanations: + for alert, is_new, severity_increased in created: + if (is_new or severity_increased) and alert.severity in {"warning", "critical"}: + explanation = llm.explain_alert(alert) + alert.llm_summary = explanation.summary + alert.llm_risk = explanation.risk + alert.llm_recommended_action = explanation.recommended_action + return created diff --git a/app/tlsrpt_parser.py b/app/tlsrpt_parser.py new file mode 100644 index 0000000..048ef5d --- /dev/null +++ b/app/tlsrpt_parser.py @@ -0,0 +1,193 @@ +from __future__ import annotations + +import json +from dataclasses import dataclass, field +from datetime import datetime, timedelta, timezone +from typing import Any + + +class TLSRPTParseError(Exception): + pass + + +@dataclass +class ParsedTLSFailure: + result_type: str | None + sending_mta_ip: str | None + receiving_mx_hostname: str | None + receiving_mx_helo: str | None + receiving_ip: str | None + failed_session_count: int + additional_information: str | None = None + failure_reason_code: str | None = None + + +@dataclass +class ParsedTLSPolicy: + policy_type: str | None + policy_domain: str + policy_strings: list[str] = field(default_factory=list) + mx_hosts: list[str] = field(default_factory=list) + total_successful_session_count: int = 0 + total_failure_session_count: int = 0 + failures: list[ParsedTLSFailure] = field(default_factory=list) + + +@dataclass +class ParsedTLSReport: + org_name: str | None + contact_info: str | None + report_id: str | None + date_begin: datetime | None + date_end: datetime | None + domain: str + policies: list[ParsedTLSPolicy] + + +def _as_mapping(value: Any, name: str) -> dict[str, Any]: + if not isinstance(value, dict): + raise TLSRPTParseError(f"{name} must be an object") + return value + + +def _as_list(value: Any, name: str) -> list[Any]: + if value is None: + return [] + if not isinstance(value, list): + raise TLSRPTParseError(f"{name} must be an array") + return value + + +def _string(value: Any) -> str | None: + if value is None: + return None + text = str(value).strip() + return text or None + + +def _string_list(value: Any) -> list[str]: + return [text for item in _as_list(value, "string list") if (text := _string(item))] + + +def _int(value: Any, name: str) -> int: + if value in (None, ""): + return 0 + try: + parsed = int(value) + except (TypeError, ValueError) as exc: + raise TLSRPTParseError(f"{name} must be an integer") from exc + if parsed < 0: + raise TLSRPTParseError(f"{name} must not be negative") + return parsed + + +def _dt(value: Any, name: str) -> datetime | None: + text = _string(value) + if not text: + return None + try: + parsed = datetime.fromisoformat(text.replace("Z", "+00:00")) + except ValueError as exc: + raise TLSRPTParseError(f"{name} must be an ISO-8601 datetime") from exc + if parsed.tzinfo is None: + return parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + + +def _validate_report_dates(date_begin: datetime | None, date_end: datetime | None, max_future_days: int, max_past_days: int) -> None: + now = datetime.now(timezone.utc) + earliest = now - timedelta(days=max_past_days) + latest = now + timedelta(days=max_future_days) + for label, value in {"begin": date_begin, "end": date_end}.items(): + if value is None: + continue + if value < earliest: + raise TLSRPTParseError(f"Report {label} date is older than {max_past_days} days") + if value > latest: + raise TLSRPTParseError(f"Report {label} date is more than {max_future_days} days in the future") + if date_begin and date_end and date_begin > date_end: + raise TLSRPTParseError("Report begin date is after end date") + + +def parse_tlsrpt_json( + payload: bytes, + *, + max_policies: int | None = None, + max_failures_per_policy: int | None = None, + max_session_count: int | None = None, + max_future_days: int = 3, + max_past_days: int = 3650, +) -> ParsedTLSReport: + try: + data = json.loads(payload.decode("utf-8")) + except UnicodeDecodeError as exc: + raise TLSRPTParseError("Invalid UTF-8 JSON") from exc + except json.JSONDecodeError as exc: + raise TLSRPTParseError(f"Invalid JSON: {exc}") from exc + + root = _as_mapping(data, "TLS report") + date_range = _as_mapping(root.get("date-range"), "date-range") + date_begin = _dt(date_range.get("start-datetime"), "date-range.start-datetime") + date_end = _dt(date_range.get("end-datetime"), "date-range.end-datetime") + _validate_report_dates(date_begin, date_end, max_future_days, max_past_days) + + policies_data = _as_list(root.get("policies"), "policies") + if not policies_data: + raise TLSRPTParseError("No TLS policies found") + if max_policies is not None and len(policies_data) > max_policies: + raise TLSRPTParseError(f"Report exceeds policy limit of {max_policies}") + + parsed_policies: list[ParsedTLSPolicy] = [] + for item in policies_data: + policy_item = _as_mapping(item, "policy item") + policy = _as_mapping(policy_item.get("policy"), "policy") + summary = _as_mapping(policy_item.get("summary"), "summary") + domain = _string(policy.get("policy-domain")) + if not domain: + raise TLSRPTParseError("Missing TLS policy domain") + successes = _int(summary.get("total-successful-session-count"), "total-successful-session-count") + failures = _int(summary.get("total-failure-session-count"), "total-failure-session-count") + if max_session_count is not None and max(successes, failures) > max_session_count: + raise TLSRPTParseError(f"Session count exceeds limit of {max_session_count}") + + failure_details: list[ParsedTLSFailure] = [] + for failure_item in _as_list(policy_item.get("failure-details"), "failure-details"): + if max_failures_per_policy is not None and len(failure_details) >= max_failures_per_policy: + raise TLSRPTParseError(f"Policy exceeds failure detail limit of {max_failures_per_policy}") + failure = _as_mapping(failure_item, "failure detail") + failed_count = _int(failure.get("failed-session-count"), "failed-session-count") + if max_session_count is not None and failed_count > max_session_count: + raise TLSRPTParseError(f"Session count exceeds limit of {max_session_count}") + failure_details.append( + ParsedTLSFailure( + result_type=_string(failure.get("result-type")), + sending_mta_ip=_string(failure.get("sending-mta-ip")), + receiving_mx_hostname=_string(failure.get("receiving-mx-hostname")), + receiving_mx_helo=_string(failure.get("receiving-mx-helo")), + receiving_ip=_string(failure.get("receiving-ip")), + failed_session_count=failed_count, + additional_information=_string(failure.get("additional-information")), + failure_reason_code=_string(failure.get("failure-reason-code")), + ) + ) + parsed_policies.append( + ParsedTLSPolicy( + policy_type=_string(policy.get("policy-type")), + policy_domain=domain, + policy_strings=_string_list(policy.get("policy-string")), + mx_hosts=_string_list(policy.get("mx-host")), + total_successful_session_count=successes, + total_failure_session_count=failures, + failures=failure_details, + ) + ) + + return ParsedTLSReport( + org_name=_string(root.get("organization-name")), + contact_info=_string(root.get("contact-info")), + report_id=_string(root.get("report-id")), + date_begin=date_begin, + date_end=date_end, + domain=parsed_policies[0].policy_domain, + policies=parsed_policies, + ) diff --git a/migrations/versions/20260623_0001_tls_reports.py b/migrations/versions/20260623_0001_tls_reports.py new file mode 100644 index 0000000..f0cdb3f --- /dev/null +++ b/migrations/versions/20260623_0001_tls_reports.py @@ -0,0 +1,96 @@ +"""add tls aggregate reports + +Revision ID: 20260623_0001 +Revises: 20260520_0002 +Create Date: 2026-06-23 00:00:00.000000+00:00 +""" +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + +revision = "20260623_0001" +down_revision = "20260520_0002" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "tls_reports", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("inbox_id", sa.String(length=120), nullable=False), + sa.Column("mail_message_id", sa.Integer(), sa.ForeignKey("mail_messages.id"), nullable=True), + sa.Column("raw_json_sha256", sa.String(length=64), nullable=False), + sa.Column("report_id", sa.String(length=500), nullable=True), + sa.Column("org_name", sa.String(length=255), nullable=True), + sa.Column("contact_info", sa.Text(), nullable=True), + sa.Column("domain", sa.String(length=255), nullable=False), + sa.Column("date_begin", sa.DateTime(timezone=True), nullable=True), + sa.Column("date_end", sa.DateTime(timezone=True), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + ) + op.create_index(op.f("ix_tls_reports_date_begin"), "tls_reports", ["date_begin"], unique=False) + op.create_index(op.f("ix_tls_reports_date_end"), "tls_reports", ["date_end"], unique=False) + op.create_index(op.f("ix_tls_reports_domain"), "tls_reports", ["domain"], unique=False) + op.create_index(op.f("ix_tls_reports_inbox_id"), "tls_reports", ["inbox_id"], unique=False) + op.create_index(op.f("ix_tls_reports_org_name"), "tls_reports", ["org_name"], unique=False) + op.create_index(op.f("ix_tls_reports_raw_json_sha256"), "tls_reports", ["raw_json_sha256"], unique=True) + op.create_index(op.f("ix_tls_reports_report_id"), "tls_reports", ["report_id"], unique=False) + + op.create_table( + "tls_policies", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("tls_report_id", sa.Integer(), sa.ForeignKey("tls_reports.id"), nullable=False), + sa.Column("policy_type", sa.String(length=80), nullable=True), + sa.Column("policy_domain", sa.String(length=255), nullable=False), + sa.Column("policy_strings_json", sa.Text(), nullable=False), + sa.Column("mx_hosts_json", sa.Text(), nullable=False), + sa.Column("total_successful_session_count", sa.Integer(), nullable=False), + sa.Column("total_failure_session_count", sa.Integer(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + ) + op.create_index(op.f("ix_tls_policies_policy_domain"), "tls_policies", ["policy_domain"], unique=False) + op.create_index(op.f("ix_tls_policies_policy_type"), "tls_policies", ["policy_type"], unique=False) + op.create_index(op.f("ix_tls_policies_tls_report_id"), "tls_policies", ["tls_report_id"], unique=False) + + op.create_table( + "tls_failures", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("tls_policy_id", sa.Integer(), sa.ForeignKey("tls_policies.id"), nullable=False), + sa.Column("result_type", sa.String(length=120), nullable=True), + sa.Column("sending_mta_ip", sa.String(length=80), nullable=True), + sa.Column("receiving_mx_hostname", sa.String(length=255), nullable=True), + sa.Column("receiving_mx_helo", sa.String(length=255), nullable=True), + sa.Column("receiving_ip", sa.String(length=80), nullable=True), + sa.Column("failed_session_count", sa.Integer(), nullable=False), + sa.Column("additional_information", sa.Text(), nullable=True), + sa.Column("failure_reason_code", sa.String(length=120), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + ) + op.create_index(op.f("ix_tls_failures_receiving_mx_hostname"), "tls_failures", ["receiving_mx_hostname"], unique=False) + op.create_index(op.f("ix_tls_failures_result_type"), "tls_failures", ["result_type"], unique=False) + op.create_index(op.f("ix_tls_failures_sending_mta_ip"), "tls_failures", ["sending_mta_ip"], unique=False) + op.create_index(op.f("ix_tls_failures_tls_policy_id"), "tls_failures", ["tls_policy_id"], unique=False) + + +def downgrade() -> None: + op.drop_index(op.f("ix_tls_failures_tls_policy_id"), table_name="tls_failures") + op.drop_index(op.f("ix_tls_failures_sending_mta_ip"), table_name="tls_failures") + op.drop_index(op.f("ix_tls_failures_result_type"), table_name="tls_failures") + op.drop_index(op.f("ix_tls_failures_receiving_mx_hostname"), table_name="tls_failures") + op.drop_table("tls_failures") + + op.drop_index(op.f("ix_tls_policies_tls_report_id"), table_name="tls_policies") + op.drop_index(op.f("ix_tls_policies_policy_type"), table_name="tls_policies") + op.drop_index(op.f("ix_tls_policies_policy_domain"), table_name="tls_policies") + op.drop_table("tls_policies") + + op.drop_index(op.f("ix_tls_reports_report_id"), table_name="tls_reports") + op.drop_index(op.f("ix_tls_reports_raw_json_sha256"), table_name="tls_reports") + op.drop_index(op.f("ix_tls_reports_org_name"), table_name="tls_reports") + op.drop_index(op.f("ix_tls_reports_inbox_id"), table_name="tls_reports") + op.drop_index(op.f("ix_tls_reports_domain"), table_name="tls_reports") + op.drop_index(op.f("ix_tls_reports_date_end"), table_name="tls_reports") + op.drop_index(op.f("ix_tls_reports_date_begin"), table_name="tls_reports") + op.drop_table("tls_reports") diff --git a/tests/fixtures/sample_tlsrpt.json b/tests/fixtures/sample_tlsrpt.json new file mode 100644 index 0000000..345faaa --- /dev/null +++ b/tests/fixtures/sample_tlsrpt.json @@ -0,0 +1,30 @@ +{ + "organization-name": "Google Inc.", + "date-range": { + "start-datetime": "2026-06-17T00:00:00Z", + "end-datetime": "2026-06-17T23:59:59Z" + }, + "contact-info": "smtp-tls-reporting@google.com", + "report-id": "2026-06-17T00:00:00Z_anamaka.net", + "policies": [ + { + "policy": { + "policy-type": "sts", + "policy-string": [ + "version: STSv1", + "mode: enforce", + "mx: mail.dynamicpress.org", + "max_age: 604800" + ], + "policy-domain": "anamaka.net", + "mx-host": [ + "mail.dynamicpress.org" + ] + }, + "summary": { + "total-successful-session-count": 5, + "total-failure-session-count": 0 + } + } + ] +} diff --git a/tests/test_analyzer.py b/tests/test_analyzer.py index 15afac0..d5909f2 100644 --- a/tests/test_analyzer.py +++ b/tests/test_analyzer.py @@ -7,7 +7,8 @@ from sqlalchemy.orm import Session from app.analyzer import analyze_report from app.config import Settings from app.db import Base -from app.models import Record, Report +from app.models import Record, Report, TLSPolicy, TLSFailure, TLSReport +from app.tlsrpt_analyzer import analyze_tls_report def _session(): @@ -209,3 +210,47 @@ def test_alert_details_include_published_policy_and_receiver_action(): assert details["published_policy"]["effective_source"] == "p" assert details["receiver_action"]["disposition"] == "reject" assert "Published DMARC policy was p=reject; pct=100" in alert.summary + + +def test_tls_failures_create_alert_with_tls_details(): + session = _session() + report = TLSReport( + inbox_id="anamaka", + raw_json_sha256="tls-sha", + report_id="tls-report", + org_name="Google Inc.", + domain="anamaka.net", + date_begin=datetime.now(timezone.utc) - timedelta(hours=1), + date_end=datetime.now(timezone.utc), + ) + session.add(report) + session.flush() + policy = TLSPolicy( + tls_report=report, + policy_type="sts", + policy_domain="anamaka.net", + mx_hosts_json='["mail.dynamicpress.org"]', + policy_strings_json='["version: STSv1", "mode: enforce"]', + total_successful_session_count=80, + total_failure_session_count=20, + ) + session.add(policy) + session.flush() + session.add( + TLSFailure( + policy=policy, + result_type="certificate-host-mismatch", + receiving_mx_hostname="mail.dynamicpress.org", + failed_session_count=20, + ) + ) + session.commit() + + alerts = analyze_tls_report(session, _settings(), report) + + assert any(alert.type == "tls_delivery_failures" for alert, _, _ in alerts) + detail = next(alert for alert, _, _ in alerts if alert.type == "tls_failure_detail") + details = json.loads(detail.details_json) + assert details["report_type"] == "tlsrpt" + assert details["policy_domain"] == "anamaka.net" + assert details["result_type"] == "certificate-host-mismatch" diff --git a/tests/test_app_surface.py b/tests/test_app_surface.py index 178bc42..a7d5bda 100644 --- a/tests/test_app_surface.py +++ b/tests/test_app_surface.py @@ -1,4 +1,7 @@ -from app.main import app +import json + +from app.main import _alert_view, app +from app.models import Alert def test_generated_api_documentation_is_disabled(): @@ -7,3 +10,37 @@ def test_generated_api_documentation_is_disabled(): assert "/docs" not in paths assert "/redoc" not in paths assert "/openapi.json" not in paths + + +def test_tls_report_detail_route_exists(): + paths = {route.path for route in app.routes} + + assert "/tls-reports/{report_id}" in paths + + +def test_tls_alert_links_to_tls_report_detail(): + alert = Alert( + fingerprint="anamaka.net:tls_delivery_failures:anamaka.net:sts", + inbox_id="anamaka", + domain="anamaka.net", + severity="warning", + type="tls_delivery_failures", + title="TLS delivery failures reported for anamaka.net", + summary="summary", + details_json=json.dumps( + { + "report_type": "tlsrpt", + "report_db_id": 42, + "date_range": { + "begin": "2026-06-17T00:00:00+00:00", + "end": "2026-06-17T23:59:59+00:00", + }, + } + ), + ) + + view = _alert_view(alert) + + assert view.report_db_id is None + assert view.tls_report_db_id == 42 + assert view.source_report_url == "/tls-reports/42" diff --git a/tests/test_attachment_extractor.py b/tests/test_attachment_extractor.py index 4045565..5689087 100644 --- a/tests/test_attachment_extractor.py +++ b/tests/test_attachment_extractor.py @@ -13,6 +13,10 @@ def _xml() -> bytes: return Path("tests/fixtures/sample_dmarc.xml").read_bytes() +def _json() -> bytes: + return Path("tests/fixtures/sample_tlsrpt.json").read_bytes() + + def test_gzip_attachment_extraction(): gz = gzip.compress(_xml()) reports = extract_payload("report.xml.gz", "application/octet-stream", gz, 20) @@ -22,6 +26,16 @@ def test_gzip_attachment_extraction(): assert len(reports[0].sha256) == 64 +def test_json_gzip_attachment_extraction(): + gz = gzip.compress(_json()) + reports = extract_payload("report.json.gz", "application/octet-stream", gz, 20) + + assert len(reports) == 1 + assert reports[0].filename == "report.json" + assert reports[0].payload.startswith(b"{") + assert len(reports[0].sha256) == 64 + + def test_zip_attachment_extraction_rejects_traversal(): buf = io.BytesIO() with zipfile.ZipFile(buf, "w") as archive: @@ -47,7 +61,17 @@ def test_zip_attachment_extraction_caps_reports_per_archive(): archive.writestr("one.xml", _xml()) archive.writestr("two.xml", _xml()) - with pytest.raises(AttachmentExtractionError, match="archive XML report limit"): + with pytest.raises(AttachmentExtractionError, match="archive report limit"): + extract_payload("reports.zip", "application/zip", buf.getvalue(), 20, max_reports_per_archive=1) + + +def test_zip_attachment_extraction_caps_json_reports_per_archive(): + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w") as archive: + archive.writestr("one.json", _json()) + archive.writestr("two.json", _json()) + + with pytest.raises(AttachmentExtractionError, match="archive report limit"): extract_payload("reports.zip", "application/zip", buf.getvalue(), 20, max_reports_per_archive=1) diff --git a/tests/test_storage.py b/tests/test_storage.py index 888f062..a1e32ae 100644 --- a/tests/test_storage.py +++ b/tests/test_storage.py @@ -7,7 +7,7 @@ from sqlalchemy.orm import Session from app.config import Settings from app.db import Base from app.attachment_extractor import ExtractedReport -from app.message_processor import _store_report +from app.message_processor import _store_report, _store_tls_report from app.models import MailMessage @@ -32,6 +32,34 @@ def test_duplicate_sha_detection(): assert second_duplicate == report +def test_tls_report_storage_and_duplicate_sha_detection(): + engine = create_engine("sqlite:///:memory:", future=True) + Base.metadata.create_all(engine) + session = Session(engine) + mail = MailMessage(inbox_id="anamaka", imap_uid="1", folder="DMARC", status="skipped") + session.add(mail) + session.commit() + payload = Path("tests/fixtures/sample_tlsrpt.json").read_bytes() + extracted = ExtractedReport("report.json", payload, "1" * 64) + settings = Settings.model_validate({"alerts": {"email": {"enabled": False}}}) + + report, duplicate = _store_tls_report(session, settings, _TLSInbox(), mail, extracted) + session.commit() + second, second_duplicate = _store_tls_report(session, settings, _TLSInbox(), mail, extracted) + + assert report is not None + assert duplicate is None + assert report.domain == "anamaka.net" + assert report.policies[0].total_successful_session_count == 5 + assert second is None + assert second_duplicate == report + + class _Inbox: id = "tukutoi" domain = "tukutoi.com" + + +class _TLSInbox: + id = "anamaka" + domain = "anamaka.net" diff --git a/tests/test_tlsrpt_parser.py b/tests/test_tlsrpt_parser.py new file mode 100644 index 0000000..a3cfb99 --- /dev/null +++ b/tests/test_tlsrpt_parser.py @@ -0,0 +1,36 @@ +from pathlib import Path + +import pytest + +from app.tlsrpt_parser import TLSRPTParseError, parse_tlsrpt_json + + +def test_parser_valid_tlsrpt_report(): + payload = Path("tests/fixtures/sample_tlsrpt.json").read_bytes() + report = parse_tlsrpt_json(payload) + + assert report.org_name == "Google Inc." + assert report.domain == "anamaka.net" + assert report.report_id == "2026-06-17T00:00:00Z_anamaka.net" + assert report.date_begin is not None + assert len(report.policies) == 1 + policy = report.policies[0] + assert policy.policy_type == "sts" + assert policy.policy_domain == "anamaka.net" + assert policy.mx_hosts == ["mail.dynamicpress.org"] + assert policy.total_successful_session_count == 5 + assert policy.total_failure_session_count == 0 + + +def test_parser_rejects_missing_policy_domain(): + payload = Path("tests/fixtures/sample_tlsrpt.json").read_text().replace('"policy-domain": "anamaka.net"', '"policy-domain": ""').encode() + + with pytest.raises(TLSRPTParseError, match="Missing TLS policy domain"): + parse_tlsrpt_json(payload) + + +def test_parser_rejects_negative_session_count(): + payload = Path("tests/fixtures/sample_tlsrpt.json").read_text().replace('"total-failure-session-count": 0', '"total-failure-session-count": -1').encode() + + with pytest.raises(TLSRPTParseError, match="must not be negative"): + parse_tlsrpt_json(payload)