mail-address-semantics/mas_gateway/main.py

"""`mas-gateway` — bridge inbound mail to an HTTP application as MAS events.

Two inputs, one core:

  mas-gateway pipe --recipient R [--recipient R2] --app-url URL < message.eml
      For an MTA pipe transport, e.g. Postfix master.cf:
        mas  unix  -  n  n  -  -  pipe
          flags=Rq user=mas argv=/usr/local/bin/mas-gateway pipe
          --app-url http://127.0.0.1:8025/mas-events --recipient ${recipient}
      Exit 0 when the app accepted or deliberately declined the message (no
      bounce, hence no backscatter); exit 75 (EX_TEMPFAIL) when the app is
      unreachable or failed, so the MTA retries.

  mas-gateway webhook --listen 127.0.0.1:8026 --app-url URL
      Provider-style inbound webhook: POST JSON
        {"recipient" | "recipients", "from", "subject", "text", "message_id", "in_reply_to", ...}
      or a raw message with Content-Type message/rfc822 and ?recipient=... .
"""

from __future__ import annotations

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

EX_TEMPFAIL = 75
MAX_BODY = 25 * 1024 * 1024


def cmd_pipe(args, resolver) -> int:
    raw = sys.stdin.buffer.read()
    meta = message_meta_from_rfc822(raw, args.trusted_authserv_id)
    deliveries = gateway(args.recipient, meta, resolver, args.app_url, _signer(args))
    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,
                    trusted_authserv_id: Optional[str] = None) -> tuple[list[str], dict]:
    if ctype.startswith("message/rfc822"):
        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", ""),
            "authentication": data.get("authentication")}  # as verified by the inbound provider
    return recipients, meta


def make_webhook_server(listen: str, app_url: str, resolver: Resolver,
                        trusted_authserv_id: Optional[str] = None,
                        signer: Optional[tuple[str, bytes]] = None) -> ThreadingHTTPServer:
    host, port = listen.rsplit(":", 1)

    class Handler(BaseHTTPRequestHandler):
        def do_POST(self):
            length = int(self.headers.get("Content-Length", "0"))
            if length > MAX_BODY:
                return self._reply(413, {"error": "message too large"})
            body = self.rfile.read(length)
            try:
                recipients, meta = webhook_message(body, self.headers.get("Content-Type", ""),
                                                   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:
                return self._reply(400, {"error": "no recipient"})
            deliveries = gateway(recipients, meta, resolver, app_url, signer)
            status = 503 if any(d.outcome == "tempfail" for d in deliveries) else 200
            self._reply(status, {"deliveries": [d.__dict__ for d in deliveries]})

        def _reply(self, status, obj):
            out = json.dumps(obj).encode()
            self.send_response(status)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(out)))
            self.end_headers()
            self.wfile.write(out)

        def log_message(self, fmt, *a):
            print(fmt % a, file=sys.stderr)

    return ThreadingHTTPServer((host, int(port)), Handler)


def _signer(args):
    if not args.sign_peer:
        return None
    with open(args.sign_key_file, "rb") as f:
        return args.sign_peer, f.read().strip()


def main(argv=None) -> int:
    ap = argparse.ArgumentParser(prog="mas-gateway", description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    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")
    ap.add_argument("--sign-peer", metavar="ID",
                    help="sign events as this peer (mode contract section 7); needed when the "
                         "application is not on loopback")
    ap.add_argument("--sign-key-file", metavar="FILE", help="key shared with the application")
    sub = ap.add_subparsers(dest="cmd", required=True)
    p = sub.add_parser("pipe")
    p.add_argument("--recipient", action="append", required=True)
    p.add_argument("--app-url", required=True)
    w = sub.add_parser("webhook")
    w.add_argument("--listen", default="127.0.0.1:8026")
    w.add_argument("--app-url", required=True)
    args = ap.parse_args(argv)
    if bool(args.sign_peer) != bool(args.sign_key_file):
        ap.error("--sign-peer and --sign-key-file go together")
    registry = json.load(open(args.registry)) if args.registry else None
    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, args.trusted_authserv_id,
                              _signer(args))
    print(f"mas-gateway webhook on http://{args.listen} -> {args.app_url}", file=sys.stderr)
    srv.serve_forever()
    return 0


if __name__ == "__main__":
    sys.exit(main())