diff --git a/src/govoplan_mail/backend/bounce_processing.py b/src/govoplan_mail/backend/bounce_processing.py new file mode 100644 index 0000000..822ff49 --- /dev/null +++ b/src/govoplan_mail/backend/bounce_processing.py @@ -0,0 +1,617 @@ +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", +] diff --git a/src/govoplan_mail/backend/db/models.py b/src/govoplan_mail/backend/db/models.py index 3b66faa..99f9d41 100644 --- a/src/govoplan_mail/backend/db/models.py +++ b/src/govoplan_mail/backend/db/models.py @@ -200,6 +200,9 @@ class MailDeliveryCommand(Base, TimestampMixin): from_header_encrypted: Mapped[str | None] = mapped_column(Text, nullable=True) message_encrypted: Mapped[str | None] = mapped_column(Text, nullable=True) message_sha256: Mapped[str] = mapped_column(String(64), nullable=False) + rfc_message_id: Mapped[str | None] = mapped_column( + String(998), nullable=True, index=True + ) message_size_bytes: Mapped[int] = mapped_column(BigInteger, nullable=False) recipient_count: Mapped[int] = mapped_column(Integer, nullable=False) status: Mapped[str] = mapped_column(String(40), nullable=False, default="pending", index=True) @@ -286,3 +289,101 @@ class MailDeliveryReconciliation(Base, TimestampMixin): nullable=True, index=True, ) + + +class MailBounceSource(Base, TimestampMixin): + __tablename__ = "mail_bounce_sources" + __table_args__ = ( + UniqueConstraint( + "profile_id", + "folder", + name="uq_mail_bounce_sources_profile_folder", + ), + Index( + "ix_mail_bounce_sources_scan", + "is_active", + "last_scanned_at", + ), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + profile_id: Mapped[str] = mapped_column( + ForeignKey("mail_server_profiles.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + folder: Mapped[str] = mapped_column( + String(255), default="INBOX", nullable=False + ) + imap_server_id: Mapped[str | None] = mapped_column(String(36), nullable=True) + imap_credential_id: Mapped[str | None] = mapped_column(String(36), nullable=True) + expected_imap_transport_revision: Mapped[str] = mapped_column( + String(120), nullable=False + ) + is_active: Mapped[bool] = mapped_column( + Boolean, default=True, nullable=False, index=True + ) + uidvalidity: Mapped[str | None] = mapped_column(String(255), nullable=True) + highest_processed_uid: Mapped[int] = mapped_column( + BigInteger, default=0, nullable=False + ) + last_scanned_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True, index=True + ) + last_success_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + last_error: Mapped[str | None] = mapped_column(String(500), nullable=True) + created_by_user_id: Mapped[str | None] = mapped_column( + ForeignKey("access_users.id", ondelete="SET NULL"), + nullable=True, + index=True, + ) + + +class MailBounceObservation(Base, TimestampMixin): + __tablename__ = "mail_bounce_observations" + __table_args__ = ( + UniqueConstraint( + "tenant_id", + "fingerprint", + name="uq_mail_bounce_observations_fingerprint", + ), + Index( + "ix_mail_bounce_observations_command", + "tenant_id", + "command_id", + "observed_at", + ), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + profile_id: Mapped[str] = mapped_column( + ForeignKey("mail_server_profiles.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + folder: Mapped[str] = mapped_column(String(255), nullable=False) + uid: Mapped[str] = mapped_column(String(255), nullable=False) + fingerprint: Mapped[str] = mapped_column(String(64), nullable=False) + raw_sha256: Mapped[str] = mapped_column(String(64), nullable=False) + original_message_id: Mapped[str | None] = mapped_column( + String(998), nullable=True, index=True + ) + command_id: Mapped[str | None] = mapped_column( + ForeignKey("mail_delivery_commands.id", ondelete="SET NULL"), + nullable=True, + index=True, + ) + recipient: Mapped[str | None] = mapped_column(String(998), nullable=True) + action: Mapped[str] = mapped_column(String(40), nullable=False, index=True) + status_code: Mapped[str | None] = mapped_column(String(80), nullable=True) + diagnostic: Mapped[str | None] = mapped_column(String(500), nullable=True) + permanent: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + observed_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, index=True + ) + matched: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + evidence: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict, nullable=False) diff --git a/src/govoplan_mail/backend/delivery_outbox.py b/src/govoplan_mail/backend/delivery_outbox.py index 8fbe6a0..6df4eef 100644 --- a/src/govoplan_mail/backend/delivery_outbox.py +++ b/src/govoplan_mail/backend/delivery_outbox.py @@ -5,6 +5,8 @@ import hashlib import json from collections import Counter from datetime import datetime, timedelta, timezone +from email import policy +from email.parser import BytesParser from typing import Any from sqlalchemy import or_, select @@ -105,6 +107,18 @@ def _message_bytes(command: MailDeliveryCommand) -> bytes: return message +def _rfc_message_id(message: bytes) -> str | None: + try: + value = BytesParser(policy=policy.default).parsebytes( + message, + headersonly=True, + ).get("Message-ID") + except Exception: + return None + normalized = " ".join(str(value or "").split()) + return normalized[:998] or None + + def _delivery_payload( *, command_type: str, @@ -169,6 +183,7 @@ def submit_delivery_command( if retention_days < 1: raise MailDeliveryError("Mail payload retention must be at least one day") digest = hashlib.sha256(message_bytes).hexdigest() + rfc_message_id = _rfc_message_id(message_bytes) request_hash = _canonical_hash( _delivery_payload( command_type=command_type, @@ -219,6 +234,7 @@ def submit_delivery_command( from_header_encrypted=encrypt_secret(from_header), message_encrypted=encrypt_secret(base64.b64encode(message_bytes).decode("ascii")), message_sha256=digest, + rfc_message_id=rfc_message_id, message_size_bytes=len(message_bytes), recipient_count=len(clean_recipients), status="pending", @@ -299,6 +315,7 @@ def delivery_command_summary( "failure_code": command.failure_code, "failure_summary": command.failure_summary, "message_sha256": command.message_sha256, + "rfc_message_id": command.rfc_message_id, "message_size_bytes": command.message_size_bytes, "created_at": command.created_at, "completed_at": command.completed_at, diff --git a/src/govoplan_mail/backend/manifest.py b/src/govoplan_mail/backend/manifest.py index 04b30cc..8433276 100644 --- a/src/govoplan_mail/backend/manifest.py +++ b/src/govoplan_mail/backend/manifest.py @@ -7,6 +7,7 @@ from sqlalchemy import inspect from govoplan_core.core.access import CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER from govoplan_core.core.mail import ( + CAPABILITY_MAIL_BOUNCE_PROCESSING, CAPABILITY_MAIL_DELIVERY_OUTBOX, CAPABILITY_MAIL_NOTIFICATION_DELIVERY, ) @@ -46,6 +47,8 @@ _mail_table_retirement_provider = drop_table_retirement_provider( mail_models.MailDeliveryReconciliation, mail_models.MailDeliveryAttempt, mail_models.MailDeliveryCommand, + mail_models.MailBounceObservation, + mail_models.MailBounceSource, label="Mail", ) @@ -111,6 +114,16 @@ PERMISSIONS = ( "Reconcile mail delivery outcomes", "Reconcile unknown delivery outcomes and explicitly authorize deliberate resend commands.", ), + _permission( + "mail:bounce:read", + "View bounce processing", + "Inspect bounded delivery-status observations and their correlation state.", + ), + _permission( + "mail:bounce:manage", + "Manage bounce processing", + "Configure and run IMAP delivery-status watchers.", + ), _permission( "mail:secret:manage_own", "Manage own mail secrets", @@ -132,6 +145,8 @@ ROLE_TEMPLATES = ( "mail:secret:manage", "mail:delivery:diagnostic", "mail:delivery:reconcile", + "mail:bounce:read", + "mail:bounce:manage", ), ), RoleTemplate( @@ -184,6 +199,7 @@ manifest = ModuleManifest( ModuleInterfaceProvider(name="mail.delivery_commands", version="0.1.0"), ModuleInterfaceProvider(name="mail.delivery_outbox", version="0.1.0"), ModuleInterfaceProvider(name="mail.notification_delivery", version="0.1.0"), + ModuleInterfaceProvider(name="mail.bounce_processing", version="0.1.0"), ), requires_interfaces=( ModuleInterfaceRequirement( @@ -208,7 +224,7 @@ manifest = ModuleManifest( permissions=PERMISSIONS, route_factory=_mail_router, role_templates=ROLE_TEMPLATES, - nav_items=(NavItem(path="/mail", label="Mail", icon="mail", required_any=("mail:mailbox:read",), order=50),), + nav_items=(NavItem(path="/mail", label="Mail", icon="mail", required_any=("mail:mailbox:read", "mail:bounce:read", "mail:bounce:manage"), order=50),), frontend=FrontendModule( module_id="mail", package_name="@govoplan/mail-webui", @@ -219,8 +235,15 @@ manifest = ModuleManifest( required_any=("mail:mailbox:read",), order=50, ), + FrontendRoute( + path="/mail/bounces", + component="MailBouncePage", + required_any=("mail:bounce:read", "mail:bounce:manage"), + order=51, + surface_id="mail.bounce-processing", + ), ), - nav_items=(NavItem(path="/mail", label="Mail", icon="mail", required_any=("mail:mailbox:read",), order=50),), + nav_items=(NavItem(path="/mail", label="Mail", icon="mail", required_any=("mail:mailbox:read", "mail:bounce:read", "mail:bounce:manage"), order=50),), view_surfaces=( ViewSurface(id="mail.admin.system-servers", module_id="mail", kind="section", label="System mail servers", order=70), ViewSurface(id="mail.admin.tenant-servers", module_id="mail", kind="section", label="Tenant mail servers", order=60), @@ -248,6 +271,8 @@ manifest = ModuleManifest( mail_models.MailDeliveryReconciliation, mail_models.MailDeliveryAttempt, mail_models.MailDeliveryCommand, + mail_models.MailBounceObservation, + mail_models.MailBounceSource, label="Mail", ), ), @@ -261,6 +286,10 @@ manifest = ModuleManifest( "govoplan_mail.backend.capabilities", fromlist=["notification_delivery_capability"], ).notification_delivery_capability(context), + CAPABILITY_MAIL_BOUNCE_PROCESSING: lambda context: __import__( + "govoplan_mail.backend.bounce_processing", + fromlist=["SqlMailBounceProcessingProvider"], + ).SqlMailBounceProcessingProvider(), }, documentation=( DocumentationTopic( @@ -300,6 +329,37 @@ manifest = ModuleManifest( ], }, ), + DocumentationTopic( + id="mail.bounce-processing", + title="Delivery-status and bounce processing", + summary="Watch an authorized IMAP folder and correlate DSN outcomes with durable Mail commands.", + body=( + "Mail stores the outgoing RFC Message-ID with its durable command, parses " + "bounded message/delivery-status reports without changing mailbox flags, " + "and records idempotent per-recipient observations. SMTP acceptance remains " + "separate from a later bounce. Unmatched reports remain visible for review; " + "Mail stores only bounded diagnostics and a raw digest, not the raw bounce body." + ), + layer="configured", + documentation_types=("admin", "user"), + audience=("mail_admin", "campaign_manager", "release_reviewer"), + order=43, + conditions=( + DocumentationCondition( + required_modules=("mail",), + any_scopes=("mail:bounce:read", "mail:bounce:manage"), + ), + ), + related_modules=("campaigns", "audit"), + metadata={ + "kind": "workflow", + "route": "/mail/bounces", + "verification": ( + "Send a message with a unique Message-ID, ingest a DSN twice, and " + "verify one correlated observation while the SMTP acceptance remains intact." + ), + }, + ), DocumentationTopic( id="mail.profile-ownership-and-consumers", title="Mail owns transport profiles and credentials", diff --git a/src/govoplan_mail/backend/migrations/versions/93b4c5d6e7f8_mail_bounce_processing.py b/src/govoplan_mail/backend/migrations/versions/93b4c5d6e7f8_mail_bounce_processing.py new file mode 100644 index 0000000..b51bc75 --- /dev/null +++ b/src/govoplan_mail/backend/migrations/versions/93b4c5d6e7f8_mail_bounce_processing.py @@ -0,0 +1,133 @@ +"""add mail bounce processing + +Revision ID: 93b4c5d6e7f8 +Revises: 82a3b4c5d6e7 +Create Date: 2026-07-31 19:00:00.000000 +""" +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "93b4c5d6e7f8" +down_revision = "82a3b4c5d6e7" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + with op.batch_alter_table("mail_delivery_commands") as batch_op: + batch_op.add_column( + sa.Column("rfc_message_id", sa.String(length=998), nullable=True) + ) + batch_op.create_index( + "ix_mail_delivery_commands_rfc_message_id", + ("rfc_message_id",), + ) + + op.create_table( + "mail_bounce_sources", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("tenant_id", sa.String(length=36), nullable=False), + sa.Column("profile_id", sa.String(length=36), nullable=False), + sa.Column("folder", sa.String(length=255), nullable=False), + sa.Column("imap_server_id", sa.String(length=36), nullable=True), + sa.Column("imap_credential_id", sa.String(length=36), nullable=True), + sa.Column( + "expected_imap_transport_revision", + sa.String(length=120), + nullable=False, + ), + sa.Column("is_active", sa.Boolean(), nullable=False), + sa.Column("uidvalidity", sa.String(length=255), nullable=True), + sa.Column("highest_processed_uid", sa.BigInteger(), nullable=False), + sa.Column("last_scanned_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("last_success_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("last_error", sa.String(length=500), nullable=True), + sa.Column("created_by_user_id", sa.String(length=36), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint( + ["profile_id"], ["mail_server_profiles.id"], ondelete="CASCADE" + ), + sa.ForeignKeyConstraint( + ["created_by_user_id"], ["access_users.id"], ondelete="SET NULL" + ), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint( + "profile_id", "folder", name="uq_mail_bounce_sources_profile_folder" + ), + ) + op.create_index("ix_mail_bounce_sources_tenant_id", "mail_bounce_sources", ["tenant_id"]) + op.create_index("ix_mail_bounce_sources_profile_id", "mail_bounce_sources", ["profile_id"]) + op.create_index("ix_mail_bounce_sources_is_active", "mail_bounce_sources", ["is_active"]) + op.create_index("ix_mail_bounce_sources_last_scanned_at", "mail_bounce_sources", ["last_scanned_at"]) + op.create_index("ix_mail_bounce_sources_created_by_user_id", "mail_bounce_sources", ["created_by_user_id"]) + op.create_index( + "ix_mail_bounce_sources_scan", + "mail_bounce_sources", + ["is_active", "last_scanned_at"], + ) + + op.create_table( + "mail_bounce_observations", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("tenant_id", sa.String(length=36), nullable=False), + sa.Column("profile_id", sa.String(length=36), nullable=False), + sa.Column("folder", sa.String(length=255), nullable=False), + sa.Column("uid", sa.String(length=255), nullable=False), + sa.Column("fingerprint", sa.String(length=64), nullable=False), + sa.Column("raw_sha256", sa.String(length=64), nullable=False), + sa.Column("original_message_id", sa.String(length=998), nullable=True), + sa.Column("command_id", sa.String(length=36), nullable=True), + sa.Column("recipient", sa.String(length=998), nullable=True), + sa.Column("action", sa.String(length=40), nullable=False), + sa.Column("status_code", sa.String(length=80), nullable=True), + sa.Column("diagnostic", sa.String(length=500), nullable=True), + sa.Column("permanent", sa.Boolean(), nullable=False), + sa.Column("observed_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("matched", sa.Boolean(), nullable=False), + sa.Column("evidence", sa.JSON(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint( + ["profile_id"], ["mail_server_profiles.id"], ondelete="CASCADE" + ), + sa.ForeignKeyConstraint( + ["command_id"], ["mail_delivery_commands.id"], ondelete="SET NULL" + ), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint( + "tenant_id", + "fingerprint", + name="uq_mail_bounce_observations_fingerprint", + ), + ) + for column in ( + "tenant_id", + "profile_id", + "original_message_id", + "command_id", + "action", + "observed_at", + "matched", + ): + op.create_index( + f"ix_mail_bounce_observations_{column}", + "mail_bounce_observations", + [column], + ) + op.create_index( + "ix_mail_bounce_observations_command", + "mail_bounce_observations", + ["tenant_id", "command_id", "observed_at"], + ) + + +def downgrade() -> None: + op.drop_table("mail_bounce_observations") + op.drop_table("mail_bounce_sources") + with op.batch_alter_table("mail_delivery_commands") as batch_op: + batch_op.drop_index("ix_mail_delivery_commands_rfc_message_id") + batch_op.drop_column("rfc_message_id") diff --git a/src/govoplan_mail/backend/router.py b/src/govoplan_mail/backend/router.py index f332e45..bd37b91 100644 --- a/src/govoplan_mail/backend/router.py +++ b/src/govoplan_mail/backend/router.py @@ -12,6 +12,12 @@ from govoplan_mail.backend.schemas import ( MailAddressLookupCandidate, MailAddressLookupResponse, MailConnectionTestResponse, + MailBounceObservationListResponse, + MailBounceObservationResponse, + MailBounceScanResponse, + MailBounceSourceListResponse, + MailBounceSourceRequest, + MailBounceSourceResponse, MailCampaignCredentialCreateRequest, MailCredentialBindRequest, MailCredentialCreateRequest, @@ -43,6 +49,7 @@ from govoplan_mail.backend.schemas import ( MailSmtpTestRequest, ) from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope +from govoplan_core.audit.logging import audit_event from govoplan_core.api.v1.schemas import DeltaDeletedItem from govoplan_core.core.change_sequence import ( ChangeSequenceEntry, @@ -90,6 +97,14 @@ from govoplan_mail.backend.delivery_outbox import ( reconcile_delivery_command, resend_delivery_command, ) +from govoplan_mail.backend.bounce_processing import ( + MailBounceError, + SqlMailBounceProcessingProvider, + configure_bounce_source, + delete_bounce_source, + list_bounce_observations, + list_bounce_sources, +) from govoplan_mail.backend.server_hierarchy import ( MailHierarchyContext, MailServerHierarchyError, @@ -129,6 +144,7 @@ MAIL_CREDENTIAL_RESOURCE = "mail_credential" MAILBOX_MESSAGES_CURSOR_SCOPE = "mail.mailbox.messages.v1" DEFAULT_MAILBOX_MESSAGE_LIMIT = 50 CAPABILITY_ADDRESSES_LOOKUP = "addresses.lookup" +bounce_provider = SqlMailBounceProcessingProvider() @router.get( @@ -239,6 +255,143 @@ def resend_mail_delivery_command( return MailDeliveryCommandResponse(result=result) +@router.get("/bounce-sources", response_model=MailBounceSourceListResponse) +def get_mail_bounce_sources( + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> MailBounceSourceListResponse: + _require_scope(principal, "mail:bounce:read") + return MailBounceSourceListResponse( + sources=[ + MailBounceSourceResponse.model_validate(item, from_attributes=True) + for item in list_bounce_sources( + session, + tenant_id=principal.tenant_id, + ) + ] + ) + + +@router.post("/bounce-sources", response_model=MailBounceSourceResponse) +def save_mail_bounce_source( + payload: MailBounceSourceRequest, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> MailBounceSourceResponse: + _require_scope(principal, "mail:bounce:manage") + try: + source = configure_bounce_source( + session, + tenant_id=principal.tenant_id, + profile_id=payload.profile_id, + folder=payload.folder, + imap_server_id=payload.imap_server_id, + imap_credential_id=payload.imap_credential_id, + is_active=payload.is_active, + created_by_user_id=principal.user.id, + ) + except MailBounceError as exc: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, + detail=str(exc), + ) from exc + audit_event( + session, + tenant_id=principal.tenant_id, + user_id=principal.user.id, + action="mail.bounce_source.saved", + object_type="mail_bounce_source", + object_id=source.id, + details={ + "profile_id": source.profile_id, + "folder": source.folder, + "is_active": source.is_active, + }, + ) + session.commit() + return MailBounceSourceResponse.model_validate(source, from_attributes=True) + + +@router.delete("/bounce-sources/{source_id}", status_code=status.HTTP_204_NO_CONTENT) +def remove_mail_bounce_source( + source_id: str, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> None: + _require_scope(principal, "mail:bounce:manage") + try: + delete_bounce_source( + session, + tenant_id=principal.tenant_id, + source_id=source_id, + ) + except MailBounceError as exc: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc + audit_event( + session, + tenant_id=principal.tenant_id, + user_id=principal.user.id, + action="mail.bounce_source.deleted", + object_type="mail_bounce_source", + object_id=source_id, + details={}, + ) + session.commit() + + +@router.post( + "/bounce-sources/{source_id}/scan", + response_model=MailBounceScanResponse, +) +def scan_mail_bounce_source( + source_id: str, + limit: int = Query(default=100, ge=1, le=1_000), + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> MailBounceScanResponse: + _require_scope(principal, "mail:bounce:manage") + try: + result = bounce_provider.scan_source( + session, + tenant_id=principal.tenant_id, + source_id=source_id, + limit=limit, + ) + except MailBounceError as exc: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, + detail=str(exc), + ) from exc + session.commit() + return MailBounceScanResponse.model_validate(result) + + +@router.get( + "/bounce-observations", + response_model=MailBounceObservationListResponse, +) +def get_mail_bounce_observations( + command_id: str | None = Query(default=None, max_length=36), + limit: int = Query(default=100, ge=1, le=500), + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> MailBounceObservationListResponse: + _require_scope(principal, "mail:bounce:read") + return MailBounceObservationListResponse( + observations=[ + MailBounceObservationResponse.model_validate( + dataclasses.asdict(item) + ) + for item in list_bounce_observations( + session, + tenant_id=principal.tenant_id, + command_id=command_id, + limit=limit, + ) + ] + ) + + def _capability_payload(value: object) -> dict[str, Any]: if dataclasses.is_dataclass(value): return dataclasses.asdict(value) diff --git a/src/govoplan_mail/backend/schemas.py b/src/govoplan_mail/backend/schemas.py index b85d47c..74c7edf 100644 --- a/src/govoplan_mail/backend/schemas.py +++ b/src/govoplan_mail/backend/schemas.py @@ -454,3 +454,62 @@ class MailDeliveryResendRequest(BaseModel): model_config = ConfigDict(extra="forbid") idempotency_key: str = Field(min_length=1, max_length=200) + + +class MailBounceSourceRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + + profile_id: str = Field(min_length=1, max_length=36) + folder: str = Field(default="INBOX", min_length=1, max_length=255) + imap_server_id: str | None = Field(default=None, max_length=36) + imap_credential_id: str | None = Field(default=None, max_length=36) + is_active: bool = True + + +class MailBounceSourceResponse(BaseModel): + id: str + tenant_id: str + profile_id: str + folder: str + imap_server_id: str | None = None + imap_credential_id: str | None = None + expected_imap_transport_revision: str + is_active: bool + uidvalidity: str | None = None + highest_processed_uid: int + last_scanned_at: datetime | None = None + last_success_at: datetime | None = None + last_error: str | None = None + + +class MailBounceSourceListResponse(BaseModel): + sources: list[MailBounceSourceResponse] + + +class MailBounceObservationResponse(BaseModel): + id: str + tenant_id: str + profile_id: str + folder: str + uid: str + original_message_id: str | None = None + command_id: str | None = None + recipient: str | None = None + action: str + status_code: str | None = None + diagnostic: str | None = None + permanent: bool + observed_at: datetime + matched: bool + evidence: dict[str, Any] = Field(default_factory=dict) + + +class MailBounceObservationListResponse(BaseModel): + observations: list[MailBounceObservationResponse] + + +class MailBounceScanResponse(BaseModel): + sources: int + processed_messages: int + observations: int + failures: list[dict[str, str]] = Field(default_factory=list) diff --git a/src/govoplan_mail/backend/sending/imap.py b/src/govoplan_mail/backend/sending/imap.py index 9014eda..e626748 100644 --- a/src/govoplan_mail/backend/sending/imap.py +++ b/src/govoplan_mail/backend/sending/imap.py @@ -180,6 +180,29 @@ class ImapMailboxMessageResult: message: ImapMailboxMessageDetail +@dataclass(frozen=True, slots=True) +class ImapRawMessageResult: + host: str + port: int + security: str + folder: str + uid: str + flags: list[str] + size_bytes: int + raw: bytes + + +@dataclass(frozen=True, slots=True) +class ImapUidListResult: + host: str + port: int + security: str + folder: str + uids: list[str] + uidvalidity: str | None + cursor_reset: bool = False + + @dataclass(frozen=True, slots=True) class ImapAppendResult: host: str @@ -1079,6 +1102,126 @@ def get_imap_message(*, imap_config: ImapConfig, folder: str, uid: str) -> ImapM _log_imap_cleanup_failure("reading message", cleanup_exc) +def get_imap_raw_message( + *, + imap_config: ImapConfig, + folder: str, + uid: str, +) -> ImapRawMessageResult: + """Fetch bounded raw MIME without changing mailbox flags.""" + + host, port = _require_imap_config(imap_config) + folder = (folder or "INBOX").strip() or "INBOX" + uid = str(uid).strip() + if not uid: + raise ImapConfigurationError("Message UID is required") + if is_mock_imap_host(imap_config.host): + record = get_record(uid, include_raw=True) + if not record or not _mock_folder_matches(record, folder): + raise ImapAppendError( + f"Mock mailbox message {uid!r} was not found in folder {folder!r}", + temporary=False, + ) + raw = _mock_raw_bytes(record) + return ImapRawMessageResult( + host=host, + port=port, + security=imap_config.security.value, + folder=folder, + uid=uid, + flags=["\\Seen"] if record.get("kind") == "imap_append" else [], + size_bytes=len(raw), + raw=raw, + ) + + client = _open_imap(imap_config) + try: + _select_readonly(client, folder) + fetched_uid, flags, size_bytes, raw = _fetch_message_by_uid(client, uid) + return ImapRawMessageResult( + host=host, + port=port, + security=imap_config.security.value, + folder=folder, + uid=fetched_uid, + flags=flags, + size_bytes=size_bytes if size_bytes is not None else len(raw), + raw=raw, + ) + finally: + try: + client.logout() + except Exception as cleanup_exc: + _log_imap_cleanup_failure("reading raw message", cleanup_exc) + + +def list_imap_uids_since( + *, + imap_config: ImapConfig, + folder: str, + highest_uid: int = 0, + expected_uidvalidity: str | None = None, + limit: int = 100, +) -> ImapUidListResult: + """Return new UIDs oldest-first so a bounded watcher cannot skip a burst.""" + + host, port = _require_imap_config(imap_config) + folder = (folder or "INBOX").strip() or "INBOX" + bounded_limit = max(1, min(int(limit), 1_000)) + if is_mock_imap_host(imap_config.host): + records = [ + record + for record in list_records(limit=5_000) + if _mock_folder_matches(record, folder) + ] + cursor_reset = bool( + expected_uidvalidity and expected_uidvalidity != "mock-v1" + ) + effective_highest = 0 if cursor_reset else highest_uid + numeric = sorted( + int(value) + for record in records + for value in (str(record.get("id") or ""),) + if value.isdigit() and int(value) > effective_highest + ) + return ImapUidListResult( + host=host, + port=port, + security=imap_config.security.value, + folder=folder, + uids=[str(value) for value in numeric[:bounded_limit]], + uidvalidity="mock-v1", + cursor_reset=cursor_reset, + ) + + client = _open_imap(imap_config) + try: + _total, uidvalidity = _select_readonly(client, folder) + cursor_reset = bool( + expected_uidvalidity and uidvalidity != expected_uidvalidity + ) + effective_highest = 0 if cursor_reset else highest_uid + numeric = sorted( + int(uid) + for uid in _search_all_uids(client) + if uid.isdigit() and int(uid) > effective_highest + ) + return ImapUidListResult( + host=host, + port=port, + security=imap_config.security.value, + folder=folder, + uids=[str(value) for value in numeric[:bounded_limit]], + uidvalidity=uidvalidity, + cursor_reset=cursor_reset, + ) + finally: + try: + client.logout() + except Exception as cleanup_exc: + _log_imap_cleanup_failure("listing watcher UIDs", cleanup_exc) + + def append_message_to_sent( message_bytes: bytes, *, diff --git a/tests/test_bounce_processing.py b/tests/test_bounce_processing.py new file mode 100644 index 0000000..906b49a --- /dev/null +++ b/tests/test_bounce_processing.py @@ -0,0 +1,186 @@ +from __future__ import annotations + +import unittest +from unittest.mock import patch + +from sqlalchemy import Column, String, Table, create_engine +from sqlalchemy.orm import Session, sessionmaker +from sqlalchemy.pool import StaticPool + +from govoplan_core.db.base import Base +from govoplan_mail.backend.bounce_processing import ( + SqlMailBounceProcessingProvider, + parse_delivery_status, +) +from govoplan_mail.backend.db.models import ( + MailBounceObservation, + MailBounceSource, + MailDeliveryCommand, + MailServerProfile, +) +from govoplan_mail.backend.delivery_outbox import submit_delivery_command + + +DSN = b"""From: Mail Delivery Subsystem +To: sender@example.test +Date: Fri, 31 Jul 2026 12:00:00 +0000 +Subject: Delivery Status Notification (Failure) +MIME-Version: 1.0 +Content-Type: multipart/report; report-type=delivery-status; boundary="dsn" + +--dsn +Content-Type: text/plain; charset=utf-8 + +Delivery failed. +--dsn +Content-Type: message/delivery-status + +Reporting-MTA: dns; mx.example.test +Original-Message-ID: + +Final-Recipient: rfc822; recipient@example.test +Action: failed +Status: 5.1.1 +Remote-MTA: dns; destination.example.test +Diagnostic-Code: smtp; 550 mailbox unavailable +Last-Attempt-Date: Fri, 31 Jul 2026 11:59:00 +0000 + +--dsn +Content-Type: message/rfc822 + +Message-ID: +From: sender@example.test +To: recipient@example.test +Subject: Original + +Body +--dsn-- +""" + + +class MailBounceProcessingTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine( + "sqlite+pysqlite://", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + access_users = Base.metadata.tables.get("access_users") + if access_users is None: + access_users = Table( + "access_users", + Base.metadata, + Column("id", String(36), primary_key=True), + ) + Base.metadata.create_all( + self.engine, + tables=[ + access_users, + MailServerProfile.__table__, + MailDeliveryCommand.__table__, + MailBounceSource.__table__, + MailBounceObservation.__table__, + ], + ) + self.SessionLocal = sessionmaker( + bind=self.engine, + class_=Session, + expire_on_commit=False, + ) + with self.SessionLocal() as session: + session.add( + MailServerProfile( + id="profile-1", + tenant_id="tenant-1", + scope_type="tenant", + scope_id="tenant-1", + name="Delivery", + slug="delivery", + smtp_config={"host": "smtp.example.test", "port": 25}, + ) + ) + session.commit() + self.audit_delivery = patch( + "govoplan_mail.backend.delivery_outbox.audit_event" + ) + self.audit_bounce = patch( + "govoplan_mail.backend.bounce_processing.audit_event" + ) + self.audit_delivery.start() + self.audit_bounce.start() + self.addCleanup(self.audit_delivery.stop) + self.addCleanup(self.audit_bounce.stop) + self.addCleanup(self.engine.dispose) + + def test_parser_extracts_structured_recipient_outcome(self) -> None: + reports = parse_delivery_status(DSN) + + self.assertEqual(1, len(reports)) + self.assertEqual("recipient@example.test", reports[0]["recipient"]) + self.assertEqual("failed", reports[0]["action"]) + self.assertEqual("5.1.1", reports[0]["status_code"]) + self.assertEqual( + "", + reports[0]["original_message_id"], + ) + + def test_processing_correlates_and_is_idempotent(self) -> None: + provider = SqlMailBounceProcessingProvider() + with self.SessionLocal() as session: + command = submit_delivery_command( + session, + tenant_id="tenant-1", + command_type="campaign_report", + source_module="campaigns", + source_resource_type="campaign", + source_resource_id="campaign-1", + source_version_id="version-1", + idempotency_key="delivery-1", + profile_id="profile-1", + message_bytes=( + b"Message-ID: \r\n" + b"Subject: Original\r\n\r\nBody" + ), + envelope_from="sender@example.test", + envelope_recipients=["recipient@example.test"], + from_header="sender@example.test", + expected_smtp_transport_revision="revision-1", + ) + first = provider.process_raw_message( + session, + tenant_id="tenant-1", + profile_id="profile-1", + folder="INBOX", + uid="42", + raw_message=DSN, + ) + second = provider.process_raw_message( + session, + tenant_id="tenant-1", + profile_id="profile-1", + folder="INBOX", + uid="42", + raw_message=DSN, + ) + + self.assertEqual(command["id"], first[0].command_id) + self.assertTrue(first[0].matched) + self.assertTrue(first[0].permanent) + self.assertEqual(first[0].id, second[0].id) + self.assertEqual(1, session.query(MailBounceObservation).count()) + self.assertEqual( + "pending", + session.get(MailDeliveryCommand, command["id"]).status, + ) + + def test_ordinary_mail_is_not_misclassified(self) -> None: + self.assertEqual( + (), + parse_delivery_status( + b"From: person@example.test\r\nSubject: Hello\r\n\r\nNot a DSN" + ), + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/webui/src/api/mail.ts b/webui/src/api/mail.ts index 8d43396..c86f37a 100644 --- a/webui/src/api/mail.ts +++ b/webui/src/api/mail.ts @@ -139,6 +139,72 @@ export type MailSettingsDeltaResponse = { full: boolean; }; +export type MailBounceSource = { + id: string; + tenant_id: string; + profile_id: string; + folder: string; + imap_server_id?: string | null; + imap_credential_id?: string | null; + expected_imap_transport_revision: string; + is_active: boolean; + uidvalidity?: string | null; + highest_processed_uid: number; + last_scanned_at?: string | null; + last_success_at?: string | null; + last_error?: string | null; +}; + +export type MailBounceObservation = { + id: string; + profile_id: string; + folder: string; + uid: string; + original_message_id?: string | null; + command_id?: string | null; + recipient?: string | null; + action: string; + status_code?: string | null; + diagnostic?: string | null; + permanent: boolean; + observed_at: string; + matched: boolean; + evidence: Record; +}; + +export type MailBounceSourcePayload = { + profile_id: string; + folder: string; + imap_server_id?: string | null; + imap_credential_id?: string | null; + is_active: boolean; +}; + +export async function listMailBounceSources(settings: ApiSettings): Promise { + const response = await apiFetch<{ sources: MailBounceSource[] }>(settings, "/api/v1/mail/bounce-sources"); + return response.sources; +} + +export async function saveMailBounceSource(settings: ApiSettings, payload: MailBounceSourcePayload): Promise { + return apiFetch(settings, "/api/v1/mail/bounce-sources", { + method: "POST", + body: JSON.stringify(payload) + }); +} + +export async function removeMailBounceSource(settings: ApiSettings, sourceId: string): Promise { + await apiFetch(settings, `/api/v1/mail/bounce-sources/${encodeURIComponent(sourceId)}`, { method: "DELETE" }); +} + +export async function scanMailBounceSource(settings: ApiSettings, sourceId: string): Promise<{ sources: number; processed_messages: number; observations: number; failures: Array> }> { + return apiFetch(settings, `/api/v1/mail/bounce-sources/${encodeURIComponent(sourceId)}/scan`, { method: "POST" }); +} + +export async function listMailBounceObservations(settings: ApiSettings, limit = 100): Promise { + const response = await apiFetch<{ observations: MailBounceObservation[] }>(settings, apiPath("/api/v1/mail/bounce-observations", { limit })); + return response.observations; +} + export async function lookupMailAddresses(settings: ApiSettings, query: string, limit = 25): Promise { return apiFetch(settings, apiPath("/api/v1/mail/address-lookup", { query, limit })); } diff --git a/webui/src/features/mail/MailBouncePage.tsx b/webui/src/features/mail/MailBouncePage.tsx new file mode 100644 index 0000000..48dee32 --- /dev/null +++ b/webui/src/features/mail/MailBouncePage.tsx @@ -0,0 +1,216 @@ +import { useEffect, useMemo, useState } from "react"; +import { ArrowLeft, Plus, RefreshCw, RotateCw, Trash2 } from "lucide-react"; +import { + Button, + Card, + DataGrid, + Dialog, + DismissibleAlert, + FormField, + LoadingFrame, + PageScrollViewport, + PageTitle, + StatusBadge, + TableActionGroup, + ToggleSwitch, + adminErrorMessage, + formatDateTime, + useGuardedNavigate, + type ApiSettings, + type DataGridColumn +} from "@govoplan/core-webui"; +import { + listMailBounceObservations, + listMailBounceSources, + listMailServerProfiles, + removeMailBounceSource, + saveMailBounceSource, + scanMailBounceSource, + type MailBounceObservation, + type MailBounceSource, + type MailServerProfile +} from "../../api/mail"; + +export default function MailBouncePage({ settings }: { settings: ApiSettings }) { + const navigate = useGuardedNavigate(); + const [sources, setSources] = useState([]); + const [observations, setObservations] = useState([]); + const [profiles, setProfiles] = useState([]); + const [loading, setLoading] = useState(true); + const [busy, setBusy] = useState(""); + const [error, setError] = useState(""); + const [message, setMessage] = useState(""); + const [addOpen, setAddOpen] = useState(false); + const [deleteSource, setDeleteSource] = useState(null); + const [profileId, setProfileId] = useState(""); + const [folder, setFolder] = useState("INBOX"); + const [active, setActive] = useState(true); + + const profileNames = useMemo( + () => new Map(profiles.map((profile) => [profile.id, profile.name])), + [profiles] + ); + + async function load() { + setLoading(true); + setError(""); + try { + const [nextSources, nextObservations, nextProfiles] = await Promise.all([ + listMailBounceSources(settings), + listMailBounceObservations(settings), + listMailServerProfiles(settings, true) + ]); + setSources(nextSources); + setObservations(nextObservations); + setProfiles(nextProfiles); + setProfileId((current) => current || nextProfiles.find((profile) => profile.imap)?.id || ""); + } catch (err) { + setError(adminErrorMessage(err)); + } finally { + setLoading(false); + } + } + + useEffect(() => { void load(); }, [settings.apiBaseUrl, settings.apiKey, settings.accessToken]); + + async function addSource() { + if (!profileId || !folder.trim()) return; + setBusy("add"); + setError(""); + try { + await saveMailBounceSource(settings, { + profile_id: profileId, + folder: folder.trim(), + is_active: active + }); + setAddOpen(false); + setMessage("Bounce mailbox watcher saved."); + await load(); + } catch (err) { + setError(adminErrorMessage(err)); + } finally { + setBusy(""); + } + } + + async function scan(source: MailBounceSource) { + setBusy(`scan:${source.id}`); + setError(""); + try { + const result = await scanMailBounceSource(settings, source.id); + setMessage(`Processed ${result.processed_messages} message(s) and recorded ${result.observations} bounce observation(s).`); + await load(); + } catch (err) { + setError(adminErrorMessage(err)); + } finally { + setBusy(""); + } + } + + async function removeSource() { + if (!deleteSource) return; + setBusy(`delete:${deleteSource.id}`); + setError(""); + try { + await removeMailBounceSource(settings, deleteSource.id); + setDeleteSource(null); + setMessage("Bounce mailbox watcher removed. Existing observations were retained."); + await load(); + } catch (err) { + setError(adminErrorMessage(err)); + } finally { + setBusy(""); + } + } + + const sourceColumns: DataGridColumn[] = [ + { + id: "profile", + header: "Mail profile", + width: "minmax(180px, 1fr)", + value: (source) => profileNames.get(source.profile_id) || source.profile_id, + render: (source) => {profileNames.get(source.profile_id) || source.profile_id} + }, + { id: "folder", header: "Folder", width: "minmax(150px, .8fr)", value: (source) => source.folder }, + { + id: "status", + header: "Status", + width: 130, + value: (source) => source.last_error ? "error" : source.is_active ? "active" : "inactive", + render: (source) => + }, + { id: "cursor", header: "Last UID", width: 110, value: (source) => source.highest_processed_uid }, + { + id: "lastScan", + header: "Last scan", + width: "minmax(180px, .8fr)", + value: (source) => source.last_scanned_at || "", + render: (source) => source.last_scanned_at ? formatDateTime(source.last_scanned_at) : "Never" + }, + { + id: "actions", + header: "Actions", + width: 92, + sticky: "end", + render: (source) =>