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, )