tegwick
aef4c5dd71
mas-email-evidence 0.1 reports MX acceptance, a bounce, or a reply, and refuses the claim that a person read the message. Ordinary product mail stays in email-connect. A product speaks the mode contract. Assistant: grok Assistant-Session: 01a198fa-fd27-7e41-a0a2-57355c3e4efd
127 lines
4.1 KiB
Python
127 lines
4.1 KiB
Python
"""Map a MAS gateway event into the email-connect evidence ceiling.
|
|
|
|
The normative outcome table is docs/mas-email-evidence-v0.1.json. The same
|
|
file is vendored in email-connect. A gateway event means the MAS MX took the
|
|
message. Bounce and reply are properties of that message. Nothing here claims
|
|
that a person read it.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
|
|
CONTRACT_PATH = Path(__file__).resolve().parents[1] / "docs" / "mas-email-evidence-v0.1.json"
|
|
_CONTRACT = json.loads(CONTRACT_PATH.read_text())
|
|
_ENHANCED = re.compile(r"\b([45]\.\d{1,3}\.\d{1,3})\b")
|
|
_SMTP = re.compile(r"\b([45]\d{2})\b")
|
|
_DSN_MARKERS = (
|
|
"mailer-daemon",
|
|
"delivery status notification",
|
|
"undelivered mail",
|
|
"mail delivery failed",
|
|
"final-recipient",
|
|
"diagnostic-code",
|
|
)
|
|
_PERMANENT_TEXT = ("user unknown", "mailbox not found", "unknown user", "undeliverable")
|
|
_TEMPORARY_TEXT = ("mailbox full", "try again later", "temporary failure")
|
|
|
|
|
|
def from_gateway_event(event: object) -> dict:
|
|
"""Return one evidence record for a gateway event."""
|
|
if not _is_event(event):
|
|
return _record("parse_failed", None)
|
|
message = event["message"]
|
|
blob = _blob(message)
|
|
if _is_bounce(message, blob):
|
|
if _permanent(blob):
|
|
outcome = "bounced_permanent"
|
|
elif _temporary(blob):
|
|
outcome = "bounced_temporary"
|
|
else:
|
|
outcome = "bounce_unclassified"
|
|
elif _is_out_of_office(message):
|
|
outcome = "out_of_office"
|
|
elif _is_reply(message):
|
|
outcome = "reply_received"
|
|
else:
|
|
outcome = "mx_accepted"
|
|
return _record(outcome, event)
|
|
|
|
|
|
def _is_event(event: object) -> bool:
|
|
return (
|
|
isinstance(event, dict)
|
|
and event.get("mas_event_version") == "0.1"
|
|
and isinstance(event.get("message"), dict)
|
|
and isinstance(event.get("recipient"), str)
|
|
)
|
|
|
|
|
|
def _blob(message: dict) -> str:
|
|
parts = [
|
|
message.get("from") or "",
|
|
message.get("subject") or "",
|
|
message.get("auto_submitted") or "",
|
|
message.get("text") or "",
|
|
]
|
|
return "\n".join(str(part) for part in parts).lower()
|
|
|
|
|
|
def _is_bounce(message: dict, blob: str) -> bool:
|
|
sender = str(message.get("from") or "").lower()
|
|
if "mailer-daemon" in sender or "postmaster@" in sender:
|
|
return True
|
|
return any(marker in blob for marker in _DSN_MARKERS if marker != "mailer-daemon")
|
|
|
|
|
|
def _permanent(blob: str) -> bool:
|
|
enhanced = _ENHANCED.search(blob)
|
|
smtp = _SMTP.search(blob)
|
|
if enhanced and enhanced.group(1).startswith("5."):
|
|
return True
|
|
if smtp and smtp.group(1).startswith("5"):
|
|
return True
|
|
return any(marker in blob for marker in _PERMANENT_TEXT)
|
|
|
|
|
|
def _temporary(blob: str) -> bool:
|
|
enhanced = _ENHANCED.search(blob)
|
|
smtp = _SMTP.search(blob)
|
|
if enhanced and enhanced.group(1).startswith("4."):
|
|
return True
|
|
if smtp and smtp.group(1).startswith("4"):
|
|
return True
|
|
return any(marker in blob for marker in _TEMPORARY_TEXT)
|
|
|
|
|
|
def _is_out_of_office(message: dict) -> bool:
|
|
auto = str(message.get("auto_submitted") or "").lower()
|
|
return auto.startswith("auto-replied")
|
|
|
|
|
|
def _is_reply(message: dict) -> bool:
|
|
reply = str(message.get("in_reply_to") or "").strip()
|
|
auto = str(message.get("auto_submitted") or "").lower()
|
|
return bool(reply) and auto in ("", "no")
|
|
|
|
|
|
def _record(outcome: str, event: dict | None) -> dict:
|
|
spec = _CONTRACT["outcomes"][outcome]
|
|
message = (event or {}).get("message") or {}
|
|
return {
|
|
"contract": _CONTRACT["contract"],
|
|
"contract_version": _CONTRACT["version"],
|
|
"source": _CONTRACT["source"],
|
|
"outcome": outcome,
|
|
"event_type": spec["event_type"],
|
|
"assessment_category": spec["assessment_category"],
|
|
"assessment_subclass": spec["assessment_subclass"],
|
|
"evidence_strength": spec["evidence_strength"],
|
|
"normalized_meaning": spec["normalized_meaning"],
|
|
"evidence_ceiling": _CONTRACT["evidence_ceiling"],
|
|
"recipient": (event or {}).get("recipient"),
|
|
"message_id": message.get("message_id"),
|
|
"in_reply_to": message.get("in_reply_to"),
|
|
}
|