tegwick
e517f2b81a
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 1370019@bnt-lap001 Assistant-Session: d462dd63-16de-4189-ae70-b9cd6566bf87
46 lines
1.5 KiB
Python
46 lines
1.5 KiB
Python
"""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()
|