From d7981f7575bbdab9d9fc56596b33e1242216408d Mon Sep 17 00:00:00 2001 From: ouyangxq <775153912@qq.com> Date: Mon, 5 Oct 2026 03:48:29 +0800 Subject: [PATCH] test(inbound): add recoverable scoped resource fault runtime --- scripts/README.md | 2 + .../inbound-plan-resource-fault-executor.mjs | 77 ++++++ ...ound-plan-resource-fault-executor.test.mjs | 54 ++++ .../inbound-plan-resource-fault-runtime.py | 238 ++++++++++++++++++ ...bound-plan-resource-fault-runtime.test.mjs | 57 +++++ 5 files changed, 428 insertions(+) create mode 100644 scripts/inbound-plan-resource-fault-executor.mjs create mode 100644 scripts/inbound-plan-resource-fault-executor.test.mjs create mode 100644 scripts/inbound-plan-resource-fault-runtime.py create mode 100644 scripts/inbound-plan-resource-fault-runtime.test.mjs diff --git a/scripts/README.md b/scripts/README.md index edf5c2393..d651a99ce 100644 --- a/scripts/README.md +++ b/scripts/README.md @@ -1,5 +1,7 @@ # 后端脚本管理与全量 HTTP 验收索引 +`inbound-plan-resource-fault-executor.mjs` 是单柜入仓计划资源确认故障的 TEST 环境执行入口，内部使用 `inbound-plan-resource-fault-runtime.py`。默认 `plan --plan <单柜计划.json>` 零网络；实际动作必须读取明确的TEST/Jenkins外部配置并传 `--confirm-test-runtime yes`。`preflight --routing <私有路由.json>` 只读校验Lastleg与Warehouse Overseas镜像/指纹、实际网络和Nacos common/lastleg配置摘要；`prepare`另要求 `--expected-image <已冻结镜像ID>` 及预检摘要。`activate`只临时改Lastleg的 `baiyue.warehouse.base-url`，配合 `mode --mode CONFIRM_ONCE --ttl 60` 精确注入一次confirm失败；业务转换和关闭重开页面由调用者独立操作，工具没有造数、SQL或业务写接口。`status/health/records`用于观测，结果未知先 `status/reconcile`，禁止因观察超时重复创建容器；`restore/verify-restored/finish`须按原镜像、完整配置/挂载/网络、Nacos及上游健康恢复后才停自有代理、清私有凭据。服务变更先记录意图，受限远端恢复状态目录为 `/tmp/chongchong-inbound-resource-`，由总控持独占Lastleg故障窗口；调用结果和业务证据另存本次独立 `reports/` 目录，失败保留、不覆盖，不与部署或其他共享业务写入并行。`inbound-plan-resource-fault-executor.test.mjs`、`inbound-plan-resource-fault-runtime.test.mjs` 为离线边界及恢复测试，不证明真实故障已跑通。 + F33 分阶段助手的组合换标顺序为 `unit → unit-carton → bind`：只有完整逐箱作业事实后才可绑定实物托盘，不用默认数量补齐缺失事实。已明确拒绝的绑定仅允许当前已冻结原票、原批次、原命令摘要的 `F33_BIND_A0409_AFTER_UNIT` 恢复，必须同时核对不可变失败journal、完整回滚证明和新作业事实/原资源版本，恢复写入独立journal，不覆盖原失败。未知结果、其他拒绝和其他阶段仍禁止自动重试；响应message脱敏留存。该受限恢复只用于原TEST验收，不放宽生产接口或变更库存规则。 `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与脱敏测试；工具检查通过不等于已执行真实故障与页面恢复。 diff --git a/scripts/inbound-plan-resource-fault-executor.mjs b/scripts/inbound-plan-resource-fault-executor.mjs new file mode 100644 index 000000000..e98556768 --- /dev/null +++ b/scripts/inbound-plan-resource-fault-executor.mjs @@ -0,0 +1,77 @@ +// Explicit one-container TEST runtime actions. Imports and default plan are offline. +import { readFileSync } from 'node:fs'; +import { createHash } from 'node:crypto'; +import { resolve } from 'node:path'; +import { pathToFileURL } from 'node:url'; +const need = (ok, message) => { if (!ok) throw Error(message); }; +const encode = value => Buffer.from(value).toString('base64'); +const source = name => readFileSync(new URL(name, import.meta.url)); +const runtime = source('./inbound-plan-resource-fault-runtime.py'); +const baseSource = encode(source('./order-submit-dependency-fault-runtime.py')); +const commonSource = encode(source('./test-event-fault-runtime.py')); +const proxySource = encode(source('./inbound-plan-resource-fault-proxy.py')); +const httpProxySource = encode(source('./order-submit-dependency-fault-proxy.py')); +export const actions = ['plan','preflight','prepare','activate','status','health','mode','records','restore','reconcile','verify-restored','finish']; +export const batchPrefixHash = booking => createHash('sha256').update(JSON.stringify(booking)).digest('hex'); +export function scopeFor(plan) { + need(plan?.environment === 'TEST' && plan.runtimeChangesExecuted === false, 'EXPLICIT_UNEXECUTED_TEST_PLAN_REQUIRED'); + need(/^[A-Za-z0-9_-]{1,80}$/.test(plan.marker || ''), 'INVALID_MARKER'); + for(const key of ['containerBookingId','warehouseId'])need(/^[1-9]\d{0,19}$/.test(plan[key] || ''), 'SINGLE_CONTAINER_WAREHOUSE_REQUIRED'); + need(plan.batchPrefixHash === batchPrefixHash(plan.containerBookingId), 'JSON_STRING_BATCH_HASH_REQUIRED'); + need(Object.keys(plan).every(key=>['environment','marker','containerBookingId','warehouseId','batchPrefixHash','runtimeChangesExecuted'].includes(key)), 'PLAN_SCHEMA_INVALID'); + return {environment:'TEST',marker:plan.marker,containerBookingId:plan.containerBookingId,warehouseId:plan.warehouseId, + batchPrefixHash:plan.batchPrefixHash,planSha256:createHash('sha256').update(JSON.stringify(plan)).digest('hex')}; +} +function routingFor(action, extra) { + if(!['preflight','prepare'].includes(action))return; + const r=extra.routing; + need(r?.configurationReviewed===true && Object.keys(r).every(k=>['configurationReviewed','sourceSha256','upstreams','nacos'].includes(k)) + && r.upstreams && Object.keys(r.upstreams).join(',')==='WAREHOUSE' && r.nacos + && Object.keys(r.nacos).sort().join(',')==='baseUrl,configPath,loginPath' + && r.nacos.loginPath==='/nacos/v3/auth/user/login' && r.nacos.configPath==='/nacos/v3/admin/cs/config','REVIEWED_ROUTING_REQUIRED'); + for(const value of [r.nacos.baseUrl,r.upstreams.WAREHOUSE]) { + const url=new URL(value);need(['http:','https:'].includes(url.protocol)&&!url.username&&!url.password&&!url.search&&!url.hash&&['','/'].includes(url.pathname),'ORIGIN_ONLY_NO_CREDENTIALS'); + } + if(action==='prepare')need(/^sha256:[a-f0-9]{64}$/.test(extra.expectedImageId||'')&&/^[a-f0-9]{64}$/.test(r.sourceSha256||''),'FROZEN_IMAGE_AND_NACOS_HASH_REQUIRED'); +} +export function renderInboundResourceAction(action, scope, agent, extra={}) { + need(actions.includes(action)&&action!=='plan','UNKNOWN_RUNTIME_ACTION'); + need(/^[A-Za-z0-9_.-]{1,128}$/.test(agent||''),'INVALID_AGENT'); + need(scope.environment==='TEST'&&/^[A-Za-z0-9_-]{1,80}$/.test(scope.marker||'')&&scope.batchPrefixHash===batchPrefixHash(scope.containerBookingId) + && /^[1-9]\d{0,19}$/.test(scope.containerBookingId||'')&&/^[1-9]\d{0,19}$/.test(scope.warehouseId||'')&&/^[a-f0-9]{64}$/.test(scope.planSha256||'') + && Object.keys(scope).sort().join(',')==='batchPrefixHash,containerBookingId,environment,marker,planSha256,warehouseId','INVALID_SCOPE'); + need(Object.keys(extra).every(k=>['routing','expectedImageId','mode','ttl'].includes(k)), 'RUNTIME_ARGUMENT_SCHEMA_INVALID'); + routingFor(action,extra); + if(action==='mode')need(['PASS','CONFIRM_ONCE'].includes(extra.mode)&&(extra.mode==='PASS'||Number.isInteger(extra.ttl)&&extra.ttl>=1&&extra.ttl<=60),'INVALID_MODE_OR_TTL'); + const config={...extra,action,scope,baseSource,commonSource,...(action==='prepare'?{proxySource,httpProxySource}:{})}; + const literal=text=>'['+encode(text).match(/.{1,4000}/g).map(part=>`'${part}'`).join(',')+"].join('')"; + const groovy=`def p=new ProcessBuilder(['python3','-c',new String(java.util.Base64.decoder.decode(${literal(runtime)}),'UTF-8'),${literal(JSON.stringify(config))}]).redirectErrorStream(true).start();def out=p.inputStream.getText('UTF-8');p.waitFor();print(out)`; + return `def c=jenkins.model.Jenkins.instance.getComputer('${agent}');if(c==null || c.channel==null)throw new RuntimeException('TEST agent unavailable');print(hudson.util.RemotingDiagnostics.executeGroovy(new String(java.util.Base64.decoder.decode(${literal(groovy)}),'UTF-8'),c.channel))`; +} +export function createInboundResourceRuntime(env, scope, fetcher=globalThis.fetch) { + need(env.CHONGCHONG_ENV==='test','EXPLICIT_TEST_ENV_REQUIRED'); + need(env.JENKINS_URL&&env.JENKINS_USER&&env.JENKINS_API_TOKEN&&env.FBA_CORE_JENKINS_AGENT,'EXPLICIT_TEST_JENKINS_REQUIRED'); + return async(action,extra={})=>{ + const script=renderInboundResourceAction(action,scope,env.FBA_CORE_JENKINS_AGENT,extra); + let response; + try {response=await fetcher(new URL('scriptText',env.JENKINS_URL.replace(/\/?$/,'/')),{method:'POST',redirect:'error',signal:AbortSignal.timeout(55000), + headers:{Authorization:'Basic '+encode(env.JENKINS_USER+':'+env.JENKINS_API_TOKEN),'Content-Type':'application/x-www-form-urlencoded'},body:new URLSearchParams({script})});} + catch {throw Error('RUNTIME_'+action.toUpperCase().replaceAll('-','_')+'_OUTCOME_UNKNOWN_USE_STATUS');} + need(response.ok,'RUNTIME_HTTP_'+response.status+'_NO_AUTO_RETRY'); + const text=await response.text();need(!text.includes('MultipleCompilationErrorsException'),'RENDERER_COMPILATION_FAILED_NO_ACTION'); + const rows=text.split(/\r?\n/).filter(row=>row.startsWith('INBOUND_RESOURCE_JSON='));need(rows.length===1,'AMBIGUOUS_RESPONSE_NO_AUTO_RETRY'); + let result;try {result=JSON.parse(rows[0].slice('INBOUND_RESOURCE_JSON='.length));}catch{throw Error('INVALID_RESPONSE_NO_AUTO_RETRY');} + need(result?.ok===true,/^[A-Z_0-9]{1,90}$/.test(result?.error||'')?result.error:'RUNTIME_ACTION_FAILED');return result.data; + }; +} +async function main(argv) { + const [action='plan',...rest]=argv,args={}; + for(let i=0;i{console.error(/^[A-Z_0-9]{1,90}$/.test(error.message)?error.message:'INBOUND_RESOURCE_ARGUMENT_ERROR');process.exitCode=1;}); diff --git a/scripts/inbound-plan-resource-fault-executor.test.mjs b/scripts/inbound-plan-resource-fault-executor.test.mjs new file mode 100644 index 000000000..f0cb85ce8 --- /dev/null +++ b/scripts/inbound-plan-resource-fault-executor.test.mjs @@ -0,0 +1,54 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import {spawnSync} from 'node:child_process'; +import {mkdtempSync,writeFileSync} from 'node:fs'; +import {tmpdir} from 'node:os'; +import {join} from 'node:path'; +import {fileURLToPath} from 'node:url'; +import {scopeFor,batchPrefixHash,renderInboundResourceAction,createInboundResourceRuntime,actions} from './inbound-plan-resource-fault-executor.mjs'; +const plan={environment:'TEST',marker:'F27-B-TEST',containerBookingId:'2106714159440728066',warehouseId:'193000000000200003',runtimeChangesExecuted:false}; +plan.batchPrefixHash=batchPrefixHash(plan.containerBookingId); +const scope=scopeFor(plan); +const env={CHONGCHONG_ENV:'test',JENKINS_URL:'https://offline.invalid/',JENKINS_USER:'offline',JENKINS_API_TOKEN:'secret-do-not-render',FBA_CORE_JENKINS_AGENT:'test-agent'}; +const routing={configurationReviewed:true,nacos:{baseUrl:'http://offline.invalid:8848',loginPath:'/nacos/v3/auth/user/login',configPath:'/nacos/v3/admin/cs/config'},upstreams:{WAREHOUSE:'http://offline.invalid:9305'},sourceSha256:'a'.repeat(64)}; +test('coordinator hashes the JSON string including quotes, never raw booking bytes',()=>{ + assert.equal(plan.batchPrefixHash,'4c9034968e117813e549a12f5e63f1c8366a7f2dbb6722a3ff4967b983a5f9e7'); +}); +test('scope rejects non-TEST, executed, multiple-container, unknown and bad hash plans',()=>{ + for(const change of [p=>p.environment='PROD',p=>p.runtimeChangesExecuted=true,p=>p.containerBookingId=['1','2'],p=>p.extra='secret',p=>p.batchPrefixHash='b'.repeat(64)]){ + const value=structuredClone(plan);change(value);assert.throws(()=>scopeFor(value)); + } +}); +const arrays=text=>Array.from(text.matchAll(/\[('[A-Za-z0-9+/=]+'(?:,'[A-Za-z0-9+/=]+')*)\]\.join\(''\)/g),m=>Array.from(m[1].matchAll(/'([^']+)'/g),p=>p[1])); +test('renderer chunks complete payload and carries both proxy modules and reviewed primitives',()=>{ + const rendered=renderInboundResourceAction('prepare',scope,'test-agent',{routing,expectedImageId:'sha256:'+'c'.repeat(64)}); + assert.ok(!rendered.includes(env.JENKINS_API_TOKEN)); + const groovy=Buffer.from(arrays(rendered)[0].join(''),'base64').toString();const embedded=arrays(groovy); + assert.ok(embedded.flat().every(s=>s.length<=4000)); + const config=JSON.parse(Buffer.from(embedded[1].join(''),'base64').toString()); + assert.deepEqual(config.scope,scope);assert.equal(config.action,'prepare'); + for(const key of ['proxySource','httpProxySource','commonSource','baseSource'])assert.ok(config[key].length>100); + const runtime=Buffer.from(embedded[0].join(''),'base64').toString();assert.ok(runtime.includes('firstleg-test-lastleg')); +}); +test('origin/schema guards reject credential embedding, wrong imports and unbounded fault windows',()=>{ + for(const change of [r=>r.password='secret',r=>r.upstreams.IAM='http://offline.invalid',r=>r.nacos.baseUrl='http://user:secret@offline.invalid',r=>r.nacos.configPath='/fallback']){ + const r=structuredClone(routing);change(r);assert.throws(()=>renderInboundResourceAction('prepare',scope,'test-agent',{routing:r,expectedImageId:'sha256:'+'c'.repeat(64)})); + } + for(const ttl of [0,61,1.5])assert.throws(()=>renderInboundResourceAction('mode',scope,'agent',{mode:'CONFIRM_ONCE',ttl})); + assert.throws(()=>renderInboundResourceAction('mode',scope,'agent',{mode:'FAIL_ALL',ttl:1})); + assert.throws(()=>renderInboundResourceAction('status',{...scope,warehouseId:'0'},'agent')); + assert.throws(()=>renderInboundResourceAction('business-convert',scope,'agent')); +}); +test('transport unknown, duplicate result and malformed result each fail after exactly one call',async()=>{ + for(const fetcher of [async()=>{throw Error('lost secret')},async()=>({ok:true,text:async()=> 'INBOUND_RESOURCE_JSON={}\nINBOUND_RESOURCE_JSON={}'}),async()=>({ok:true,text:async()=> 'INBOUND_RESOURCE_JSON={'})]){ + let calls=0;const runtime=createInboundResourceRuntime(env,scope,async(...args)=>{calls++;return fetcher(...args)}); + await assert.rejects(runtime('activate'),/UNKNOWN|AMBIGUOUS|INVALID_RESPONSE/);assert.equal(calls,1); + } + assert.throws(()=>createInboundResourceRuntime({...env,CHONGCHONG_ENV:'prod'},scope)); +}); +test('default plan CLI has zero HTTP and exposes no business conversion action',()=>{ + const dir=mkdtempSync(join(tmpdir(),'inbound-fault-plan-'));const file=join(dir,'plan.json');writeFileSync(file,JSON.stringify(plan)); + const run=spawnSync(process.execPath,[fileURLToPath(new URL('./inbound-plan-resource-fault-executor.mjs',import.meta.url)),'plan','--plan',file],{encoding:'utf8',env:{...process.env,CHONGCHONG_ENV:'prod',JENKINS_URL:'http://127.0.0.1:1'}}); + assert.equal(run.status,0,run.stderr);const output=JSON.parse(run.stdout);assert.equal(output.networkCalls,0);assert.equal(output.businessWritesSupported,false); + assert.equal(output.target,'firstleg-test-lastleg');assert.equal(output.property,'baiyue.warehouse.base-url');assert.deepEqual(output.actions,actions); +}); diff --git a/scripts/inbound-plan-resource-fault-runtime.py b/scripts/inbound-plan-resource-fault-runtime.py new file mode 100644 index 000000000..9f1473d2e --- /dev/null +++ b/scripts/inbound-plan-resource-fault-runtime.py @@ -0,0 +1,238 @@ +"""Explicit TEST/single-container runtime. Import performs no network or mutation. + +Reuses the reviewed Order Docker journal/reconstruction/reconciliation methods; +only Lastleg's baiyue.warehouse.base-url is temporarily changed. No business/SQL. +""" +import ast +import base64 +import copy +import hashlib +import json +import os +import re +import secrets +import signal +import socket +import subprocess +import sys +import time +import urllib.parse +import urllib.request +from pathlib import Path + +TARGET = 'firstleg-test-lastleg' +UPSTREAM = 'firstleg-test-warehouse-overseas' +PROPERTY = 'baiyue.warehouse.base-url' + + +def validate_scope(scope): + if set(scope) != {'environment', 'marker', 'containerBookingId', 'warehouseId', 'batchPrefixHash', 'planSha256'}: + raise RuntimeError('INVALID_SCOPE') + if scope['environment'] != 'TEST' or not re.fullmatch(r'[A-Za-z0-9_-]{1,80}', scope['marker']): + raise RuntimeError('EXPLICIT_TEST_SCOPE_REQUIRED') + if any(not re.fullmatch(r'[1-9][0-9]{0,19}', scope[key]) for key in ('containerBookingId', 'warehouseId')): + raise RuntimeError('SINGLE_CONTAINER_WAREHOUSE_REQUIRED') + expected = hashlib.sha256(json.dumps(scope['containerBookingId'], separators=(',', ':')).encode()).hexdigest() + if scope['batchPrefixHash'] != expected or not re.fullmatch(r'[a-f0-9]{64}', scope['planSha256']): + raise RuntimeError('BATCH_OR_PLAN_HASH_MISMATCH') + + +def runtime_class(base_source): + # No __main__ executes. Only the known config import's dataId is adapted. + tree = ast.parse(base_source) + class Imports(ast.NodeTransformer): + def visit_Constant(self, node): + if node.value == 'order-service.yml': return ast.copy_location(ast.Constant('lastleg-service.yml'), node) + return node + tree = ast.fix_missing_locations(Imports().visit(tree)) + namespace = {'__name__': 'reviewed_resource_docker_runtime'} + exec(compile(tree, '', 'exec'), namespace) + namespace.update(ORDER=TARGET, PROPERTIES={'WAREHOUSE': PROPERTY}, validate_scope=validate_scope) + Parent = namespace['Runtime'] + + class Runtime(Parent): + def __init__(self, scope, tools, base=Path('/tmp')): + validate_scope(scope) + self.scope, self.tools = scope, tools + self.directory = base / ('chongchong-inbound-resource-' + scope['marker']) + self.path = self.directory / 'state.json' + self.lock = base / 'chongchong-inbound-resource.lock' + + def probes(self, state): + target = self.verified_upstream(self.container(), state['routing'], state['internalToken']) + if (target['Id'] != state['upstreamId'] or target['Image'] != state['upstreamImageId'] + or self.tools['fingerprint'](target) != state['upstreamFingerprint']): + raise RuntimeError('WAREHOUSE_UPSTREAM_CONFIGURATION_CHANGED') + request = urllib.request.Request(state['upstreams']['WAREHOUSE'] + '/actuator/health', + headers={'X-Internal-Service-Token': state['internalToken']}) + with namespace['http_open'](request, timeout=8) as response: + value = json.load(response) + if response.status != 200 or value.get('status') != 'UP': + raise RuntimeError('WAREHOUSE_UPSTREAM_HEALTH_FAILED') + return {'WAREHOUSE': True} + + def verified_upstream(self, original, routing, token): + target = self.tools['inspect'](UPSTREAM); self.tools['require_test'](target, UPSTREAM) + target_env = self.tools['env_of'](target) + address = urllib.parse.urlsplit(routing['upstreams']['WAREHOUSE']) + ips = {n['IPAddress'] for name, n in target['NetworkSettings']['Networks'].items() + if name in original['NetworkSettings']['Networks']} + if (address.scheme != 'http' or address.hostname not in ips or str(address.port) != target_env.get('SERVER_PORT') + or address.username or address.password or address.path not in ('', '/') or address.query or address.fragment + or target_env.get('DEPLOY_ENV') != 'test' or target['State'].get('Health', {}).get('Status') != 'healthy' + or len(token) < 16 or target_env.get('INTERNAL_SERVICE_TOKEN') != token): + raise RuntimeError('UPSTREAM_NOT_VERIFIED_TEST_CONTAINER') + return target + + def prepare(self, config): + if self.directory.exists(): raise RuntimeError('STATE_EXISTS_USE_STATUS_OR_RESTORE') + original = self.container() + if original['Image'] != config.get('expectedImageId') or not re.fullmatch(r'sha256:[a-f0-9]{64}', original['Image']): + raise RuntimeError('ORIGINAL_IMAGE_MISMATCH') + if original['State'].get('Health', {}).get('Status') != 'healthy': raise RuntimeError('LASTLEG_NOT_HEALTHY') + if original['Config']['Entrypoint'] != namespace['ENTRYPOINT'] or original['Config'].get('Cmd') not in (None, []): + raise RuntimeError('UNREVIEWED_JAVA_ENTRYPOINT') + if 'host.docker.internal:host-gateway' not in (original['HostConfig'].get('ExtraHosts') or []): + raise RuntimeError('REVIEWED_PROXY_HOST_ROUTE_REQUIRED') + env = self.tools['env_of'](original) + token = env.get('INTERNAL_SERVICE_TOKEN', '') + if len(token) < 16: raise RuntimeError('INTERNAL_TOKEN_UNAVAILABLE') + routing = config['routing'] + if routing.get('configurationReviewed') is not True or set(routing['upstreams']) != {'WAREHOUSE'}: + raise RuntimeError('REVIEWED_WAREHOUSE_ROUTING_REQUIRED') + source = self.config_fingerprint(original, routing) + if source != routing.get('sourceSha256'): raise RuntimeError('REVIEWED_NACOS_SOURCE_DIGEST_CHANGED') + target = self.verified_upstream(original, routing, token) + fd = os.open(self.lock, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, 'w') as stream: stream.write(self.scope['marker']); stream.flush(); os.fsync(stream.fileno()) + self.directory.mkdir(mode=0o700) + sockets = [socket.socket(), socket.socket()] + for sock in sockets: sock.bind(('127.0.0.1', 0)) + proxy_port, control_port = [sock.getsockname()[1] for sock in sockets] + for sock in sockets: sock.close() + state = {'scope': self.scope, 'phase': 'PREPARED', 'originalId': original['Id'], 'originalImageId': original['Image'], + 'originalFingerprint': self.tools['fingerprint'](original), 'originalBody': self.tools['recreation'](original, env, full=True), + 'internalToken': token, 'controlToken': secrets.token_urlsafe(32), 'proxyPort': proxy_port, 'controlPort': control_port, + 'upstreams': {'WAREHOUSE': routing['upstreams']['WAREHOUSE'].rstrip('/')}, 'routing': routing, + 'routingSourceSha256': source, 'lastlegIps': [n['IPAddress'] for n in original['NetworkSettings']['Networks'].values()], + 'upstreamId': target['Id'], 'upstreamImageId': target['Image'], 'upstreamFingerprint': self.tools['fingerprint'](target), + 'mutation': None, 'proxyLaunchIntent': False} + self.save(state) + self.probes(state) + for name, field in [('proxy.py', 'proxySource'), ('order-submit-dependency-fault-proxy.py', 'httpProxySource')]: + file = self.directory / name; file.write_bytes(base64.b64decode(config[field])); os.chmod(file, 0o600) + namespace['private_write'](self.directory / 'proxy-config.json', { + key: state[key] for key in ('internalToken', 'controlToken', 'proxyPort', 'controlPort', 'lastlegIps')} | { + 'environment': 'TEST', 'containerBookingId': self.scope['containerBookingId'], 'warehouseId': self.scope['warehouseId'], + 'batchPrefixHash': self.scope['batchPrefixHash'], 'upstream': state['upstreams']['WAREHOUSE']}) + state['phase'] = 'PROXY_START_INTENT'; state['proxyLaunchIntent'] = True; self.save(state) + with open(self.directory / 'proxy-output.log', 'ab') as output: + os.chmod(self.directory / 'proxy-output.log', 0o600) + process = subprocess.Popen(['python3', str(self.directory / 'proxy.py'), '--config', str(self.directory / 'proxy-config.json')], + stdout=output, stderr=output, start_new_session=True) + state['proxyPid'] = process.pid; state['phase'] = 'PROXY_STARTED'; self.save(state) + return self.summary(state) + + def injected_entrypoint(self, state): + value = copy.deepcopy(state['originalBody']['Entrypoint']) + value[2] += ' --' + PROPERTY + '=http://host.docker.internal:' + str(state['proxyPort']) + return value + + def matches_injected(self, current, state): + view = copy.deepcopy(current); view['Config']['Entrypoint'] = state['originalBody']['Entrypoint'] + return (current['Image'] == state['originalImageId'] and current['Config']['Entrypoint'] == self.injected_entrypoint(state) + and self.tools['fingerprint'](view) == state['originalFingerprint']) + + def activate(self, state): + if state['phase'] != 'PROXY_STARTED' or state.get('uncertain'): raise RuntimeError('ACTIVATE_NOT_RETRYABLE') + if self.control(state)['mode'] != 'PASS': raise RuntimeError('PROXY_NOT_PASS') + current = self.container() + if current['Id'] != state['originalId'] or self.tools['fingerprint'](current) != state['originalFingerprint']: + raise RuntimeError('ORIGINAL_CONFIGURATION_CHANGED') + desired = copy.deepcopy(state['originalBody']); desired['Entrypoint'] = self.injected_entrypoint(state) + self.replace(state, current, desired, 'INJECTED') + state['injectedFingerprint'] = self.tools['fingerprint'](self.container()); self.save(state) + return self.summary(state) + + def finish(self, state, _business=None): + self.verify_restored(state) # Includes healthy, full config/mount/network/image + Nacos + real upstream. + if self.lock.is_symlink() or self.lock.read_text() != self.scope['marker']: raise RuntimeError('RUNTIME_LOCK_OWNER_CHANGED') + pid = state.get('proxyPid') + if not pid and state.get('proxyLaunchIntent'): + file = self.directory / 'proxy.pid' + if file.is_symlink() or not file.exists() or not file.read_text().isdigit(): raise RuntimeError('PROXY_PID_UNCERTAIN') + pid = int(file.read_text()) + command = Path('/proc/' + str(pid) + '/cmdline') if pid else None + if command and command.exists(): + if str(self.directory / 'proxy.py').encode() not in command.read_bytes().split(b'\0'): + raise RuntimeError('PROXY_PROCESS_OWNERSHIP_MISMATCH') + os.kill(pid, signal.SIGTERM) + for _ in range(40): + if not command.exists() or not command.read_bytes(): break + time.sleep(.1) + else: raise RuntimeError('PROXY_NOT_STOPPED') + state['restored'] = True; state['phase'] = 'FINISHED' + for field in ('originalBody', 'desiredBody', 'internalToken', 'controlToken'): state.pop(field, None) + self.save(state) + file = self.directory / 'proxy-config.json' + if file.exists(): file.unlink() + self.lock.unlink() + return {'restored': True, 'healthy': True, 'proxyStopped': True, 'lastlegRoutesVerified': True, + 'originalImageId': state['originalImageId'], 'originalFingerprint': state['originalFingerprint']} + return Runtime, namespace + + +def execute(config): + Runtime, namespace = runtime_class(base64.b64decode(config['baseSource']).decode()) + runtime = Runtime(config['scope'], namespace['common_tools'](base64.b64decode(config['commonSource']).decode())) + action = config['action'] + if action == 'preflight': + current = runtime.container() + target = runtime.verified_upstream(current, config['routing'], runtime.tools['env_of'](current).get('INTERNAL_SERVICE_TOKEN', '')) + return {'originalImageId': current['Image'], 'originalFingerprint': runtime.tools['fingerprint'](current), + 'sourceSha256': runtime.config_fingerprint(current, config['routing']), 'runtimeChanged': False, + 'upstreamImageId': target['Image'], 'upstreamFingerprint': runtime.tools['fingerprint'](target)} + if action == 'prepare': return runtime.prepare(config) + state = runtime.load() + if action == 'status': + result = runtime.summary(state) + if not state.get('restored'): + try: result['proxy'] = runtime.control(state) + except Exception: result['proxy'] = {'unavailable': True} + return result + if state.get('restored'): raise RuntimeError('ALREADY_FINISHED_USE_STATUS') + if runtime.lock.is_symlink() or runtime.lock.read_text() != runtime.scope['marker']: raise RuntimeError('RUNTIME_LOCK_OWNER_CHANGED') + if action == 'records': + file = runtime.directory / 'proxy-evidence.jsonl' + return {'records': [json.loads(line) for line in file.read_text().splitlines()] if file.exists() else []} + if action == 'mode': + if state['phase'] != 'INJECTED' or state.get('uncertain'): raise RuntimeError('FAULT_WINDOW_NOT_READY') + mode, ttl = config.get('mode'), config.get('ttl', 0) + if mode not in ('PASS', 'CONFIRM_ONCE') or (mode != 'PASS' and (type(ttl) is not int or not 1 <= ttl <= 60)): + raise RuntimeError('INVALID_MODE_OR_TTL') + return runtime.control(state, mode, ttl) + if action == 'health': return {'healthy': runtime.container()['State'].get('Health', {}).get('Status') == 'healthy'} + actions = {'activate': runtime.activate, 'restore': runtime.restore, 'reconcile': runtime.reconcile, + 'verify-restored': runtime.verify_restored, 'finish': runtime.finish} + if action not in actions: raise RuntimeError('UNKNOWN_ACTION') + return actions[action](state) + + +def main(config): + import fcntl + validate_scope(config['scope']) + file = Path('/tmp') / ('chongchong-inbound-resource-' + config['scope']['marker'] + '.action.lock') + if file.is_symlink(): raise RuntimeError('UNSAFE_ACTION_LOCK') + fd = os.open(file, os.O_RDWR | os.O_CREAT, 0o600) + try: + try: fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: raise RuntimeError('RUNTIME_ACTION_IN_PROGRESS_USE_STATUS_LATER') + return execute(config) + finally: os.close(fd) + + +if __name__ == '__main__': + try: print('INBOUND_RESOURCE_JSON=' + json.dumps({'ok': True, 'data': main(json.loads(base64.b64decode(sys.argv[1])))})) + except Exception as error: + message = str(error) + print('INBOUND_RESOURCE_JSON=' + json.dumps({'ok': False, 'error': message if re.fullmatch(r'[A-Z_0-9]{1,90}', message) else 'INBOUND_RESOURCE_ACTION_FAILED'})) diff --git a/scripts/inbound-plan-resource-fault-runtime.test.mjs b/scripts/inbound-plan-resource-fault-runtime.test.mjs new file mode 100644 index 000000000..4e215d9e2 --- /dev/null +++ b/scripts/inbound-plan-resource-fault-runtime.test.mjs @@ -0,0 +1,57 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import {spawnSync} from 'node:child_process'; +import {fileURLToPath} from 'node:url'; +const script=String.raw` +import importlib.util,sys,json,hashlib,copy +from pathlib import Path +spec=importlib.util.spec_from_file_location('resource_runtime',sys.argv[1]);m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m) +base=Path(sys.argv[1]).with_name('order-submit-dependency-fault-runtime.py').read_text() +C,ns=m.runtime_class(base) +scope={'environment':'TEST','marker':'OFFLINE','containerBookingId':'2106714159440728066','warehouseId':'193000000000200003','batchPrefixHash':hashlib.sha256(json.dumps('2106714159440728066').encode()).hexdigest(),'planSha256':'a'*64} +r=C(scope,{}) +assert ns['ORDER']=='firstleg-test-lastleg' and ns['PROPERTIES']=={'WAREHOUSE':'baiyue.warehouse.base-url'} +assert r.lock.name=='chongchong-inbound-resource.lock' +assert 'lastleg-service.yml' in C.config_fingerprint.__code__.co_consts or any(isinstance(c,tuple) and 'lastleg-service.yml' in c for c in C.config_fingerprint.__code__.co_consts) +state={'originalBody':{'Entrypoint':['sh','-c','exec java $JAVA_OPTS org.springframework.boot.loader.launch.JarLauncher']},'proxyPort':4321,'originalImageId':'sha256:'+'c'*64,'originalFingerprint':'f'} +entry=r.injected_entrypoint(state) +assert entry[2].endswith(' --baiyue.warehouse.base-url=http://host.docker.internal:4321') +assert '/warehouse' not in entry[2] and state['originalBody']['Entrypoint'][2].endswith('JarLauncher') +saved=[];calls=[] +r.save=lambda s:saved.append(copy.deepcopy(s)) +def lost(*args):calls.append(args);raise RuntimeError('connection lost') +r.tools={'docker':lost};journal={} +try:r.mutation(journal,'STOP_INTENT','POST','/containers/frozen/stop?t=15');raise AssertionError('unknown accepted') +except RuntimeError as e:assert str(e)=='UNCERTAIN_DOCKER_MUTATION' +assert saved[0]['mutation']['step']=='STOP_INTENT' and journal['uncertain'] +try:r.mutation(journal,'DELETE_INTENT','DELETE','/containers/frozen');raise AssertionError('blind retry') +except RuntimeError as e:assert str(e)=='UNCERTAIN_MUTATION_REQUIRES_RECONCILE' +assert len(calls)==1 +original={'NetworkSettings':{'Networks':{'test-net':{'IPAddress':'172.18.0.7'}}}} +warehouse={'Id':'warehouse-frozen','Image':'sha256:'+'e'*64,'State':{'Health':{'Status':'healthy'}},'NetworkSettings':{'Networks':{'test-net':{'IPAddress':'172.18.0.8'}}}} +token='offline-internal-identity' +r.tools={'inspect':lambda name: warehouse if name=='firstleg-test-warehouse-overseas' else (_ for _ in ()).throw(AssertionError('wrong upstream')), + 'require_test':lambda c,name:None,'env_of':lambda c:{'SERVER_PORT':'9305','DEPLOY_ENV':'test','INTERNAL_SERVICE_TOKEN':token}} +routing={'upstreams':{'WAREHOUSE':'http://172.18.0.8:9305'}} +assert r.verified_upstream(original,routing,token)['Id']=='warehouse-frozen' +for url in ['http://172.18.0.9:9305','http://172.18.0.8:9999','http://user:secret@172.18.0.8:9305','http://172.18.0.8:9305/fallback']: + try:r.verified_upstream(original,{'upstreams':{'WAREHOUSE':url}},token);raise AssertionError('wrong route accepted') + except RuntimeError as e:assert str(e)=='UPSTREAM_NOT_VERIFIED_TEST_CONTAINER' +current={'Image':state['originalImageId'],'Config':{'Entrypoint':entry},'State':{'Health':{'Status':'healthy'}}} +r.tools={'fingerprint':lambda c:'f' if c['Config']['Entrypoint']==state['originalBody']['Entrypoint'] else 'wrong'} +assert r.matches_injected(current,state) +changed=copy.deepcopy(current);changed['Config']['Entrypoint'][2]+=' --unrelated=true';assert not r.matches_injected(changed,state) +def refuse(_):raise RuntimeError('RESTORED_NOT_HEALTHY') +r.verify_restored=refuse +try:r.finish(state);raise AssertionError('finished before restored') +except RuntimeError as e:assert str(e)=='RESTORED_NOT_HEALTHY' +for key,value in [('environment','PROD'),('warehouseId','0'),('batchPrefixHash','b'*64)]: + bad=scope|{key:value} + try:m.validate_scope(bad);raise AssertionError('invalid scope accepted') + except RuntimeError:pass +print('OFFLINE_RUNTIME_PASS') +`; +test('reviewed Docker methods retain exact target, one-property overlay, journal uncertainty and restore-before-cleanup',()=>{ + const result=spawnSync(process.env.PYTHON||'python',['-c',script,fileURLToPath(new URL('./inbound-plan-resource-fault-runtime.py',import.meta.url))],{encoding:'utf8',env:{...process.env,PYTHONDONTWRITEBYTECODE:'1'}}); + assert.equal(result.status,0,result.stderr||result.error?.message);assert.ok(result.stdout.includes('OFFLINE_RUNTIME_PASS')); +});