"""Explicitly selected test-event ACK loss. Does not create or replay events. Run with an explicit consumer URL, event type and exactly one event selector. EVENT_FAULT_CONTROL_TOKEN protects the separate loopback control listener. Default PASS; authenticated POST /mode {"mode":"FAULT"} enables ACK loss. The parent operator owns producer configuration, DEAD/retry and restoration. """ import argparse import hashlib import hmac import json import os import re import signal import threading import urllib.error import urllib.parse import urllib.request from datetime import datetime, timezone from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer MAX_BODY = 8 * 1024 * 1024 EVENT_ID = re.compile(r"(?:[0-9a-fA-F]{8}(?:-[0-9a-fA-F]{4}){3}-[0-9a-fA-F]{12}|[0-9A-HJKMNP-TV-Z]{26})\Z") HASH = re.compile(r"sha256:[0-9a-fA-F]{64}\Z") HOP_HEADERS = {"connection", "keep-alive", "proxy-authenticate", "proxy-authorization", "te", "trailer", "transfer-encoding", "upgrade", "host", "content-length"} def encoded(value): return json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode("utf-8") class State: def __init__(self, event_type, event_id, loading_plan_id, log_path=None): self.event_type = event_type self.selector_id = event_id self.plan_id = loading_plan_id self.log_path = log_path self.lock = threading.Lock() self.mode = "PASS" self.event_id = None self.payload_hash = None self.envelope_hash = None self.attempt = 0 self.records = [] def select(self, body): try: event = json.loads(body) except (ValueError, UnicodeError): return None, False, None, 0, self.mode if not isinstance(event, dict): return None, False, None, 0, self.mode event_id, payload_hash = event.get("eventId"), event.get("payloadHash") # Only valid bounded identifiers may enter evidence, never arbitrary payload strings. if not isinstance(event_id, str) or not EVENT_ID.fullmatch(event_id): return None, False, None, 0, self.mode if not isinstance(payload_hash, str) or not HASH.fullmatch(payload_hash): with self.lock: if self.event_id == event_id: self.attempt += 1 return {"eventId": event_id, "payloadHash": self.payload_hash}, True, \ "TARGET_EVENT_HASH_CONFLICT", self.attempt, self.mode return None, False, None, 0, self.mode identity = {"eventId": event_id, "payloadHash": payload_hash.lower()} payload = event.get("payload") qualifies = event.get("eventType") == self.event_type and ( event_id == self.selector_id if self.selector_id else isinstance(payload, dict) and str(payload.get("loadingPlanId")) == self.plan_id) envelope_hash = hashlib.sha256(json.dumps(event, sort_keys=True, ensure_ascii=False, separators=(",", ":")).encode("utf-8")).hexdigest() with self.lock: # Reusing the pinned ID with another hash/content/type is rejected even if a # changed selector field would otherwise make the request look unrelated. if self.event_id == event_id and (self.payload_hash != payload_hash.lower() or self.envelope_hash != envelope_hash): self.attempt += 1 return identity, True, "TARGET_EVENT_HASH_CONFLICT", self.attempt, self.mode if not qualifies or (self.event_id is not None and self.event_id != event_id): return identity, False, None, 0, self.mode if self.event_id is None: self.event_id, self.payload_hash, self.envelope_hash = event_id, payload_hash.lower(), envelope_hash self.attempt += 1 return identity, True, None, self.attempt, self.mode def record(self, identity, matched, attempt, outcome, error_code): record = {"time": datetime.now(timezone.utc).isoformat(), "eventId": identity.get("eventId") if identity else None, "payloadHash": identity.get("payloadHash") if identity else None, "outcome": outcome, "attempt": attempt, "matched": matched, "errorCode": error_code} with self.lock: # Store only target evidence; unrelated delivery payloads/identities are not retained. if matched: self.records.append(record) if self.log_path: with open(self.log_path, "a", encoding="utf-8") as log: log.write(json.dumps(record, separators=(",", ":")) + "\n") def snapshot(self): with self.lock: return {"mode": self.mode, "eventId": self.event_id, "payloadHash": self.payload_hash, "attempt": self.attempt, "records": list(self.records)} class QuietHandler(BaseHTTPRequestHandler): def log_message(self, _format, *_args): pass def reply(self, status, body, content_type="application/json", headers=None): self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(body))) for name, value in (headers or {}).items(): if name.lower() not in HOP_HEADERS | {"content-type"}: self.send_header(name, value) self.end_headers() self.wfile.write(body) def safe_error(self, status, code): self.reply(status, encoded({"errorCode": code})) def read_body(self): try: length = int(self.headers.get("Content-Length", "-1")) except ValueError: length = -1 if not 0 <= length <= MAX_BODY: self.safe_error(413, "INVALID_BODY_LENGTH") return None body = self.rfile.read(length) if len(body) != length: self.safe_error(400, "INCOMPLETE_BODY") return None return body def data_handler(state, upstream, timeout): parts = urllib.parse.urlsplit(upstream) endpoint = urllib.parse.urlunsplit(("", "", parts.path or "/", parts.query, "")) class DataHandler(QuietHandler): def do_POST(self): if self.path != endpoint: self.safe_error(404, "UNEXPECTED_EVENT_ENDPOINT") return body = self.read_body() if body is None: return identity, matched, error_code, attempt, mode = state.select(body) if error_code: state.record(identity, matched, attempt, None, error_code) self.safe_error(409, error_code) return # Forward the exact body and end-to-end headers. Never persist auth headers. connection_headers = set(self.headers.get("Connection", "").lower().replace(" ", "").split(",")) headers = {name: value for name, value in self.headers.items() if name.lower() not in HOP_HEADERS | connection_headers} request = urllib.request.Request(upstream, data=body, headers=headers, method="POST") try: # Redirects are forbidden: a redirect could expose a service identity token. class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, *_args, **_kwargs): return None opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirect()) try: consumer = opener.open(request, timeout=timeout) except urllib.error.HTTPError as error: consumer = error with consumer: status = consumer.code result = consumer.read(MAX_BODY + 1) result_headers = dict(consumer.headers.items()) except (OSError, urllib.error.URLError): state.record(identity, matched, attempt, None, "UPSTREAM_TRANSPORT_ERROR") self.safe_error(502, "UPSTREAM_TRANSPORT_ERROR") return if len(result) > MAX_BODY: state.record(identity, matched, attempt, None, "UPSTREAM_BODY_TOO_LARGE") self.safe_error(502, "UPSTREAM_BODY_TOO_LARGE") return outcome = None if matched: try: wrapper = json.loads(result) ack = wrapper.get("data") valid = status == 200 and str(wrapper.get("code")) == "200" and isinstance(ack, dict) \ and ack.get("eventId") == identity["eventId"] \ and str(ack.get("payloadHash", "")).lower() == identity["payloadHash"] \ and ack.get("outcome") in {"APPLIED", "DUPLICATE"} if valid: outcome = ack["outcome"] except (ValueError, UnicodeError, AttributeError): valid = False if not valid: state.record(identity, matched, attempt, None, "UPSTREAM_ACK_INVALID") self.safe_error(502, "UPSTREAM_ACK_INVALID") return if mode == "FAULT": state.record(identity, matched, attempt, outcome, "ACK_LOSS_INJECTED") self.safe_error(503, "ACK_LOSS_INJECTED") return state.record(identity, matched, attempt, outcome, None) self.reply(status, result, result_headers.get("Content-Type", "application/json"), result_headers) return DataHandler def control_handler(state, control_token): class ControlHandler(QuietHandler): def authorized(self): received = self.headers.get("X-Event-Fault-Control-Token", "") if not hmac.compare_digest(received.encode(), control_token.encode()): self.safe_error(403, "CONTROL_FORBIDDEN") return False return True def do_GET(self): if not self.authorized(): return if self.path != "/status": self.safe_error(404, "UNKNOWN_CONTROL_ENDPOINT") return self.reply(200, encoded(state.snapshot())) def do_POST(self): if not self.authorized(): return if self.path != "/mode": self.safe_error(404, "UNKNOWN_CONTROL_ENDPOINT") return body = self.read_body() if body is None: return try: value = json.loads(body) mode = value["mode"] if mode not in {"PASS", "FAULT"}: raise ValueError() except (ValueError, KeyError, TypeError, UnicodeError): self.safe_error(400, "INVALID_MODE") return with state.lock: state.mode = mode self.reply(200, encoded(state.snapshot())) return ControlHandler def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--upstream", required=True) parser.add_argument("--event-type", required=True) selector = parser.add_mutually_exclusive_group(required=True) selector.add_argument("--event-id") selector.add_argument("--loading-plan-id") parser.add_argument("--bind", default="127.0.0.1") parser.add_argument("--port", type=int, default=0) parser.add_argument("--control-port", type=int, default=0) parser.add_argument("--timeout", type=float, default=20) parser.add_argument("--log") args = parser.parse_args() parts = urllib.parse.urlsplit(args.upstream) token = os.environ.get("EVENT_FAULT_CONTROL_TOKEN", "") if parts.scheme not in {"http", "https"} or not parts.hostname or parts.username or parts.password or parts.fragment: parser.error("upstream must be an explicit HTTP consumer endpoint without credentials or fragment") if len(token) < 24: parser.error("EVENT_FAULT_CONTROL_TOKEN must have at least 24 characters") if args.event_id and not EVENT_ID.fullmatch(args.event_id): parser.error("event-id must be a UUID or ULID") if args.loading_plan_id and not re.fullmatch(r"[1-9][0-9]{0,19}", args.loading_plan_id): parser.error("loading-plan-id must be a positive decimal identifier") if not re.fullmatch(r"[A-Za-z][A-Za-z0-9]{0,79}", args.event_type) or not 0 < args.timeout <= 120: parser.error("invalid event type or timeout") state = State(args.event_type, args.event_id, args.loading_plan_id, args.log) data = ThreadingHTTPServer((args.bind, args.port), data_handler(state, args.upstream, args.timeout)) control = ThreadingHTTPServer(("127.0.0.1", args.control_port), control_handler(state, token)) for server in (data, control): server.daemon_threads = True stop = threading.Event() for number in (signal.SIGINT, signal.SIGTERM): signal.signal(number, lambda *_args: stop.set()) threads = [threading.Thread(target=server.serve_forever, daemon=True) for server in (data, control)] for thread in threads: thread.start() print(json.dumps({"ready": True, "dataPort": data.server_port, "controlPort": control.server_port}), flush=True) stop.wait() for server in (data, control): server.shutdown() server.server_close() for thread in threads: thread.join() if __name__ == "__main__": main()