From 9e4a838ba1028099c7f18f807020ee276b52d5e2 Mon Sep 17 00:00:00 2001 From: ouyangxq <775153912@qq.com> Date: Mon, 5 Oct 2026 02:45:27 +0800 Subject: [PATCH] test(inbound): add scoped one-shot resource confirmation fault proxy --- scripts/README.md | 2 + scripts/inbound-plan-resource-fault-proxy.py | 113 ++++++++++++++++++ ...inbound-plan-resource-fault-proxy.test.mjs | 65 ++++++++++ 3 files changed, 180 insertions(+) create mode 100644 scripts/inbound-plan-resource-fault-proxy.py create mode 100644 scripts/inbound-plan-resource-fault-proxy.test.mjs diff --git a/scripts/README.md b/scripts/README.md index 5de33449a..fa86a9f49 100644 --- a/scripts/README.md +++ b/scripts/README.md @@ -1,5 +1,7 @@ # 后端脚本管理与全量 HTTP 验收索引 +`inbound-plan-resource-fault-proxy.py --config <权限0600的私有配置>` 是柜级入仓计划资源确认的 TEST 定向故障工具。导入不执行，运行才监听业务端口与回环控制端口；复用 `order-submit-dependency-fault-proxy.py` 的 HTTP 传输和服务，配置必须显式 `environment=TEST`，冻结柜ID、海外仓ID、按当前协调器JSON字符串算法计算的柜批次前缀、Lastleg来源IP和内部身份。默认直通；持有私有控制凭据的 `CONFIRM_ONCE` 最长60秒，只让该柜精确 `/internal/v1/warehouse-pallet-batches/confirm` POST 返回一次503，随后自动直通。其他柜、预留/查询、错身份及不匹配请求保留原请求响应。凭据只读私有配置，脱敏证据写同目录 `proxy-evidence.jsonl`，不保存正文或令牌。工具不修改SQL或Docker，不自行重试业务或恢复服务路由；运行者须先安排独占故障窗口、记录并恢复原Lastleg配置/镜像及健康、停止本代理，结果未知先核对，不盲目重启。`inbound-plan-resource-fault-proxy.test.mjs` 为离线范围、单次并发、TTL与脱敏测试；工具检查通过不等于已执行真实故障与页面恢复。 + `partial-loading-actual-test-env.mjs` 是 F21/F22 两柜实装与部分装柜的 TEST 专项入口，默认无参数或 `--plan` 零网络。使用显式授权 TEST 环境文件，匹配 Firstleg/触发器就绪并开放独占样本窗口后，才可传 `--execute --environment=TEST --run-id=PARTIAL-ACTUAL-<新批次> --fixture=NEW`，或用 `--fixture=` 提供本轮已核对来源。新准备只走客户资格、下单、测量、装柜正式接口；A 覆盖混合客户/税制、短拣和跳过，B 的全部实装行均为 PARTIAL、退税来源全部跳过。检查实际数量/指标、报关范围及同命令重放；合成图片只证明 HTTP，不是现场证据。报告保存至 `reports/partial-loading-actual-PARTIAL-ACTUAL-<批次>/`，先记录写入意图，拒绝覆盖；失败保留原样本和报告，未知结果不得自动重造或重试。仅在明确续跑授权和原快照核对后接续，退出自己的会话；不执行 SQL/部署、船节点或历史回填，不代替事件 ACK/下游库存、完整报关文件和页面/设备验收。直接测试 `partial-loading-actual-test-env.test.mjs`。 `fba-remaining-nodes-acceptance-transfer.mjs` 是 F33 原已有库存转运的内部验收助手，导入不执行。`freezeTransfer` 使用显式 TEST 配置建立独立会话，只读原票和来源库存后退出；`prepareOriginalTransferPlan` / `prepareTransferStage` 只准备本票真实关联换标的逐阶段命令，不能预设关联号。`runPreparedTransferStage` 默认零网络，只有总控明确 TEST、原票、客户、阶段和稳定批次并注入已核身份的单请求流 transport 后才执行；照片以原字节摘要校验，客户确认使用本票 MERCHANT 身份，不调用被禁止的 PDA 独立复核批准。报告与写入 journal 在 `reports/fba-remaining-nodes-20261004/transfer/`，写前持久化意图；只有已明确成功的原键原载荷可显式重放一次，上传不可重放，失败或结果未知先回查，不能新建替代票、解冻其他业务或重新收货。调用者须逐阶段独立核对库存、其他 HOLD、追加流水和主体归属，HTTP 成功不自动标记业务通过；退出自身会话，不执行 Firstleg 写入、SQL、部署或删除。直接测试 `fba-remaining-nodes-acceptance-transfer.test.mjs`。 diff --git a/scripts/inbound-plan-resource-fault-proxy.py b/scripts/inbound-plan-resource-fault-proxy.py new file mode 100644 index 000000000..90c731020 --- /dev/null +++ b/scripts/inbound-plan-resource-fault-proxy.py @@ -0,0 +1,113 @@ +"""TEST-only one-shot resource-confirm fault for one frozen container batch. + +No Docker, SQL or business mutation is performed by this module. A separately +reviewed runtime must install/restore routing and own the private configuration. +The shared HTTP server/transport provides bounded bodies and a loopback control +listener; only this Router decides which request may receive the synthetic 503. +""" +import argparse +import hashlib +import hmac +import importlib.util +import json +import re +import threading +import time +from pathlib import Path + +_spec = importlib.util.spec_from_file_location( + 'dependency_proxy_http', Path(__file__).with_name('order-submit-dependency-fault-proxy.py')) +_http = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(_http) + +CONFIRM_PATH = '/internal/v1/warehouse-pallet-batches/confirm' + + +class Router: + def __init__(self, config, forward=_http.transport, clock=time.monotonic, record=None): + self.config, self.forward, self.clock = config, forward, clock + self.record = record or (lambda row: None) + self.lock = threading.Lock() + self.mode, self.deadline, self.generation = 'PASS', 0, 0 + self.counts = {'forwarded': 0, 'faulted': 0} + if config.get('environment') != 'TEST': + raise ValueError('EXPLICIT_TEST_REQUIRED') + for key in ('containerBookingId', 'warehouseId'): + if not re.fullmatch(r'[1-9][0-9]{0,19}', str(config.get(key, ''))): + raise ValueError('EXACT_CONTAINER_AND_WAREHOUSE_REQUIRED') + if not config.get('lastlegIps') or not config.get('internalToken') or not config.get('controlToken'): + raise ValueError('AUTHENTICATED_CALLER_SCOPE_REQUIRED') + if not re.fullmatch(r'[a-f0-9]{64}', config.get('batchPrefixHash', '')): + raise ValueError('REVIEWED_BATCH_PREFIX_REQUIRED') + self.prefix = 'CIP:' + config['batchPrefixHash'] + ':' + # Coordinator hashes ObjectMapper.writeValueAsString(String bookingId), including JSON quotes. + expected = hashlib.sha256(json.dumps(str(config['containerBookingId']), separators=(',', ':')).encode()).hexdigest() + if expected != config['batchPrefixHash']: + raise ValueError('CONTAINER_BATCH_PREFIX_MISMATCH') + + def status(self): + with self.lock: + self._expire() + return {'mode': self.mode, 'generation': self.generation, + 'remainingSeconds': max(0, self.deadline - self.clock()), **self.counts} + + def _expire(self): + if self.mode != 'PASS' and self.clock() >= self.deadline: + self.mode, self.deadline = 'PASS', 0 + + def set_mode(self, token, mode, ttl=0): + if not hmac.compare_digest(str(token), self.config['controlToken']): + raise PermissionError('CONTROL_AUTH_REQUIRED') + with self.lock: + self.mode, self.deadline = 'PASS', 0 + if mode not in ('PASS', 'CONFIRM_ONCE') or (mode != 'PASS' and + (type(ttl) is not int or not 1 <= ttl <= 60)): + raise ValueError('INVALID_MODE_OR_TTL') + self.mode = mode + self.deadline = self.clock() + ttl if mode != 'PASS' else 0 + self.generation += 1 + return self.status() + + def dispatch(self, method, path, headers, body, peer): + normalized = {k.lower(): v for k, v in headers.items()} + token = normalized.get('x-internal-service-token', '') + authenticated = bool(token) and hmac.compare_digest(token, self.config['internalToken']) + payload = None + if method == 'POST' and path == CONFIRM_PATH: + try: + payload = json.loads(body) + except (ValueError, UnicodeDecodeError): + pass + selected = (isinstance(payload, dict) and authenticated and peer in self.config['lastlegIps'] + and str(payload.get('warehouseId')) == str(self.config['warehouseId']) + and isinstance(payload.get('batchKey'), str) + and re.fullmatch(re.escape(self.prefix) + r'[A-Za-z0-9_-]{1,80}', payload['batchKey']) is not None + and isinstance(payload.get('commandKey'), str) and bool(payload['commandKey'])) + with self.lock: + self._expire() + fault = selected and self.mode == 'CONFIRM_ONCE' + generation = self.generation + if fault: + self.mode, self.deadline = 'PASS', 0 + self.counts['faulted'] += 1 + if fault: + result = (503, {'Content-Type': 'text/plain'}, b'TEST selected resource confirmation unavailable') + else: + result = self.forward(self.config['upstream'], method, path, headers, body) + with self.lock: + self.counts['forwarded'] += 1 + if selected: + self.record({'method': method, 'path': path, 'matched': fault, 'forwarded': not fault, + 'responseStatus': result[0], 'generation': generation, + 'bodySha256': hashlib.sha256(body).hexdigest(), + 'containerBookingId': self.config['containerBookingId'], + 'warehouseId': self.config['warehouseId'], 'internalTokenValidated': True}) + return result + + +if __name__ == '__main__': + parser = argparse.ArgumentParser() + parser.add_argument('--config', required=True) + # Reuse only the existing transport/server, not its Order routing policy. + _http.Router = Router + _http.main(parser.parse_args().config) diff --git a/scripts/inbound-plan-resource-fault-proxy.test.mjs b/scripts/inbound-plan-resource-fault-proxy.test.mjs new file mode 100644 index 000000000..0cdc20804 --- /dev/null +++ b/scripts/inbound-plan-resource-fault-proxy.test.mjs @@ -0,0 +1,65 @@ +import test from 'node:test' +import assert from 'node:assert/strict' +import { spawnSync } from 'node:child_process' +import { fileURLToPath } from 'node:url' + +test('one container resource-confirm fault is scoped, one-shot, expiring and transparent (offline)', () => { + const source=String.raw` +import hashlib,importlib.util,json,sys,unittest,concurrent.futures +spec=importlib.util.spec_from_file_location('fault',sys.argv[1]);p=importlib.util.module_from_spec(spec);spec.loader.exec_module(p) +class Tests(unittest.TestCase): + def setUp(self): + self.now=[10.0];self.calls=[];self.logs=[] + self.config={'environment':'TEST','containerBookingId':'7','warehouseId':'30','batchPrefixHash':hashlib.sha256(b'"7"').hexdigest(), + 'controlToken':'private-control','internalToken':'private-token','lastlegIps':['172.20.0.10'],'upstream':'http://warehouse.invalid:9305'} + def forward(*args):self.calls.append(args);return 200,{'X-Upstream':'original'},b'original-response' + self.router=p.Router(self.config,forward=forward,clock=lambda:self.now[0],record=self.logs.append) + self.body={'warehouseId':30,'batchKey':self.router.prefix+'generation-1','commandKey':'original-command'} + self.headers={'X-Internal-Service-Token':'private-token'} + def send(self,body=None,path=p.CONFIRM_PATH,method='POST',headers=None,peer='172.20.0.10'): + raw=json.dumps(self.body if body is None else body).encode() + return self.router.dispatch(method,path,self.headers if headers is None else headers,raw,peer) + def test_default_pass_and_original_payload_preserved(self): + self.assertEqual(self.send(),(200,{'X-Upstream':'original'},b'original-response')) + self.assertEqual(json.loads(self.calls[0][4]),self.body) + self.assertEqual(self.calls[0][3],self.headers) + def test_only_exact_confirm_body_caller_and_identity_can_consume_fault(self): + self.router.set_mode('private-control','CONFIRM_ONCE',60) + for kwargs in [dict(path=p.CONFIRM_PATH+'?other=1'),dict(path='/internal/v1/warehouse-pallet-batches/ensure-reserved'), + dict(method='GET'),dict(body={**self.body,'warehouseId':31}),dict(body={**self.body,'batchKey':'another-container'}), + dict(body=[]),dict(body={**self.body,'commandKey':''}),dict(peer='172.20.0.11'),dict(headers={}), + dict(headers={'X-Internal-Service-Token':'wrong'})]:self.assertEqual(self.send(**kwargs)[0],200) + self.assertEqual(self.router.status()['mode'],'CONFIRM_ONCE') + self.assertEqual(self.send()[0],503);self.assertEqual(self.send()[0],200) + self.assertEqual(self.router.status()['faulted'],1) + def test_one_shot_is_atomic_under_concurrent_requests(self): + self.router.set_mode('private-control','CONFIRM_ONCE',60) + with concurrent.futures.ThreadPoolExecutor(max_workers=4) as e:codes=list(e.map(lambda _:self.send()[0],range(8))) + self.assertEqual(codes.count(503),1);self.assertEqual(codes.count(200),7) + def test_ttl_expiry_and_malformed_control_disarm(self): + self.router.set_mode('private-control','CONFIRM_ONCE',2);self.now[0]=12 + self.assertEqual(self.send()[0],200) + for mode,ttl in [('OTHER',1),('CONFIRM_ONCE',0),('CONFIRM_ONCE',61),('CONFIRM_ONCE',True)]: + with self.assertRaises(ValueError):self.router.set_mode('private-control',mode,ttl) + self.assertEqual(self.router.status()['mode'],'PASS') + with self.assertRaises(PermissionError):self.router.set_mode('wrong','CONFIRM_ONCE',1) + def test_production_or_mismatched_container_scope_rejected(self): + for override in [{'environment':'PROD'},{'containerBookingId':'8'},{'warehouseId':'0'},{'lastlegIps':[]}]: + with self.assertRaises(ValueError):p.Router({**self.config,**override}) + def test_malformed_json_is_forwarded_unchanged_and_never_consumes_fault(self): + self.router.set_mode('private-control','CONFIRM_ONCE',60) + raw=b'invalid-json' + self.assertEqual(self.router.dispatch('POST',p.CONFIRM_PATH,self.headers,raw,'172.20.0.10')[0],200) + self.assertEqual(self.calls[0][4],raw);self.assertEqual(self.router.status()['mode'],'CONFIRM_ONCE') + def test_evidence_redacts_credentials_and_raw_body(self): + self.router.set_mode('private-control','CONFIRM_ONCE',60);self.send() + evidence=json.dumps(self.logs) + for secret in ['private-control','private-token','original-command','generation-1']:self.assertNotIn(secret,evidence) + self.assertTrue(self.logs[0]['internalTokenValidated']);self.assertFalse(self.logs[0]['forwarded']) +result=unittest.TextTestRunner().run(unittest.defaultTestLoader.loadTestsFromTestCase(Tests)) +sys.exit(0 if result.wasSuccessful() else 1) +` + const r=spawnSync(process.env.PYTHON||'python',['-c',source,fileURLToPath(new URL('./inbound-plan-resource-fault-proxy.py',import.meta.url))], + {encoding:'utf8',env:{...process.env,PYTHONDONTWRITEBYTECODE:'1'}}) + assert.equal(r.status,0,r.stderr||r.error?.message) +})