"""The email side of the MAS mode contract, in both roles (CONTRACT.md section 3). Takes normalized gateway events (POST /mas-events), mode statuses (POST /mode-status) and, in the slave role, outside-master commands (POST /master). The role is configured per resource type: - master: the case lives in a `MasterStore` (the reference ticket service). Email negotiates by mail, hands off, records every status in the thread, owns the deadline (`tick()`) and resumes by mail when a handoff ends. - slave: an outside master issues case ids, gets every inbound mail as an event, and decides what email sends and where mode requests go. Every outbound mail goes through principle 10 (`mailer.refusal`), whoever asked for it. Outbound contract messages are queued while the lock is held and sent after it is released, so peers may call back at once. """ from __future__ import annotations import json import re import sqlite3 import threading import time import uuid from dataclasses import dataclass, field from typing import Callable, Optional, Protocol, Union from mas_core.generator import resource_local_part from mas_core.parser import MasSyntaxError, parse_address from mas_core.policy import resolve from mas_core.statute import Statute from . import message_errors from .contract import TERMINAL, Idempotent, allowed, json_server, message, post, refuse from .mailer import Mailer NOT_ENABLED = {"InteractionModeNotEnabled", "ProfileNotEnabled"} SCHEMA = """ CREATE TABLE IF NOT EXISTS handoffs ( id TEXT PRIMARY KEY, role TEXT, resource_type TEXT, case_address TEXT, case_key TEXT, case_ref TEXT, mode TEXT, status TEXT, seq INTEGER DEFAULT 0, deadline REAL, needs TEXT, participant TEXT, endpoint TEXT, relay_seq INTEGER DEFAULT 0 ); CREATE TABLE IF NOT EXISTS intakes ( ref TEXT PRIMARY KEY, resource_type TEXT, domain TEXT, message_ref TEXT, requested_mode TEXT, resource_id TEXT ); CREATE TABLE IF NOT EXISTS cases ( key TEXT PRIMARY KEY, address TEXT, resource_type TEXT, resource_id TEXT, policy TEXT ); CREATE TABLE IF NOT EXISTS case_messages ( ref TEXT PRIMARY KEY, case_key TEXT, message TEXT, disposition TEXT ); """ class MasterStore(Protocol): """Where email-mastered cases live (implemented by the reference ticket service).""" def handle(self, event: dict) -> tuple[int, dict]: ... def open_case(self, response: dict) -> Optional[str]: ... # case ref, if open def case_address(self, case_ref: str) -> str: ... def record(self, case_ref: str, kind: str, text: str) -> None: ... def reply(self, case_ref: str, msg: dict, subject: str, body: str) -> bool: ... @dataclass class RoleConfig: role: str # "master" | "slave" modes: dict = field(default_factory=dict) # master: mode -> endpoint URL fields: tuple = ("phone", "time") # details read from "Key: value" lines timeout: float = 3600.0 # master: handoff deadline (seconds) master_url: Optional[str] = None # slave: outside master's event URL autonomous_send: bool = False # slave: default policy transfers: Union[str, dict] = "block" # slave: "block" | {"webhook": URL} def extract_details(text: str, fields) -> dict: """`Phone: +49 ...` style lines, for the fields a mode asks for.""" out = {} for line in (text or "").splitlines(): m = re.match(r"\s*([A-Za-z_-]+)\s*:\s*(.+?)\s*$", line) if m and m.group(1).lower() in fields and m.group(1).lower() not in out: out[m.group(1).lower()] = m.group(2) return out def _resume_text(mode: str, status: str, info: dict) -> str: why = info.get("reason") or info.get("detail") or "" texts = { "completed": f"Your {mode} has taken place.", "failed": f"Your {mode} did not succeed{f' ({why})' if why else ''}.", "expired": f"Your {mode} request expired{f' ({why})' if why else ''}.", "refused": f"We cannot continue by {mode} ({why}).", "cancelled": f"Your {mode} was cancelled.", "timed-out": f"We could not arrange your {mode} in time.", } return texts[status] + " We continue by email: reply to this message." class EmailMode: def __init__(self, roles: dict[str, RoleConfig], store: Optional[MasterStore] = None, db_path: str = ":memory:", statutes: Optional[dict[str, Statute]] = None, clock: Callable[[], float] = time.time, smtp=None, max_sent=None, require_dmarc_pass: bool = True): if any(c.role == "master" for c in roles.values()) and store is None: raise ValueError("a master role needs a MasterStore") self.roles = roles self.store = store self.statutes = statutes or {} self.clock = clock self.db = sqlite3.connect(db_path, check_same_thread=False) self.db.row_factory = sqlite3.Row self.db.executescript(SCHEMA) self.mailer = Mailer(self.db, smtp, max_sent, require_dmarc_pass) self.lock = threading.RLock() self.base_url = "" # set by serve(); status_url = base_url + /mode-status self._queue: list = [] self._once = Idempotent() # -- HTTP entry points --------------------------------------------------------- def server(self, host="127.0.0.1", port=0): return json_server({"/mas-events": self.handle_event, "/mode-status": self.mode_status, "/master": self.master_command, "/tick": lambda body: (200, {"ok": True, "timed_out": self.tick()})}, host, port) def _run(self, fn, *args): with self.lock, self.db: out = fn(*args) self._flush() return out def _send(self, url, msg, on_result=None): self._queue.append((url, msg, on_result)) def _flush(self): while True: with self.lock: if not self._queue: return url, msg, on_result = self._queue.pop(0) status, body = post(url, msg) if on_result: with self.lock, self.db: on_result(status, body) # -- inbound mail -------------------------------------------------------------- def handle_event(self, event: dict) -> tuple[int, dict]: mas = event.get("mas", {}) rtype = mas.get("resource_type") if mas.get("kind") == "resource" else mas.get("intake_for") cfg = self.roles.get(rtype) if mas.get("valid") and mas.get("namespace_confirmed") else None if cfg is None: return self.store.handle(event) if self.store else (404, {"status": "not-handled"}) if cfg.role == "master": return self._run(self._master_event, event, rtype, cfg) return self._run(self._slave_event, event, rtype, cfg) def _case(self, address: str) -> dict: try: addr = parse_address(address) except MasSyntaxError: return {"address": address, "key": address.rsplit("@", 1)[0]} sem = resolve(addr, self.statutes.get(addr.domain)) return {"address": address, "key": sem.equivalence_key} def _open_handoff(self, case_key): rows = self.db.execute("SELECT * FROM handoffs WHERE case_key=? ORDER BY rowid", (case_key,)) return next((r for r in rows if r["status"] not in TERMINAL), None) def _master_event(self, event, rtype, cfg): code, resp = self.store.handle(event) ref = self.store.open_case(resp) if ref is None: return code, resp mas, msg = event["mas"], event.get("message", {}) case = self._case(self.store.case_address(ref)) h = self._open_handoff(case["key"]) if h is not None: if h["status"] == "negotiating": details = extract_details(msg.get("text"), json.loads(h["needs"] or "[]")) if details: self.store.record(ref, "mode-update", f"{h['mode']}: details {sorted(details)}") self._send(h["endpoint"], message("handoff.update", case, handoff_id=h["id"], details=details)) return code, {**resp, "handoff_id": h["id"]} mode = mas.get("interaction") if not mode: return code, resp failures = {f["code"] for f in mas.get("failures", [])} hid = self._new_handoff("master", rtype, case, ref, mode, msg, cfg.modes.get(mode), self.clock() + cfg.timeout) if failures & NOT_ENABLED: # The creation reply already says so; only an existing case gets a notice. self._status(hid, "refused", {"reason": "mode-not-enabled"}, notify=resp.get("status") != "created") elif mode not in cfg.modes: self._status(hid, "refused", {"reason": "mode-unavailable"}) else: self._request(hid, case, mode, extract_details(msg.get("text"), cfg.fields), msg) return code, {**resp, "handoff_id": hid} def _new_handoff(self, role, rtype, case, ref, mode, msg, endpoint, deadline): hid = "h-" + uuid.uuid4().hex[:12] self.db.execute("INSERT INTO handoffs (id, role, resource_type, case_address, case_key," " case_ref, mode, status, deadline, participant, endpoint)" " VALUES (?,?,?,?,?,?,?,?,?,?,?)", (hid, role, rtype, case["address"], case["key"], ref, mode, None, deadline, json.dumps(msg), endpoint)) return hid def _request(self, hid, case, mode, details, msg): h = self._handoff(hid) self._status(hid, "requested", {}) req = message("handoff.request", case, handoff_id=hid, mode=mode, details=details, status_url=self.base_url + "/mode-status", participant={"address": msg.get("from")} if msg.get("from") else None) def on_result(status, body): if status is None or status >= 400: self._status(hid, "refused", {"reason": "mode-unavailable", "detail": f"mode endpoint answered {status}"}) self._send(h["endpoint"], req, on_result) def _handoff(self, hid): return self.db.execute("SELECT * FROM handoffs WHERE id=?", (hid,)).fetchone() # -- statuses ------------------------------------------------------------------ def mode_status(self, msg: dict) -> tuple[int, dict]: return self._once(msg, lambda m: self._run(self._mode_status, m)) def _mode_status(self, msg): errors = message_errors(msg) if errors or msg.get("type") != "handoff.status": return refuse("schema", "; ".join(errors) or "expected handoff.status", 400) h = self._handoff(msg["handoff_id"]) if h is None or h["case_key"] != msg["case"]["key"]: return refuse("unknown-handoff", msg["handoff_id"], 404) if msg["seq"] <= h["seq"]: return 200, {"ok": True, "ignored": "stale"} if msg["status"] == "timed-out": return refuse("invalid-transition", "only the timeout owner records timed-out") if not allowed(h["status"], msg["status"]): return refuse("invalid-transition", f"{h['status']} -> {msg['status']}") self.db.execute("UPDATE handoffs SET seq=? WHERE id=?", (msg["seq"], h["id"])) info = {k: msg[k] for k in ("needs", "at", "outcome", "reason", "detail") if k in msg} self._status(h["id"], msg["status"], info) return 200, {"ok": True} def _status(self, hid, status, info, notify=True): """Record a status and react to it, per role.""" h = self._handoff(hid) self.db.execute("UPDATE handoffs SET status=?, needs=? WHERE id=?", (status, json.dumps(info.get("needs")) if "needs" in info else h["needs"], hid)) case = {"address": h["case_address"], "key": h["case_key"]} participant = json.loads(h["participant"]) if h["role"] == "master": self.store.record(h["case_ref"], "mode-status", f"{h['mode']}: {status}" + "".join(f" {k}={v}" for k, v in info.items())) if status == "negotiating": self._ask(h, participant, info["needs"], lambda subj, body: self.store.reply(h["case_ref"], participant, subj, body)) elif status in TERMINAL and notify: self.store.reply(h["case_ref"], participant, participant.get("subject") or "Your request", _resume_text(h["mode"], status, info)) return # slave: the master owns the flow; relay, and ask only when allowed to cfg = self.roles[h["resource_type"]] if status != "requested": self._send(cfg.master_url, message("handoff.status", case, handoff_id=hid, seq=self._relay_seq(hid), status=status, **info)) if status == "negotiating" and self._policy(h["case_key"])["autonomous_send"]: self._ask(h, participant, info["needs"], lambda subj, body: self.mailer.reply( h["case_key"], h["case_address"], participant, subj, body) is None) def _relay_seq(self, hid): # Statuses relayed to the master are numbered by email, per handoff. self.db.execute("UPDATE handoffs SET relay_seq=relay_seq+1 WHERE id=?", (hid,)) return self._handoff(hid)["relay_seq"] def _ask(self, h, participant, needs, send): lines = [f"To arrange the {h['mode']}, please reply to this message with:"] lines += [f"{n.capitalize()}: ..." for n in needs] send(participant.get("subject") or "Your request", "\n".join(lines)) def tick(self) -> list[str]: """Master role: time out handoffs past their deadline and resume by email.""" with self.lock, self.db: now = self.clock() rows = self.db.execute("SELECT * FROM handoffs WHERE role='master' AND deadline<=?", (now,)).fetchall() out = [] for h in rows: if h["status"] in TERMINAL or h["status"] is None: continue self._status(h["id"], "timed-out", {}) if h["endpoint"]: self._send(h["endpoint"], message( "handoff.cancel", {"address": h["case_address"], "key": h["case_key"]}, handoff_id=h["id"], reason="timed-out")) out.append(h["id"]) self._flush() return out # -- slave role ---------------------------------------------------------------- def _store_message(self, case_key, msg, disposition): ref = "m-" + uuid.uuid4().hex[:12] self.db.execute("INSERT INTO case_messages VALUES (?,?,?,?)", (ref, case_key, json.dumps(msg), disposition)) return ref def _slave_event(self, event, rtype, cfg): mas, msg = event["mas"], event.get("message", {}) if mas.get("kind") != "resource": iref = "i-" + uuid.uuid4().hex[:12] mref = self._store_message(None, msg, "intake") self.db.execute("INSERT INTO intakes VALUES (?,?,?,?,?,NULL)", (iref, rtype, mas["domain"], mref, mas.get("interaction"))) self._send(cfg.master_url, message("case.intake", intake_ref=iref, resource_type=rtype, event=event, requested_mode=mas.get("interaction"))) return 202, {"status": "intake-reported", "intake_ref": iref} row = self.db.execute("SELECT * FROM cases WHERE key=?", (mas["equivalence_key"],)).fetchone() if row is None: self._store_message(None, msg, "quarantined-unknown") return 202, {"status": "unknown-resource", "action": "quarantined, no reply"} case = {"address": row["address"], "key": row["key"]} mref = self._store_message(row["key"], msg, "attached") self._send(cfg.master_url, message("case.message", case, event=event, message_ref=mref)) out = {"status": "reported", "case": case, "message_ref": mref} h = self._open_handoff(row["key"]) if h is not None and h["status"] == "negotiating" and h["endpoint"]: details = extract_details(msg.get("text"), json.loads(h["needs"] or "[]")) if details: self._send(h["endpoint"], message("handoff.update", case, handoff_id=h["id"], details=details)) elif h is None and mas.get("interaction"): out["handoff_id"] = self._transfer(rtype, cfg, case, mas["interaction"], msg) return 200, out def _policy(self, case_key) -> dict: row = self.db.execute("SELECT policy FROM cases WHERE key=?", (case_key,)).fetchone() return json.loads(row["policy"]) def _transfer(self, rtype, cfg, case, mode, msg): """A !MODE request in the slave role: blocked, or sent to the primed webhook.""" transfers = self._policy(case["key"])["transfers"] webhook = transfers.get("webhook") if isinstance(transfers, dict) else None hid = self._new_handoff("slave", rtype, case, case["key"], mode, msg, webhook, None) self._send(cfg.master_url, message("handoff.requested", case, handoff_id=hid, mode=mode, forwarded=webhook is not None)) if webhook is None: self._status(hid, "requested", {}) self._status(hid, "refused", {"reason": "blocked"}) else: self._request(hid, case, mode, extract_details(msg.get("text"), cfg.fields), msg) return hid def master_command(self, msg: dict) -> tuple[int, dict]: return self._once(msg, lambda m: self._run(self._master_command, m)) def _master_command(self, msg): errors = message_errors(msg) if errors: return refuse("schema", "; ".join(errors), 400) t = msg["type"] if t == "case.assign": return self._assign(msg) row = self.db.execute("SELECT * FROM cases WHERE key=?", (msg["case"]["key"],)).fetchone() if row is None: return refuse("unknown-case", msg["case"]["key"], 404) if t == "case.policy": pol = {**self._policy(row["key"]), **{k: msg[k] for k in ("autonomous_send", "transfers") if k in msg}} self.db.execute("UPDATE cases SET policy=? WHERE key=?", (json.dumps(pol), row["key"])) return 200, {"ok": True, "policy": pol} if t == "mail.send": m = self.db.execute("SELECT * FROM case_messages WHERE ref=? AND case_key=?", (msg["in_reply_to"], row["key"])).fetchone() if m is None: return refuse("refused", msg["in_reply_to"], reason="unknown-message") orig = json.loads(m["message"]) why = self.mailer.reply(row["key"], row["address"], orig, msg.get("subject") or orig.get("subject") or "Your request", msg["body"]) return (200, {"ok": True, "sent": True}) if why is None else refuse("refused", why, reason=why) if t == "handoff.cancel": h = self._handoff(msg["handoff_id"]) if h is None or h["case_key"] != row["key"]: return refuse("unknown-handoff", msg["handoff_id"], 404) if h["endpoint"] and h["status"] not in TERMINAL: self._send(h["endpoint"], message("handoff.cancel", msg["case"], handoff_id=h["id"], reason=msg.get("reason"))) return 200, {"ok": True} return refuse("unsupported", t, 400) def _assign(self, msg): it = self.db.execute("SELECT * FROM intakes WHERE ref=?", (msg["intake_ref"],)).fetchone() if it is None: return refuse("unknown-intake", msg["intake_ref"], 404) if it["resource_id"] is not None: return refuse("already-assigned", it["resource_id"]) try: address = f"{resource_local_part(it['resource_type'], msg['resource_id'])}@{it['domain']}" except MasSyntaxError as e: return refuse("invalid-resource-id", str(e), 400) case = self._case(address) if self.db.execute("SELECT 1 FROM cases WHERE key=?", (case["key"],)).fetchone(): return refuse("case-exists", case["key"]) cfg = self.roles[it["resource_type"]] pol = {"autonomous_send": cfg.autonomous_send, "transfers": cfg.transfers, **msg.get("policy", {})} self.db.execute("INSERT INTO cases VALUES (?,?,?,?,?)", (case["key"], address, it["resource_type"], msg["resource_id"], json.dumps(pol))) self.db.execute("UPDATE intakes SET resource_id=? WHERE ref=?", (msg["resource_id"], it["ref"])) self.db.execute("UPDATE case_messages SET case_key=? WHERE ref=?", (case["key"], it["message_ref"])) intake = json.loads(self.db.execute("SELECT message FROM case_messages WHERE ref=?", (it["message_ref"],)).fetchone()["message"]) out = {"ok": True, "case": case, "message_ref": it["message_ref"]} if pol["autonomous_send"]: why = self.mailer.reply(case["key"], address, intake, intake.get("subject") or "Your request", f"Your request is now case {msg['resource_id']}.\n" f"Reply to {address} to add to it.") out["acknowledged"] = why is None if it["requested_mode"]: out["handoff_id"] = self._transfer(it["resource_type"], cfg, case, it["requested_mode"], intake) return 200, out