618 lines
21 KiB
Python
618 lines
21 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
from collections import defaultdict
|
|
from collections.abc import Mapping, Sequence
|
|
from datetime import datetime, timezone
|
|
from email import policy
|
|
from email.message import EmailMessage
|
|
from email.parser import BytesParser
|
|
from email.utils import parsedate_to_datetime
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.audit.logging import audit_event
|
|
from govoplan_core.core.mail import (
|
|
MailBounceObservationRef,
|
|
MailBounceProcessingProvider,
|
|
)
|
|
from govoplan_mail.backend.db.models import (
|
|
MailBounceObservation,
|
|
MailBounceSource,
|
|
MailDeliveryCommand,
|
|
MailServerProfile,
|
|
)
|
|
from govoplan_mail.backend.sending.imap import (
|
|
get_imap_raw_message,
|
|
list_imap_uids_since,
|
|
)
|
|
from govoplan_mail.backend.server_hierarchy import (
|
|
MailServerHierarchyError,
|
|
hierarchy_context_for_profile,
|
|
resolve_mail_transport,
|
|
)
|
|
|
|
|
|
class MailBounceError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def utcnow() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
def normalize_message_id(value: object | None) -> str | None:
|
|
normalized = " ".join(str(value or "").split())
|
|
return normalized[:998] or None
|
|
|
|
|
|
def configure_bounce_source(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
profile_id: str,
|
|
folder: str = "INBOX",
|
|
imap_server_id: str | None = None,
|
|
imap_credential_id: str | None = None,
|
|
is_active: bool = True,
|
|
created_by_user_id: str | None = None,
|
|
) -> MailBounceSource:
|
|
profile = _profile(session, tenant_id=tenant_id, profile_id=profile_id)
|
|
clean_folder = folder.strip() or "INBOX"
|
|
resolved = _resolve_imap(
|
|
session,
|
|
profile=profile,
|
|
server_id=imap_server_id,
|
|
credential_id=imap_credential_id,
|
|
)
|
|
source = session.scalar(
|
|
select(MailBounceSource).where(
|
|
MailBounceSource.profile_id == profile.id,
|
|
MailBounceSource.folder == clean_folder,
|
|
)
|
|
)
|
|
if source is None:
|
|
source = MailBounceSource(
|
|
tenant_id=tenant_id,
|
|
profile_id=profile.id,
|
|
folder=clean_folder,
|
|
expected_imap_transport_revision=resolved.transport_revision,
|
|
created_by_user_id=created_by_user_id,
|
|
)
|
|
session.add(source)
|
|
source.imap_server_id = resolved.server.id if resolved.server else None
|
|
source.imap_credential_id = (
|
|
resolved.credential.id if resolved.credential else None
|
|
)
|
|
source.expected_imap_transport_revision = resolved.transport_revision
|
|
source.is_active = is_active
|
|
source.last_error = None
|
|
session.flush()
|
|
return source
|
|
|
|
|
|
def list_bounce_sources(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
) -> tuple[MailBounceSource, ...]:
|
|
return tuple(
|
|
session.scalars(
|
|
select(MailBounceSource)
|
|
.where(MailBounceSource.tenant_id == tenant_id)
|
|
.order_by(MailBounceSource.created_at, MailBounceSource.id)
|
|
)
|
|
)
|
|
|
|
|
|
def delete_bounce_source(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
source_id: str,
|
|
) -> None:
|
|
source = session.get(MailBounceSource, source_id)
|
|
if source is None or source.tenant_id != tenant_id:
|
|
raise MailBounceError("Bounce source not found.")
|
|
session.delete(source)
|
|
session.flush()
|
|
|
|
|
|
def list_bounce_observations(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
command_id: str | None = None,
|
|
limit: int = 100,
|
|
) -> tuple[MailBounceObservationRef, ...]:
|
|
statement = select(MailBounceObservation).where(
|
|
MailBounceObservation.tenant_id == tenant_id
|
|
)
|
|
if command_id:
|
|
statement = statement.where(MailBounceObservation.command_id == command_id)
|
|
rows = session.scalars(
|
|
statement.order_by(
|
|
MailBounceObservation.observed_at.desc(),
|
|
MailBounceObservation.id,
|
|
).limit(max(1, min(int(limit), 500)))
|
|
)
|
|
return tuple(_observation_ref(item) for item in rows)
|
|
|
|
|
|
class SqlMailBounceProcessingProvider(MailBounceProcessingProvider):
|
|
def scan_source(
|
|
self,
|
|
session: object,
|
|
*,
|
|
tenant_id: str,
|
|
source_id: str,
|
|
limit: int = 100,
|
|
) -> Mapping[str, object]:
|
|
db = _session(session)
|
|
source = db.get(MailBounceSource, source_id)
|
|
if source is None or source.tenant_id != tenant_id:
|
|
raise MailBounceError("Bounce source not found.")
|
|
processed, observations = self._scan_source(
|
|
db,
|
|
source,
|
|
limit=max(1, min(int(limit), 1_000)),
|
|
)
|
|
return {
|
|
"sources": 1,
|
|
"processed_messages": processed,
|
|
"observations": observations,
|
|
"failures": [],
|
|
}
|
|
|
|
def process_raw_message(
|
|
self,
|
|
session: object,
|
|
*,
|
|
tenant_id: str,
|
|
profile_id: str,
|
|
folder: str,
|
|
uid: str,
|
|
raw_message: bytes,
|
|
) -> tuple[MailBounceObservationRef, ...]:
|
|
db = _session(session)
|
|
raw_sha256 = hashlib.sha256(raw_message).hexdigest()
|
|
reports = parse_delivery_status(raw_message)
|
|
observations: list[MailBounceObservationRef] = []
|
|
for report in reports:
|
|
original_message_id = normalize_message_id(
|
|
report.get("original_message_id")
|
|
)
|
|
command = _correlated_command(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
original_message_id=original_message_id,
|
|
command_id=str(report.get("command_id") or "") or None,
|
|
)
|
|
fingerprint = _observation_fingerprint(
|
|
tenant_id=tenant_id,
|
|
profile_id=profile_id,
|
|
folder=folder,
|
|
uid=uid,
|
|
raw_sha256=raw_sha256,
|
|
recipient=report.get("recipient"),
|
|
action=report.get("action"),
|
|
status_code=report.get("status_code"),
|
|
)
|
|
existing = db.scalar(
|
|
select(MailBounceObservation).where(
|
|
MailBounceObservation.tenant_id == tenant_id,
|
|
MailBounceObservation.fingerprint == fingerprint,
|
|
)
|
|
)
|
|
if existing is not None:
|
|
observations.append(_observation_ref(existing))
|
|
continue
|
|
action = str(report.get("action") or "unknown").casefold()[:40]
|
|
status_code = _bounded(report.get("status_code"), 80)
|
|
item = MailBounceObservation(
|
|
tenant_id=tenant_id,
|
|
profile_id=profile_id,
|
|
folder=folder[:255],
|
|
uid=uid[:255],
|
|
fingerprint=fingerprint,
|
|
raw_sha256=raw_sha256,
|
|
original_message_id=original_message_id,
|
|
command_id=command.id if command else None,
|
|
recipient=_bounded(report.get("recipient"), 998),
|
|
action=action,
|
|
status_code=status_code,
|
|
diagnostic=_bounded(report.get("diagnostic"), 500),
|
|
permanent=action == "failed" or bool(status_code and status_code.startswith("5")),
|
|
observed_at=report.get("observed_at") or utcnow(),
|
|
matched=command is not None,
|
|
evidence={
|
|
"reporting_mta": report.get("reporting_mta"),
|
|
"remote_mta": report.get("remote_mta"),
|
|
"diagnostic_type": report.get("diagnostic_type"),
|
|
},
|
|
)
|
|
try:
|
|
with db.begin_nested():
|
|
db.add(item)
|
|
db.flush()
|
|
except IntegrityError:
|
|
item = db.scalar(
|
|
select(MailBounceObservation).where(
|
|
MailBounceObservation.tenant_id == tenant_id,
|
|
MailBounceObservation.fingerprint == fingerprint,
|
|
)
|
|
)
|
|
if item is None:
|
|
raise
|
|
audit_event(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
user_id=None,
|
|
action="mail.bounce.observed",
|
|
object_type="mail_bounce_observation",
|
|
object_id=item.id,
|
|
details={
|
|
"profile_id": profile_id,
|
|
"folder": folder,
|
|
"uid": uid,
|
|
"command_id": item.command_id,
|
|
"action": item.action,
|
|
"status_code": item.status_code,
|
|
"matched": item.matched,
|
|
"raw_sha256": raw_sha256,
|
|
},
|
|
)
|
|
observations.append(_observation_ref(item))
|
|
return tuple(observations)
|
|
|
|
def scan_due(
|
|
self,
|
|
session: object,
|
|
*,
|
|
tenant_id: str | None = None,
|
|
limit: int = 100,
|
|
) -> Mapping[str, object]:
|
|
db = _session(session)
|
|
remaining = max(1, min(int(limit), 1_000))
|
|
statement = select(MailBounceSource).where(
|
|
MailBounceSource.is_active.is_(True)
|
|
)
|
|
if tenant_id:
|
|
statement = statement.where(MailBounceSource.tenant_id == tenant_id)
|
|
sources = tuple(
|
|
db.scalars(statement.order_by(MailBounceSource.last_scanned_at, MailBounceSource.id))
|
|
)
|
|
processed = 0
|
|
observations = 0
|
|
failures: list[dict[str, str]] = []
|
|
for source in sources:
|
|
if remaining <= 0:
|
|
break
|
|
try:
|
|
count, found = self._scan_source(db, source, limit=remaining)
|
|
processed += count
|
|
observations += found
|
|
remaining -= count
|
|
except Exception as exc:
|
|
source.last_scanned_at = utcnow()
|
|
source.last_error = _bounded(exc, 500)
|
|
db.flush()
|
|
failures.append({"source_id": source.id, "error": source.last_error or "Scan failed"})
|
|
return {
|
|
"sources": len(sources),
|
|
"processed_messages": processed,
|
|
"observations": observations,
|
|
"failures": failures,
|
|
}
|
|
|
|
def observations_for_commands(
|
|
self,
|
|
session: object,
|
|
*,
|
|
tenant_id: str,
|
|
command_ids: tuple[str, ...],
|
|
) -> Mapping[str, tuple[MailBounceObservationRef, ...]]:
|
|
if not command_ids:
|
|
return {}
|
|
rows = _session(session).scalars(
|
|
select(MailBounceObservation)
|
|
.where(
|
|
MailBounceObservation.tenant_id == tenant_id,
|
|
MailBounceObservation.command_id.in_(tuple(set(command_ids))),
|
|
)
|
|
.order_by(MailBounceObservation.observed_at, MailBounceObservation.id)
|
|
)
|
|
grouped: defaultdict[str, list[MailBounceObservationRef]] = defaultdict(list)
|
|
for row in rows:
|
|
if row.command_id:
|
|
grouped[row.command_id].append(_observation_ref(row))
|
|
return {key: tuple(value) for key, value in grouped.items()}
|
|
|
|
def _scan_source(
|
|
self,
|
|
session: Session,
|
|
source: MailBounceSource,
|
|
*,
|
|
limit: int,
|
|
) -> tuple[int, int]:
|
|
profile = _profile(
|
|
session,
|
|
tenant_id=source.tenant_id,
|
|
profile_id=source.profile_id,
|
|
)
|
|
resolved = _resolve_imap(
|
|
session,
|
|
profile=profile,
|
|
server_id=source.imap_server_id,
|
|
credential_id=source.imap_credential_id,
|
|
)
|
|
if resolved.transport_revision != source.expected_imap_transport_revision:
|
|
raise MailBounceError(
|
|
"Bounce-source IMAP settings changed; review and save the source before scanning."
|
|
)
|
|
page = list_imap_uids_since(
|
|
imap_config=resolved.config,
|
|
folder=source.folder,
|
|
highest_uid=source.highest_processed_uid,
|
|
expected_uidvalidity=source.uidvalidity,
|
|
limit=limit,
|
|
)
|
|
found = 0
|
|
highest = 0 if page.cursor_reset else source.highest_processed_uid
|
|
for uid in page.uids:
|
|
raw = get_imap_raw_message(
|
|
imap_config=resolved.config,
|
|
folder=source.folder,
|
|
uid=uid,
|
|
)
|
|
found += len(
|
|
self.process_raw_message(
|
|
session,
|
|
tenant_id=source.tenant_id,
|
|
profile_id=source.profile_id,
|
|
folder=source.folder,
|
|
uid=uid,
|
|
raw_message=raw.raw,
|
|
)
|
|
)
|
|
highest = max(highest, int(uid))
|
|
now = utcnow()
|
|
source.uidvalidity = page.uidvalidity
|
|
source.highest_processed_uid = highest
|
|
source.last_scanned_at = now
|
|
source.last_success_at = now
|
|
source.last_error = None
|
|
session.flush()
|
|
return len(page.uids), found
|
|
|
|
|
|
def parse_delivery_status(raw_message: bytes) -> tuple[Mapping[str, object], ...]:
|
|
try:
|
|
message = BytesParser(policy=policy.default).parsebytes(raw_message)
|
|
except Exception as exc:
|
|
raise MailBounceError("Bounce message could not be parsed.") from exc
|
|
original_message_id = normalize_message_id(message.get("Original-Message-ID"))
|
|
command_id = _bounded(message.get("X-GovOPlaN-Delivery-ID"), 36)
|
|
reporting_mta = None
|
|
reports: list[dict[str, object]] = []
|
|
is_report = message.get_content_type() == "multipart/report"
|
|
for part in message.walk():
|
|
content_type = part.get_content_type()
|
|
if content_type == "message/delivery-status":
|
|
payload = part.get_payload()
|
|
blocks = payload if isinstance(payload, list) else []
|
|
for position, block in enumerate(blocks):
|
|
if not isinstance(block, EmailMessage):
|
|
continue
|
|
if position == 0:
|
|
reporting_mta = block.get("Reporting-MTA") or reporting_mta
|
|
original_message_id = normalize_message_id(
|
|
block.get("Original-Message-ID") or original_message_id
|
|
)
|
|
continue
|
|
reports.append(
|
|
_delivery_status_block(
|
|
block,
|
|
original_message_id=original_message_id,
|
|
command_id=command_id,
|
|
reporting_mta=reporting_mta,
|
|
fallback_date=message.get("Date"),
|
|
)
|
|
)
|
|
elif content_type == "message/rfc822":
|
|
payload = part.get_payload()
|
|
if isinstance(payload, list) and payload and isinstance(payload[0], EmailMessage):
|
|
original_message_id = normalize_message_id(
|
|
payload[0].get("Message-ID") or original_message_id
|
|
)
|
|
command_id = _bounded(
|
|
payload[0].get("X-GovOPlaN-Delivery-ID") or command_id,
|
|
36,
|
|
)
|
|
elif content_type == "text/rfc822-headers":
|
|
try:
|
|
headers = BytesParser(policy=policy.default).parsebytes(
|
|
part.get_payload(decode=True) or b"",
|
|
headersonly=True,
|
|
)
|
|
except Exception:
|
|
continue
|
|
original_message_id = normalize_message_id(
|
|
headers.get("Message-ID") or original_message_id
|
|
)
|
|
command_id = _bounded(
|
|
headers.get("X-GovOPlaN-Delivery-ID") or command_id,
|
|
36,
|
|
)
|
|
if reports:
|
|
for report in reports:
|
|
report["original_message_id"] = (
|
|
report.get("original_message_id") or original_message_id
|
|
)
|
|
report["command_id"] = report.get("command_id") or command_id
|
|
return tuple(reports)
|
|
failed = message.get_all("X-Failed-Recipients", [])
|
|
recipients = [value.strip() for header in failed for value in str(header).split(",") if value.strip()]
|
|
if not recipients and not is_report:
|
|
return ()
|
|
return tuple(
|
|
{
|
|
"original_message_id": original_message_id,
|
|
"command_id": command_id,
|
|
"recipient": recipient or None,
|
|
"action": "failed" if recipient else "unknown",
|
|
"status_code": None,
|
|
"diagnostic": "Unstructured delivery-status report",
|
|
"observed_at": _parsed_date(message.get("Date")) or utcnow(),
|
|
"reporting_mta": reporting_mta,
|
|
}
|
|
for recipient in (recipients or [""])
|
|
)
|
|
|
|
|
|
def _delivery_status_block(
|
|
block: EmailMessage,
|
|
*,
|
|
original_message_id: str | None,
|
|
command_id: str | None,
|
|
reporting_mta: object | None,
|
|
fallback_date: object | None,
|
|
) -> dict[str, object]:
|
|
diagnostic = str(block.get("Diagnostic-Code") or "")
|
|
diagnostic_type, _, diagnostic_text = diagnostic.partition(";")
|
|
return {
|
|
"original_message_id": normalize_message_id(
|
|
block.get("Original-Message-ID") or original_message_id
|
|
),
|
|
"command_id": command_id,
|
|
"recipient": _dsn_address(
|
|
block.get("Final-Recipient") or block.get("Original-Recipient")
|
|
),
|
|
"action": str(block.get("Action") or "unknown").casefold(),
|
|
"status_code": _bounded(block.get("Status"), 80),
|
|
"diagnostic": diagnostic_text.strip() or diagnostic_type.strip() or None,
|
|
"diagnostic_type": diagnostic_type.strip() or None,
|
|
"reporting_mta": str(reporting_mta or "") or None,
|
|
"remote_mta": _dsn_address(block.get("Remote-MTA")),
|
|
"observed_at": _parsed_date(
|
|
block.get("Last-Attempt-Date") or fallback_date
|
|
) or utcnow(),
|
|
}
|
|
|
|
|
|
def _correlated_command(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
original_message_id: str | None,
|
|
command_id: str | None,
|
|
) -> MailDeliveryCommand | None:
|
|
if command_id:
|
|
command = session.get(MailDeliveryCommand, command_id)
|
|
if command is not None and command.tenant_id == tenant_id:
|
|
return command
|
|
if not original_message_id:
|
|
return None
|
|
return session.scalar(
|
|
select(MailDeliveryCommand).where(
|
|
MailDeliveryCommand.tenant_id == tenant_id,
|
|
MailDeliveryCommand.rfc_message_id == original_message_id,
|
|
)
|
|
)
|
|
|
|
|
|
def _observation_fingerprint(**values: object) -> str:
|
|
canonical = "\x1f".join(str(values[key] or "") for key in sorted(values))
|
|
return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def _observation_ref(item: MailBounceObservation) -> MailBounceObservationRef:
|
|
return MailBounceObservationRef(
|
|
id=item.id,
|
|
tenant_id=item.tenant_id,
|
|
profile_id=item.profile_id,
|
|
folder=item.folder,
|
|
uid=item.uid,
|
|
original_message_id=item.original_message_id,
|
|
command_id=item.command_id,
|
|
recipient=item.recipient,
|
|
action=item.action,
|
|
status_code=item.status_code,
|
|
diagnostic=item.diagnostic,
|
|
permanent=item.permanent,
|
|
observed_at=item.observed_at,
|
|
matched=item.matched,
|
|
evidence=dict(item.evidence or {}),
|
|
)
|
|
|
|
|
|
def _profile(session: Session, *, tenant_id: str, profile_id: str) -> MailServerProfile:
|
|
profile = session.get(MailServerProfile, profile_id)
|
|
if profile is None or profile.tenant_id not in {None, tenant_id} or not profile.is_active:
|
|
raise MailBounceError("Active Mail profile not found.")
|
|
return profile
|
|
|
|
|
|
def _resolve_imap(
|
|
session: Session,
|
|
*,
|
|
profile: MailServerProfile,
|
|
server_id: str | None,
|
|
credential_id: str | None,
|
|
):
|
|
try:
|
|
resolved = resolve_mail_transport(
|
|
session,
|
|
profile=profile,
|
|
protocol="imap",
|
|
context=hierarchy_context_for_profile(profile, administrative=True),
|
|
server_id=server_id,
|
|
credential_id=credential_id,
|
|
)
|
|
except MailServerHierarchyError as exc:
|
|
raise MailBounceError(str(exc)) from exc
|
|
return resolved
|
|
|
|
|
|
def _session(value: object) -> Session:
|
|
if not isinstance(value, Session):
|
|
raise MailBounceError("Bounce processing requires a SQLAlchemy Session.")
|
|
return value
|
|
|
|
|
|
def _bounded(value: object | None, limit: int) -> str | None:
|
|
text = " ".join(str(value or "").split())
|
|
return text[:limit] or None
|
|
|
|
|
|
def _dsn_address(value: object | None) -> str | None:
|
|
text = str(value or "")
|
|
_, separator, address = text.partition(";")
|
|
return _bounded(address if separator else text, 998)
|
|
|
|
|
|
def _parsed_date(value: object | None) -> datetime | None:
|
|
if not value:
|
|
return None
|
|
try:
|
|
parsed = parsedate_to_datetime(str(value))
|
|
except (TypeError, ValueError, OverflowError):
|
|
return None
|
|
if parsed.tzinfo is None:
|
|
return parsed.replace(tzinfo=timezone.utc)
|
|
return parsed.astimezone(timezone.utc)
|
|
|
|
|
|
__all__ = [
|
|
"MailBounceError",
|
|
"SqlMailBounceProcessingProvider",
|
|
"configure_bounce_source",
|
|
"delete_bounce_source",
|
|
"list_bounce_sources",
|
|
"list_bounce_observations",
|
|
"normalize_message_id",
|
|
"parse_delivery_status",
|
|
]
|