MAS-WP-0005-T12: gateway reads trusted A-R verdict; ticket service replies only on DMARC pass

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 252470@bnt-lap001
Assistant-Session: ad628c8f-4be2-4979-94c9-76c7ed6f8241

deploy/railiance/entrypoint.sh

@@ -63,7 +63,7 @@ if [ -n "$MAS_RELAY" ] && [ -s /etc/mas-relay/username ]; then
   # service cap below really bounds everything sent through TEM.
   postconf -e "sender_dependent_default_transport_maps=inline:{<>=discard:}" \
     "empty_address_default_transport_maps_lookup_key=<>"
-  SMTP_ARG="--smtp 127.0.0.1:25 --max-sent ${MAS_MAX_SENT:-300}"
+  SMTP_ARG="--smtp 127.0.0.1:25 --max-sent ${MAS_MAX_SENT:-300} --require-dmarc-pass"
 else
   postconf -e "default_transport=error:5.7.1 outbound mail disabled"
 fi
@@ -111,7 +111,7 @@ postconf -e "header_checks=regexp:/etc/postfix/ar_strip.regexp" "smtpd_milters=i
 cat >> /etc/postfix/master.cf <<CF
 mas       unix  -       n       n       -       -       pipe
   flags=Rq user=mas argv=/usr/local/bin/mas-gateway
-  --statute $STATUTE_DIR/.well-known/mas
+  --statute $STATUTE_DIR/.well-known/mas --trusted-authserv-id $DOMAIN
   pipe --app-url http://127.0.0.1:8024/mas-events --recipient \${recipient}
 CF
 

examples/ticket_service/ticket_service.py

@@ -65,7 +65,7 @@ CREATE TABLE IF NOT EXISTS outbox (
 class TicketService:
     def __init__(self, db_path: str = ":memory:", id_factory: Callable[[], str] = default_ticket_id,
                  smtp: Optional[tuple[str, int]] = None, closed_notice: bool = True,
-                 max_sent: Optional[int] = None):
+                 max_sent: Optional[int] = None, require_dmarc_pass: bool = False):
         self.db = sqlite3.connect(db_path, check_same_thread=False)
         self.db.row_factory = sqlite3.Row
         self.db.executescript(SCHEMA)
@@ -74,6 +74,8 @@ class TicketService:
         self.smtp = smtp
         self.closed_notice = closed_notice
         self.max_sent = max_sent  # lifetime cap on mail actually sent via smtp
+        # spec 14 item 10: auto-reply only to authenticated senders
+        self.require_dmarc_pass = require_dmarc_pass
 
     # -- event handling -------------------------------------------------------------
 
@@ -149,6 +151,8 @@ class TicketService:
         bulk = (msg.get("precedence") or "").lower() in ("bulk", "list", "junk")
         if not to or auto or bulk:
             return False  # never auto-reply to automated mail
+        if self.require_dmarc_pass and (msg.get("authentication") or {}).get("dmarc") != "pass":
+            return False  # unauthenticated sender: keep the ticket, send nothing
         if not subject.lower().startswith("re:"):
             subject = "Re: " + subject
         self.db.execute("INSERT INTO outbox (ticket_id, mail_from, mail_to, subject, body, in_reply_to)"
@@ -241,10 +245,12 @@ if __name__ == "__main__":
     ap.add_argument("--db", default="tickets.sqlite3")
     ap.add_argument("--listen", default="127.0.0.1:8025")
     ap.add_argument("--smtp", metavar="HOST:PORT", help="send replies via this SMTP relay")
+    ap.add_argument("--require-dmarc-pass", action="store_true",
+                    help="auto-reply only when the event says the sender passed DMARC")
     ap.add_argument("--max-sent", type=int, metavar="N",
                     help="stop sending once N replies have been sent in total (kept in the db)")
     a = ap.parse_args()
     smtp = (a.smtp.rsplit(":", 1)[0], int(a.smtp.rsplit(":", 1)[1])) if a.smtp else None
     h, p = a.listen.rsplit(":", 1)
     print(f"ticket service on http://{a.listen}/mas-events")
-    make_server(TicketService(a.db, smtp=smtp, max_sent=a.max_sent), h, int(p)).serve_forever()
+    make_server(TicketService(a.db, smtp=smtp, max_sent=a.max_sent, require_dmarc_pass=a.require_dmarc_pass), h, int(p)).serve_forever()

mas_gateway/core.py

@@ -29,7 +29,45 @@ from mas_core.statute import build_statute
 EVENT_VERSION = "0.1"
 
 
-def message_meta_from_rfc822(raw: bytes) -> dict:
+def authentication_from_headers(headers: list[tuple[str, str]], trusted_authserv_id: Optional[str]) -> Optional[dict]:
+    """Sender-authentication verdict from our own MTA's Authentication-Results.
+
+    Only A-R headers whose authserv-id is `trusted_authserv_id` count, and only
+    those above the first Received header, i.e. added at our hop (RFC 8601
+    section 5). Anything a sender brought along sits below that line. The MTA
+    also strips incoming A-R headers claiming our id; this is the second layer.
+    """
+    if not trusted_authserv_id:
+        return None
+    want = trusted_authserv_id.lower()
+    verdict: dict = {}
+    for name, value in headers:
+        lname = name.lower()
+        if lname == "received":
+            break
+        if lname != "authentication-results":
+            continue
+        authserv, _, results = str(value).partition(";")
+        ids = authserv.split()
+        if not ids or ids[0].lower() != want:
+            continue
+        for clause in results.split(";"):
+            words = clause.split()
+            if not words or "=" not in words[0]:
+                continue
+            method, result = words[0].split("=", 1)
+            method, result = method.lower(), result.lower()
+            if method in ("dmarc", "spf", "dkim"):
+                # several DKIM signatures: one pass is enough
+                if not (method == "dkim" and verdict.get("dkim") == "pass"):
+                    verdict[method] = result
+            for w in words[1:]:
+                if method == "dmarc" and w.lower().startswith("header.from="):
+                    verdict["header_from"] = w.split("=", 1)[1]
+    return verdict or None
+
+
+def message_meta_from_rfc822(raw: bytes, trusted_authserv_id: Optional[str] = None) -> dict:
     msg: EmailMessage = BytesParser(policy=policy.default).parsebytes(raw)
     body = msg.get_body(preferencelist=("plain",))
     text = body.get_content() if body is not None else ""
@@ -43,6 +81,7 @@ def message_meta_from_rfc822(raw: bytes) -> dict:
         "date": str(msg.get("Date", "")) or None,
         "auto_submitted": str(msg.get("Auto-Submitted", "")) or None,
         "precedence": str(msg.get("Precedence", "")) or None,
+        "authentication": authentication_from_headers(msg.items(), trusted_authserv_id),
         "text": text,
     }
 

