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 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()
    deliveries = gateway(args.recipient, message_meta_from_rfc822(raw), 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]:
    if ctype.startswith("message/rfc822"):
        return query.get("recipient", []), message_meta_from_rfc822(body)
    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", "")}
    return recipients, meta


def make_webhook_server(listen: str, app_url: str, resolver: Resolver) -> 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))
            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)
            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 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")
    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)
    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)
    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())