348 lines
17 KiB
Python
348 lines
17 KiB
Python
"""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, container_booking_id=None, delivery_id=None,
|
|
fee_source_id=None, source_version=None):
|
|
self.event_type = event_type
|
|
self.selector_id = event_id
|
|
self.plan_id = loading_plan_id
|
|
self.booking_id = container_booking_id
|
|
self.delivery_id = delivery_id
|
|
self.fee_source_id = fee_source_id
|
|
self.source_version = source_version
|
|
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")
|
|
if self.selector_id:
|
|
selected = event_id == self.selector_id
|
|
elif self.booking_id:
|
|
selected = isinstance(payload, dict) and str(payload.get("sourceContainerBookingId")) == self.booking_id
|
|
elif self.delivery_id:
|
|
selected = isinstance(payload, dict) and str(payload.get("deliveryOrderId")) == self.delivery_id
|
|
elif self.fee_source_id:
|
|
selected = (isinstance(payload, dict) and str(payload.get("sourceFeeId")) == self.fee_source_id
|
|
and str(payload.get("sourceVersion")) == self.source_version and payload.get("action") == "VOIDED")
|
|
else:
|
|
selected = isinstance(payload, dict) and str(payload.get("loadingPlanId")) == self.plan_id
|
|
qualifies = event.get("eventType") == self.event_type and selected
|
|
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, ""))
|
|
# Both consumers share FINANCE_SERVICE_URL. Preserve wallet delivery while
|
|
# targeting one fee event; never select wallet bodies for fault injection.
|
|
wallet_endpoint = "/internal/finance/v1/wallet-command-events"
|
|
wallet_passthrough = (state.fee_source_id is not None
|
|
and endpoint == "/internal/finance/v1/business-fee-source-events")
|
|
# Firstleg 的同步只读客户端与装柜事件共用 LASTLEG_SERVICE_URL。
|
|
loading_reads = state.event_type == "FirstLegLoadingPlanCompleted" and endpoint == "/internal/lastleg/events"
|
|
workflow_endpoint = "/internal/lastleg/v1/order-workflow-facts"
|
|
|
|
class DataHandler(QuietHandler):
|
|
def do_POST(self):
|
|
passthrough = (wallet_passthrough and self.path == wallet_endpoint
|
|
or loading_reads and self.path == workflow_endpoint)
|
|
if self.path != endpoint and not passthrough:
|
|
self.safe_error(404, "UNEXPECTED_EVENT_ENDPOINT")
|
|
return
|
|
body = self.read_body()
|
|
if body is None:
|
|
return
|
|
identity, matched, error_code, attempt, mode = (None, False, None, 0, "PASS") if passthrough else state.select(body)
|
|
if error_code:
|
|
state.record(identity, matched, attempt, None, error_code)
|
|
self.safe_error(409, error_code)
|
|
return
|
|
destination = urllib.parse.urlunsplit((parts.scheme, parts.netloc, self.path, "", "")) if passthrough else upstream
|
|
self.forward(body, identity, matched, attempt, mode, destination, "POST")
|
|
|
|
def do_GET(self):
|
|
if not loading_reads or not re.fullmatch(r"/internal/lastleg/appointment-work/containers/[1-9][0-9]*/route", self.path):
|
|
self.safe_error(404, "UNEXPECTED_EVENT_ENDPOINT")
|
|
return
|
|
destination = urllib.parse.urlunsplit((parts.scheme, parts.netloc, self.path, "", ""))
|
|
self.forward(None, None, False, 0, "PASS", destination, "GET")
|
|
|
|
def forward(self, body, identity, matched, attempt, mode, destination, method):
|
|
# 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(destination, data=body, headers=headers, method=method)
|
|
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")
|
|
selector.add_argument("--container-booking-id")
|
|
selector.add_argument("--delivery-id")
|
|
selector.add_argument("--fee-source-id")
|
|
parser.add_argument("--source-version")
|
|
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 args.container_booking_id and not re.fullmatch(r"[1-9][0-9]{0,19}", args.container_booking_id):
|
|
parser.error("container-booking-id must be a positive decimal identifier")
|
|
if args.delivery_id and not re.fullmatch(r"[1-9][0-9]{0,19}", args.delivery_id):
|
|
parser.error("delivery-id must be a positive decimal identifier")
|
|
if bool(args.fee_source_id) != bool(args.source_version):
|
|
parser.error("fee-source-id and source-version are required together")
|
|
if args.fee_source_id and (not re.fullmatch(r"[1-9][0-9]{0,19}", args.fee_source_id)
|
|
or not re.fullmatch(r"[1-9][0-9]{0,9}", args.source_version)
|
|
or int(args.source_version) > 2147483647
|
|
or args.event_type != "BusinessFeeSourceChanged"):
|
|
parser.error("fee void requires valid source identity, version and BusinessFeeSourceChanged")
|
|
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, args.container_booking_id, args.delivery_id,
|
|
args.fee_source_id, args.source_version)
|
|
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()
|