diff --git a/scripts/event-delivery-fault-proxy.py b/scripts/event-delivery-fault-proxy.py index 28d18ce37..b853ee601 100644 --- a/scripts/event-delivery-fault-proxy.py +++ b/scripts/event-delivery-fault-proxy.py @@ -32,12 +32,15 @@ def encoded(value): class State: - def __init__(self, event_type, event_id, loading_plan_id, log_path=None, container_booking_id=None, delivery_id=None): + 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" @@ -73,6 +76,9 @@ class State: 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 @@ -146,16 +152,22 @@ class QuietHandler(BaseHTTPRequestHandler): 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") class DataHandler(QuietHandler): def do_POST(self): - if self.path != endpoint: + passthrough = wallet_passthrough and self.path == wallet_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 = state.select(body) + 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) @@ -164,7 +176,8 @@ def data_handler(state, upstream, timeout): 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") + destination = urllib.parse.urlunsplit((parts.scheme, parts.netloc, wallet_endpoint, "", "")) if passthrough else upstream + request = urllib.request.Request(destination, data=body, headers=headers, method="POST") try: # Redirects are forbidden: a redirect could expose a service identity token. class NoRedirect(urllib.request.HTTPRedirectHandler): @@ -264,6 +277,8 @@ def main(): 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) @@ -284,9 +299,17 @@ def main(): 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) + 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): diff --git a/scripts/event-delivery-fault-proxy.test.mjs b/scripts/event-delivery-fault-proxy.test.mjs index bd149224b..0d3480724 100644 --- a/scripts/event-delivery-fault-proxy.test.mjs +++ b/scripts/event-delivery-fault-proxy.test.mjs @@ -19,7 +19,7 @@ const event = (id = targetId, planId = '2100000000000000001') => ({ eventId: id, payload: { loadingPlanId: planId, syntheticPrivateField: secret } }) const close = server => new Promise(resolve => server.close(resolve)) -async function fixture(selector = ['--loading-plan-id', '2100000000000000001'], eventType = 'LoadingCompleted') { +async function fixture(selector = ['--loading-plan-id', '2100000000000000001'], eventType = 'LoadingCompleted', endpoint = '/internal/events') { const calls = [] const applied = new Set() let invalidAck = false @@ -28,7 +28,7 @@ async function fixture(selector = ['--loading-plan-id', '2100000000000000001'], for await (const chunk of request) chunks.push(chunk) const raw = Buffer.concat(chunks).toString() const envelope = JSON.parse(raw) - calls.push({ raw, authorization: request.headers.authorization, eventId: envelope.eventId }) + calls.push({ raw, authorization: request.headers.authorization, eventId: envelope.eventId, path: request.url }) const outcome = applied.has(envelope.eventId) ? 'DUPLICATE' : 'APPLIED' applied.add(envelope.eventId) const body = JSON.stringify({ code: '200', data: { eventId: invalidAck ? otherId : envelope.eventId, @@ -40,7 +40,7 @@ async function fixture(selector = ['--loading-plan-id', '2100000000000000001'], const directory = await mkdtemp(join(tmpdir(), 'event-fault-offline-')) const log = join(directory, 'evidence.jsonl') const child = spawn(process.env.PYTHON || 'python', [script, '--upstream', - `http://127.0.0.1:${consumer.address().port}/internal/events`, '--event-type', eventType, + `http://127.0.0.1:${consumer.address().port}${endpoint}`, '--event-type', eventType, ...selector, '--log', log], { env: { ...process.env, EVENT_FAULT_CONTROL_TOKEN: controlToken }, stdio: ['ignore', 'pipe', 'pipe'] }) let output = ''; let errors = '' @@ -58,12 +58,12 @@ async function fixture(selector = ['--loading-plan-id', '2100000000000000001'], child.once('error', error => { clearTimeout(timer); reject(error) }) child.once('exit', code => { if (!output) { clearTimeout(timer); reject(new Error(`Proxy exit ${code}`)) } }) }) - const dataUrl = `http://127.0.0.1:${ready.dataPort}/internal/events` + const dataUrl = `http://127.0.0.1:${ready.dataPort}${endpoint}` const controlUrl = `http://127.0.0.1:${ready.controlPort}` return { calls, applied, log, invalidAck: value => { invalidAck = value }, - send: async envelope => { + send: async (envelope, path = endpoint) => { const sent = JSON.stringify(envelope) - const response = await fetch(dataUrl, { method: 'POST', body: sent, + const response = await fetch(new URL(path, dataUrl), { method: 'POST', body: sent, headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${secret}` } }) return { status: response.status, raw: await response.text(), headers: response.headers, sent } }, @@ -85,6 +85,26 @@ async function fixture(selector = ['--loading-plan-id', '2100000000000000001'], } } +test('fee fault preserves wallet endpoint bytes and auth without selecting its body', async t => { + const proxy = await fixture(['--fee-source-id', '2100000000000000001', '--source-version', '1'], + 'BusinessFeeSourceChanged', '/internal/finance/v1/business-fee-source-events') + t.after(() => proxy.stop()) + await proxy.control('/mode', { mode: 'FAULT' }) + const envelope = { ...event(), eventType: 'BusinessFeeSourceChanged', + payload: { sourceFeeId: '2100000000000000001', sourceVersion: 1, action: 'VOIDED' } } + const walletPath = '/internal/finance/v1/wallet-command-events' + const wallet = await proxy.send(envelope, walletPath) + assert.equal(wallet.status, 200) + assert.equal(proxy.calls[0].path, walletPath) + assert.equal(proxy.calls[0].raw, wallet.sent) + assert.equal(proxy.calls[0].authorization, `Bearer ${secret}`) + assert.equal((await (await proxy.control('/status')).json()).attempt, 0) + assert.equal((await proxy.send(envelope, '/internal/finance/v1/unknown')).status, 404) + assert.equal(proxy.calls.length, 1) + assert.equal((await proxy.send(envelope)).status, 503) + assert.equal((await (await proxy.control('/status')).json()).attempt, 1) +}) + test('ACK-loss proxy only targets one event and forwards committed upstream results before dropping ACK', async t => { const proxy = await fixture() t.after(() => proxy.stop()) @@ -260,3 +280,33 @@ test('delivery selection rejects malformed IDs or a second selector before openi assert.ok(!result.stdout.includes('"ready"')) } }) + +test('fee void selector leaves publication, other sources and versions unchanged while recovering exact VOIDED ACK', async t => { + const sourceFeeId='2100000000000000011' + const proxy=await fixture(['--fee-source-id',sourceFeeId,'--source-version','2'],'BusinessFeeSourceChanged') + t.after(()=>proxy.stop()) + await proxy.control('/mode',{mode:'FAULT'}) + const voided={...event(),eventType:'BusinessFeeSourceChanged',payload:{sourceFeeId,sourceVersion:2,action:'VOIDED',totalAmount:30}} + for(const payload of [{...voided.payload,action:'PUBLISHED'},{...voided.payload,sourceVersion:1},{...voided.payload,sourceFeeId:'2100000000000000012'}, + {...voided.payload,action:'VOID'}]) assert.equal((await proxy.send({...voided,eventId:otherId,payload})).status,200) + assert.equal((await proxy.send(voided)).status,503) + assert.equal((await proxy.send(voided)).status,503) + assert.equal((await proxy.send({...voided,eventId:samePlanOtherId})).status,200) + const before=proxy.calls.length + assert.equal((await proxy.send({...voided,payload:{...voided.payload,sourceVersion:3}})).status,409) + assert.equal(proxy.calls.length,before) + await proxy.control('/mode',{mode:'PASS'}) + assert.equal(JSON.parse((await proxy.send(voided)).raw).data.outcome,'DUPLICATE') + const state=await (await proxy.control('/status')).json() + assert.deepEqual(state.records.filter(r=>r.errorCode!=='TARGET_EVENT_HASH_CONFLICT').map(r=>r.outcome),['APPLIED','DUPLICATE','DUPLICATE']) +}) + +test('fee void selector requires a bounded version paired with only its own source selector',()=>{ + for(const args of [['--fee-source-id','11'],['--event-id',targetId,'--source-version','1'],['--fee-source-id','0','--source-version','1'], + ['--fee-source-id','11','--source-version','0'],['--fee-source-id','11','--source-version','2147483648'], + ['--fee-source-id','../11','--source-version','1'],['--fee-source-id','11','--source-version','1','--delivery-id','88']]) { + const result=spawnSync(process.env.PYTHON||'python',[script,'--upstream','http://127.0.0.1:1/internal/events','--event-type','BusinessFeeSourceChanged',...args], + {encoding:'utf8',timeout:5000,env:{...process.env,EVENT_FAULT_CONTROL_TOKEN:controlToken}}) + assert.equal(result.status,2);assert.ok(!result.stdout.includes('"ready"')) + } +})