test(inbound): add scoped one-shot resource confirmation fault proxy

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=<manifest.json>` 提供本轮已核对来源。新准备只走客户资格、下单、测量、装柜正式接口；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`。

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)

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)
+})