194 lines
7.3 KiB
Python
194 lines
7.3 KiB
Python
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,
|
|
)
|