"""Python 3.10+, standard library. Local receiver; production needs HTTPS:443. An operator configures the webhook. Persist the SQLite inbox on durable storage. An application worker must apply business changes transactionally/idempotently from this inbox. HTTP 200 acknowledges durable receipt, not a business action. """ import hashlib import hmac from contextlib import closing from http.server import BaseHTTPRequestHandler, HTTPServer import json import math import os import re import sqlite3 import time MAX_BODY = 65_536 EVENT_STATUSES = {"accepted", "delayed", "delivered", "bounced", "complained", "rejected", "failed", "ambiguous"} def verify_webhook(raw_body, headers, secret, *, now=None, tolerance_seconds=300): if not isinstance(raw_body, bytes) or len(raw_body) > MAX_BODY or not isinstance(secret, str) or not secret: raise ValueError("Invalid webhook input") normalized = {} for name, value in headers.items(): key = name.lower() if not key.startswith("webhook-"): continue if key in normalized or (isinstance(value, (list, tuple)) and len(value) != 1): raise ValueError("Duplicate header") text = value[0] if isinstance(value, (list, tuple)) else value if not isinstance(text, str): raise ValueError("Invalid header") normalized[key] = text event_id = normalized.get("webhook-id", "") timestamp = normalized.get("webhook-timestamp", "") signature = normalized.get("webhook-signature", "") if not re.fullmatch(r"evt_[a-f0-9]{24}", event_id) or not re.fullmatch(r"[0-9]{1,13}", timestamp) or not re.fullmatch(r"v1,[a-f0-9]{64}", signature): raise ValueError("Invalid signed headers") current = time.time() if now is None else now if not math.isfinite(current) or not math.isfinite(tolerance_seconds) or tolerance_seconds < 0 or abs(current - int(timestamp)) > tolerance_seconds: raise ValueError("Timestamp outside tolerance") # Exact raw request bytes. The literal secret is UTF-8, not base64-decoded. signed = (event_id + "." + timestamp + ".").encode("ascii") + raw_body expected = hmac.new(secret.encode("utf-8"), signed, hashlib.sha256).hexdigest() if not hmac.compare_digest(expected, signature[3:]): raise ValueError("Invalid signature") event = json.loads(raw_body.decode("utf-8")) if not isinstance(event, dict) or event.get("id") != event_id or not isinstance(event.get("createdAt"), str): raise ValueError("Invalid event") data = event.get("data") if not isinstance(data, dict) or not isinstance(data.get("emailId"), str) or not re.fullmatch(r"eml_[a-f0-9]{24}", data["emailId"]): raise ValueError("Invalid email ID") status = data.get("status") if not isinstance(status, str) or status not in EVENT_STATUSES or event.get("type") != "email." + status or "detail" not in data or not (data["detail"] is None or isinstance(data["detail"], str)): raise ValueError("Invalid event data") return event def main(): secret = os.environ.get("MAILCORE_WEBHOOK_SECRET", "") if not secret: raise SystemExit("Set MAILCORE_WEBHOOK_SECRET; never log it") inbox_path = os.environ.get("MAILCORE_WEBHOOK_INBOX", "mailcore-webhook-inbox.sqlite3") with closing(sqlite3.connect(inbox_path)) as connection, connection: connection.execute("PRAGMA journal_mode=WAL") connection.execute("PRAGMA synchronous=FULL") connection.execute("CREATE TABLE IF NOT EXISTS inbox (id TEXT PRIMARY KEY, raw_body BLOB NOT NULL, received_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP)") class Receiver(BaseHTTPRequestHandler): # Quiet by default: never log bodies, tokens or webhook secrets. def log_message(self, format, *args): pass def respond(self, status, error=None): body = json.dumps({"ok": not bool(error), **({"error": error} if error else {})}).encode() self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Cache-Control", "no-store") self.send_header("Content-Length", str(len(body))) self.send_header("Connection", "close") if status == 405: self.send_header("Allow", "POST") self.end_headers() if self.command != "HEAD": self.wfile.write(body) self.close_connection = True def do_POST(self): self.connection.settimeout(10) if self.path != "/mailcore-webhook": self.respond(404, "not_found") return if self.headers.get("Content-Type", "").split(";")[0].strip().lower() != "application/json": self.respond(415, "content_type_required") return # This small demo does not parse chunked transfer coding. A reverse # proxy must buffer/dechunk and supply one Content-Length header. lengths = self.headers.get_all("Content-Length", []) if self.headers.get("Transfer-Encoding") or len(lengths) != 1 or not re.fullmatch(r"[0-9]{1,10}", lengths[0]): self.respond(400, "content_length_required") return size = int(lengths[0]) if size > MAX_BODY: self.respond(413, "body_too_large") return try: raw_body = self.rfile.read(size) if len(raw_body) != size: self.respond(400, "request_invalid") return signed_headers = {name: self.headers.get_all(name, []) for name in ("Webhook-Id", "Webhook-Timestamp", "Webhook-Signature")} event = verify_webhook(raw_body, signed_headers, secret) except (ValueError, UnicodeError, OSError, TypeError): self.respond(401, "webhook_invalid") return try: with closing(sqlite3.connect(inbox_path)) as connection, connection: connection.execute("PRAGMA synchronous=FULL") connection.execute("INSERT OR IGNORE INTO inbox (id, raw_body) VALUES (?, ?)", (event["id"], raw_body)) saved = connection.execute("SELECT raw_body FROM inbox WHERE id = ?", (event["id"],)).fetchone()[0] if saved != raw_body: self.respond(409, "event_id_conflict") return # Transaction has committed. A repeat now safely acknowledges # the persisted ID; process the inbox separately and monitor it. self.respond(200) except sqlite3.Error: self.respond(500, "inbox_unavailable") def do_GET(self): self.respond(405, "method_not_allowed") do_PUT = do_DELETE = do_PATCH = do_OPTIONS = do_HEAD = do_TRACE = do_GET port = int(os.environ.get("PORT", "8080")) HTTPServer(("127.0.0.1", port), Receiver).serve_forever() if __name__ == "__main__": main()