mas_gateway/main.py

@@ -23,6 +23,7 @@ import argparse
 import json
 import sys
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from typing import Optional
 from urllib.parse import parse_qs, urlsplit
 
 from .core import Resolver, gateway, message_meta_from_rfc822
@@ -33,26 +34,30 @@ MAX_BODY = 25 * 1024 * 1024
 
 def cmd_pipe(args, resolver) -> int:
     raw = sys.stdin.buffer.read()
-    deliveries = gateway(args.recipient, message_meta_from_rfc822(raw), resolver, args.app_url)
+    meta = message_meta_from_rfc822(raw, args.trusted_authserv_id)
+    deliveries = gateway(args.recipient, meta, resolver, args.app_url)
     for d in deliveries:
         print(json.dumps(d.__dict__), file=sys.stderr)
     return EX_TEMPFAIL if any(d.outcome == "tempfail" for d in deliveries) else 0
 
 
-def webhook_message(body: bytes, ctype: str, query: dict) -> tuple[list[str], dict]:
+def webhook_message(body: bytes, ctype: str, query: dict,
+                    trusted_authserv_id: Optional[str] = None) -> tuple[list[str], dict]:
     if ctype.startswith("message/rfc822"):
-        return query.get("recipient", []), message_meta_from_rfc822(body)
+        return query.get("recipient", []), message_meta_from_rfc822(body, trusted_authserv_id)
     data = json.loads(body.decode())
     recipients = data.pop("recipients", None) or [data.pop("recipient")]
     meta = {"from": data.get("from"), "to": data.get("to", recipients),
             "subject": data.get("subject", ""), "message_id": data.get("message_id"),
             "in_reply_to": data.get("in_reply_to"), "references": data.get("references"),
             "date": data.get("date"), "auto_submitted": data.get("auto_submitted"),
-            "precedence": data.get("precedence"), "text": data.get("text", "")}
+            "precedence": data.get("precedence"), "text": data.get("text", ""),
+            "authentication": data.get("authentication")}  # as verified by the inbound provider
     return recipients, meta
 
 
-def make_webhook_server(listen: str, app_url: str, resolver: Resolver) -> ThreadingHTTPServer:
+def make_webhook_server(listen: str, app_url: str, resolver: Resolver,
+                        trusted_authserv_id: Optional[str] = None) -> ThreadingHTTPServer:
     host, port = listen.rsplit(":", 1)
 
     class Handler(BaseHTTPRequestHandler):
@@ -63,7 +68,8 @@ def make_webhook_server(listen: str, app_url: str, resolver: Resolver) -> Thread
             body = self.rfile.read(length)
             try:
                 recipients, meta = webhook_message(body, self.headers.get("Content-Type", ""),
-                                                   parse_qs(urlsplit(self.path).query))
+                                                   parse_qs(urlsplit(self.path).query),
+                                                   trusted_authserv_id)
             except (ValueError, KeyError) as e:
                 return self._reply(400, {"error": f"bad inbound payload: {e}"})
             if not recipients:
@@ -92,6 +98,9 @@ def main(argv=None) -> int:
     ap.add_argument("--statute", action="append", metavar="FILE",
                     help="pin a statute from local configuration (keyed by its namespace)")
     ap.add_argument("--registry", metavar="FILE", help="local registry JSON")
