mail-address-semantics/experiments/postfix-probe/capture.py

"""Probe-only proxy: records each MAS event as delivered by Postfix, then
forwards it to the ticket service and records the app's answer."""

import json
import sys
import urllib.error
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

UPSTREAM = sys.argv[1]
LOG = sys.argv[2] if len(sys.argv) > 2 else None    # optional JSONL evidence file
BIND = sys.argv[3] if len(sys.argv) > 3 else "0.0.0.0"
EVENTS = []


class H(BaseHTTPRequestHandler):
    def do_POST(self):
        body = self.rfile.read(int(self.headers.get("Content-Length", "0")))
        req = urllib.request.Request(UPSTREAM, body, {"Content-Type": "application/json"})
        try:
            with urllib.request.urlopen(req) as r:
                status, resp = r.status, r.read()
        except urllib.error.HTTPError as e:
            status, resp = e.code, e.read()
        rec = {"event": json.loads(body), "app_status": status, "app": json.loads(resp or b"null")}
        EVENTS.append(rec)
        if LOG:
            with open(LOG, "a") as f:
                f.write(json.dumps(rec) + "\n")
        self._send(status, resp)

    def do_GET(self):
        self._send(200, json.dumps(EVENTS).encode())

    def _send(self, status, body):
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *a):
        pass


ThreadingHTTPServer((BIND, 8024), H).serve_forever()