+    ap.add_argument("--trusted-authserv-id", metavar="ID",
+                    help="read sender authentication only from Authentication-Results with this id "
+                         "(our own MTA); without it, no verdict is reported")
     sub = ap.add_subparsers(dest="cmd", required=True)
     p = sub.add_parser("pipe")
     p.add_argument("--recipient", action="append", required=True)
@@ -104,7 +113,7 @@ def main(argv=None) -> int:
     resolver = Resolver.with_statute_files(args.statute, local_registry=registry)
     if args.cmd == "pipe":
         return cmd_pipe(args, resolver)
-    srv = make_webhook_server(args.listen, args.app_url, resolver)
+    srv = make_webhook_server(args.listen, args.app_url, resolver, args.trusted_authserv_id)
     print(f"mas-gateway webhook on http://{args.listen} -> {args.app_url}", file=sys.stderr)
     srv.serve_forever()
     return 0

tests/test_end_to_end.py

@@ -203,3 +203,47 @@ def test_gateway_is_resource_type_agnostic():
             ("resource", "invoice", "RE=2026-4711", "dispute")
     finally:
         srv.shutdown()
+
+
+# -- sender authentication (spec 14 item 10) ----------------------------------------
+
+AR_OURS = "Authentication-Results: great-service.example; dkim=pass header.d=example.net; " \
+          "spf=pass smtp.mailfrom=example.net; dmarc=pass header.from=example.net\r\n"
+RECEIVED = "Received: from mx.example.net by great-service.example; Sun, 18 Oct 2026 20:00:00 +0000\r\n"
+
+
+def _meta(headers):
+    return message_meta_from_rfc822(raw(f"ticket!callback@{D}", "hi", extra=headers),
+                                    trusted_authserv_id=D)
+
+
+def test_authentication_read_from_our_own_header():
+    a = _meta(AR_OURS + RECEIVED)["authentication"]
+    assert a == {"dkim": "pass", "spf": "pass", "dmarc": "pass", "header_from": "example.net"}
+
+
+def test_forged_or_foreign_authentication_is_ignored():
+    forged = RECEIVED + AR_OURS                      # below our Received: brought by the sender
+    foreign = AR_OURS.replace("great-service.example;", "other.example;") + RECEIVED
+    assert _meta(forged)["authentication"] is None
+    assert _meta(foreign)["authentication"] is None
+    assert message_meta_from_rfc822(raw("x@y", "s", extra=AR_OURS))["authentication"] is None  # no trusted id
+
+
+def test_reply_only_to_authenticated_sender(env):
+    svc, send, *_ = env
+    svc.require_dmarc_pass = True
+    sent = []
+    svc.smtp, svc._send = ("127.0.0.1", 25), lambda *a: sent.append(a)
+    send(f"ticket!callback@{D}", "spoofed")                         # no verdict
+    assert sent == [] and svc.outbox() == [] and svc.ticket("GS-41872") is not None
+
+
+def test_authenticated_sender_gets_reply(env):
+    svc, _, _, resolver = env
+    svc.require_dmarc_pass = True
+    sent = []
+    svc.smtp, svc._send = ("127.0.0.1", 25), lambda *a: sent.append(a)
+    event = build_event(f"ticket!callback@{D}", _meta(AR_OURS + RECEIVED), resolver)
+    status, body = svc.handle(event)
+    assert status == 201 and len(sent) == 1 and sent[0][1] == CUSTOMER

workplans/MAS-WP-0005-spec-errata-v0-1-1.md

@@ -393,7 +393,7 @@ never reject on DMARC failure. Milter failure means accept
 
 ```task
 id: MAS-WP-0005-T12
-status: todo
+status: done
 priority: medium
 ```
 
@@ -404,6 +404,23 @@ header_from}` into the event. The ticket service gets
 sends auto-replies (intake confirmation, closed notice) only when
 `dmarc == pass`. Unit tests cover forged A-R headers from other ids.
 
+- 2026-10-18 done:
+  - `mas_gateway.core.authentication_from_headers` reads only A-R headers
+    with the trusted authserv-id **above the first Received header**
+    (RFC 8601 §5). Forged copies below our hop are ignored even if the
+    MTA did not strip them. Several DKIM results count as pass if any one
+    passes.
+  - The event gets `message.authentication = {dmarc, spf, dkim,
+    header_from}` or null. `--trusted-authserv-id` works on both pipe and
+    webhook; a JSON webhook passes the provider's verdict through.
+  - Ticket service `--require-dmarc-pass`: tickets are still created and
+    mail attached, but replies go out only on `dmarc == pass`.
+  - Entrypoint: gateway `--trusted-authserv-id $DOMAIN`, ticket service
+    `--require-dmarc-pass`.
+  - Tests: own header read; forged (below Received) and foreign ids
+    ignored; no trusted id gives no verdict; reply gated off without a
+    verdict and sent on pass. 183 tests pass.
+
 ## Live check: authenticated vs forged sender
 
 ```task