From bde1cd532ed22a7e9894affb231638a7d0f7744c Mon Sep 17 00:00:00 2001 From: ouyangxq <775153912@qq.com> Date: Sat, 3 Oct 2026 05:25:30 +0800 Subject: [PATCH] test(order): add isolated dependency failure and recovery verification --- scripts/README.md | 2 + ...rs-lifecycle-idempotency-contract.test.mjs | 49 +- scripts/admin-test-env-modules/orders.mjs | 76 ++- ...order-submit-dependency-fault-executor.mjs | 251 ++++++++++ ...-submit-dependency-fault-executor.test.mjs | 209 ++++++++ .../order-submit-dependency-fault-proxy.py | 176 +++++++ ...der-submit-dependency-fault-proxy.test.mjs | 69 +++ .../order-submit-dependency-fault-runtime.py | 445 ++++++++++++++++++ 8 files changed, 1272 insertions(+), 5 deletions(-) create mode 100644 scripts/order-submit-dependency-fault-executor.mjs create mode 100644 scripts/order-submit-dependency-fault-executor.test.mjs create mode 100644 scripts/order-submit-dependency-fault-proxy.py create mode 100644 scripts/order-submit-dependency-fault-proxy.test.mjs create mode 100644 scripts/order-submit-dependency-fault-runtime.py diff --git a/scripts/README.md b/scripts/README.md index b3bb97ad6..f6e535b2f 100644 --- a/scripts/README.md +++ b/scripts/README.md @@ -1,6 +1,8 @@ # 后端脚本管理与全量 HTTP 验收索引 头程订单提交使用首次原始详情的 `expectedVersion` 与原 `Idempotency-Key`。生产单票的 `SUBMIT_INTENT` 必须先持久化 `submitCommand` 再发送，显式续跑不得使用当前版本替换原命令，缺少原命令的提交意图及后续检查点停止并等待人工核对。源 Excel 造数同样在提交前保存订单条目的 `submitCommand`；中断或失败条目保留原检查点且不自动重跑，不重建订单或换版本继续。 + +`order-submit-dependency-fault-executor.mjs` 是显式 TEST 下单依赖故障入口，内部使用 `order-submit-dependency-fault-proxy.py`、`order-submit-dependency-fault-runtime.py`。`plan --plan <文件>` 只读校验；`preflight --plan <文件> --routing <文件> --confirm-test-runtime yes` 核对测试 Order 镜像、实际依赖和 Nacos 配置；`run` 另需 `--expected-image <镜像ID>`、全新 `--report <文件>`。凭据仅从明确指定的 TEST 与 Jenkins 环境文件读取，路由文件不得包含凭据。必须事先准备本轮两个独立合法草稿、原始版本与幂等键及完整快照；仅对本轮客户的 IAM 资格和 Finance 钱包 GET 注入短时故障，验证两端提交失败后事实不变。运行会临时重建同镜像 Order 容器并在恢复后真实提交草稿，因此不得与部署或其他共享环境造数并行；不写业务 SQL。结束先恢复直通、原镜像配置、健康与依赖读取，再停止代理；失败保留报告和受限恢复文件，不自动重跑或删除数据。结果未知时使用 `status`、`reconcile` 核对原操作，必要时显式 `restore --plan <文件> --report <新文件> --confirm-test-runtime yes`，不得因观察超时重复创建容器。离线合同为对应 proxy/executor `.test.mjs`，离线通过不代表真实故障注入已验收。 `event-delivery-fault-proxy.py` 是 TEST 环境定向事件 ACK 丢失工具，不创建或重放事件。运行者显式传入 `--upstream` 完整消费者地址、`--event-type`，并只选 `--event-id`、`--loading-plan-id` 或 `--container-booking-id`；柜次选择精确匹配 payload.sourceContainerBookingId，用于海外卸柜回写，不接受混合选择条件。`EVENT_FAULT_CONTROL_TOKEN` 从外部环境注入，保护固定本机监听的控制端口。默认 PASS；经鉴权切为 FAULT 后，仅在选定事件已真实转发且收到匹配的 APPLIED/DUPLICATE ACK 时向发送方返回 503，其他事件直通。同一选择范围内锁定首个事件后不改选其他事件 ID，原 ID 改摘要或载荷直接拒绝。日志只保留事件标识、摘要和结果，不保留正文或凭据。它会影响实际投递，仅限明确授权的测试环境；生产者配置、重试、DEAD 原事件人工恢复及最终恢复直连由运行者负责，工具退出不自动恢复服务配置。`--log` 可显式指定脱敏证据文件，失败保留原事件与日志，禁止改摘要或借其他事件重跑。`event-delivery-fault-proxy.test.mjs` 使用本机替身服务验证行为，不访问共享环境；工具离线通过不能替代真实重试验收。 `admin-test-env-acceptance.mjs` 与 `client-test-env-acceptance.mjs` 仅接受统一 TEST Gateway 和 Auth 会话；移除环境文件中的 `ADMIN_OVERSEAS_GATEWAY`，不再传 `--overseas-gateway` 或非 test 环境。平台、客户、PDA 身份仍分别登录并验证权限及客户隔离。海外资源接口使用 `/api/platform/v1/overseas-warehouse-resources/**`，仓库主档仍归 Warehouse。跨服务模块的重复 GET 不证明事件重放；真实重试/幂等证据缺失会报告部署阻断，详见模块 README 和下方事件人工重试入口。离线合同不会执行真实 HTTP；环境执行仍需本轮写入授权。 diff --git a/scripts/admin-test-env-modules/orders-lifecycle-idempotency-contract.test.mjs b/scripts/admin-test-env-modules/orders-lifecycle-idempotency-contract.test.mjs index 55210f33a..2de2df749 100644 --- a/scripts/admin-test-env-modules/orders-lifecycle-idempotency-contract.test.mjs +++ b/scripts/admin-test-env-modules/orders-lifecycle-idempotency-contract.test.mjs @@ -2,7 +2,54 @@ import assert from 'node:assert/strict' import { readFileSync } from 'node:fs' import test from 'node:test' -import { fixtureInboundCode, orderPersistenceFacts, requestedOrderDestinationType, freezeOrderCommand, verifyPlatformDraftEdit, verifyPlatformSubmitCancel } from './orders.mjs' +import { fixtureInboundCode, orderPersistenceFacts, requestedOrderDestinationType, freezeOrderCommand, verifyPlatformDraftEdit, verifyPlatformSubmitCancel, + buildOrderDependencyFaultPlan, assertOrderDependencyFaultObservation } from './orders.mjs' + +function dependencyFixture() { + const detail = id => ({order:{id,orderNo:`FL${id}`,tenantId:'20',status:'DRAFT',version:4}, + cargoes:[{skuNo:'current'}],parties:[],internalNotes:[],statusHistory:[]}) + return {marker:'CURRENT-RUN',customerId:'20',registeredCustomerId:'20',merchant:detail('101'),platform:detail('102')} +} +function dependencyObservation(plan, fixture, dependency, role) { + const phase=plan.phases.find(row=>row.dependency===dependency),command=plan.commands[role] + const snapshot={detail:structuredClone(fixture[role]),task:null,inbounds:[],inboundPageComplete:true} + return {request:{method:command.method,path:command.path,body:structuredClone(command.body),headers:{...command.headers,'X-Request-Id':`${dependency}-${role}`}}, + response:{http:200,body:{code:'S0503',message:phase.expectedMessages[0]}},before:snapshot,after:structuredClone(snapshot), + proxyRecords:[{requestId:`${dependency}-${role}`,method:'GET',path:phase.targetPath,matched:true,responseStatus:503,forwarded:false,internalTokenPresent:true}]} +} +test('dependency plan binds two new-customer drafts and only named Order upstream properties; it performs no runtime actions',()=>{ + const fixture=dependencyFixture(),plan=buildOrderDependencyFaultPlan(fixture) + assert.equal(plan.runtimeChangesExecuted,false) + assert.deepEqual(plan.phases.map(row=>row.urlProperty),['baiyue.clients.iam.url','baiyue.clients.finance.url']) + assert.equal(plan.commands.merchant.body.expectedVersion,4) + fixture.merchant.order.version=9 + assert.equal(plan.commands.merchant.body.expectedVersion,4) + assert.ok(Object.isFrozen(plan.commands.merchant.body)) + for(const change of [f=>f.customerId='0',f=>f.registeredCustomerId='21',f=>f.platform.order.id='101',f=>f.merchant.order.status='SUBMITTED',f=>delete f.merchant.cargoes]) { + const value=dependencyFixture();change(value);assert.throws(()=>buildOrderDependencyFaultPlan(value)) + } +}) +test('all four dependency-role observations require matching authenticated 503 injection and actual Order S0503 with zero drift',()=>{ + const fixture=dependencyFixture(),plan=buildOrderDependencyFaultPlan(fixture) + for(const dependency of ['IAM','FINANCE'])for(const role of ['merchant','platform']) { + const evidence=dependencyObservation(plan,fixture,dependency,role) + assert.equal(assertOrderDependencyFaultObservation(plan,dependency,role,evidence).verified,true) + const wrapped=structuredClone(evidence) + wrapped.response.body.message=plan.phases.find(row=>row.dependency===dependency).expectedMessages[1] + assert.equal(assertOrderDependencyFaultObservation(plan,dependency,role,wrapped).verified,true) + for(const mutate of [ + value=>value.response.http=503,value=>value.response.body.code='A0403',value=>value.response.body.message='资格未审核', + value=>value.request.body.expectedVersion=5,value=>value.request.headers['Idempotency-Key']='new-key', + value=>value.proxyRecords=[],value=>value.proxyRecords[0].internalTokenPresent=false,value=>value.proxyRecords[0].forwarded=true, + value=>value.proxyRecords[0].path+='/other',value=>value.proxyRecords[0].requestId='other-request', + value=>value.proxyRecords.push({...value.proxyRecords[0],path:'/internal/iam/tenants/21/order-eligibility'}), + value=>value.after.detail.order.version++,value=>value.after.detail.statusHistory.push({id:'unexpected'}), + value=>value.after.task={orderId:plan.commands[role].orderId,tenantId:'20',id:'unexpected'}, + value=>value.after.inbounds.push({tenantId:'20',orderNo:plan.commands[role].orderNo,id:'unexpected'}), + value=>value.after.inboundPageComplete=false, + ]) { const changed=structuredClone(evidence);mutate(changed);assert.throws(()=>assertOrderDependencyFaultObservation(plan,dependency,role,changed)) } + } +}) function platformLifecycleFixture(fault = '', faultAction = 'submit') { const drafts = new Map(), memo = new Map(), calls = [] diff --git a/scripts/admin-test-env-modules/orders.mjs b/scripts/admin-test-env-modules/orders.mjs index c85d77c77..d43436d7f 100644 --- a/scripts/admin-test-env-modules/orders.mjs +++ b/scripts/admin-test-env-modules/orders.mjs @@ -2,8 +2,6 @@ import { createHash } from 'node:crypto' const MODULE = 'orders' const CONTRACT_OPTIONS = { scenario: 'isolated-order-chain', deploymentBlocker: true } -const MERCHANT_USERNAME = process.env.MERCHANT_TEST_USERNAME || '' -const MERCHANT_PASSWORD = process.env.MERCHANT_TEST_PASSWORD || '' const BOXES = [ { boxType: 'SMALL', lengthCm: 30, widthCm: 20, heightCm: 20 }, { boxType: 'MEDIUM', lengthCm: 40, widthCm: 30, heightCm: 30 }, @@ -243,6 +241,69 @@ export function orderPersistenceFacts(detail) { statusHistory: Array.isArray(detail?.statusHistory) ? detail.statusHistory : [], }) } +// Read-only planning and evidence validation only. These exports never configure services or send requests. +export function buildOrderDependencyFaultPlan(fixture) { + const customerId = id(fixture?.customerId) + if (!/^[1-9]\d*$/.test(customerId) || customerId !== id(fixture?.registeredCustomerId) + || !/^[A-Za-z0-9_-]{1,80}$/.test(fixture?.marker || '')) throw new Error('Dependency fault fixture must belong to this run registered customer') + const commands = {} + for (const role of ['merchant','platform']) { + const detail = fixture[role] + if (id(detail?.order?.tenantId) !== customerId || !/^[1-9]\d*$/.test(id(detail?.order?.id)) + || detail.order.status !== 'DRAFT' || !detail.order.orderNo + || !['cargoes','parties','internalNotes','statusHistory'].every(field => Array.isArray(detail[field]))) { + throw new Error(`Incomplete owned ${role} DRAFT evidence`) + } + const command = freezeOrderCommand(detail, `dep-${role}-${fixture.marker}`) + commands[role] = { method:'POST', path:`/api/${role}/v1/orders/first-leg/${detail.order.id}/submit${role==='platform'?`?tenantId=${customerId}`:''}`, + ...command, orderId:id(detail.order.id), orderNo:detail.order.orderNo, + initialOrderSha256:createHash('sha256').update(orderPersistenceFacts(detail)).digest('hex') } + } + if (commands.merchant.orderId === commands.platform.orderId) throw new Error('Use two independent drafts for same-key recovery') + return { marker:fixture.marker, customerId, runtimeChangesExecuted:false, commands, + phases:[ + { dependency:'IAM', urlProperty:'baiyue.clients.iam.url', targetMethod:'GET', targetPath:`/internal/iam/tenants/${customerId}/order-eligibility`, + faultHttp:503, expectedHttp:200, expectedCode:'S0503', expectedMessages:['IAM eligibility service is unavailable','IAM 下单资格响应无效'] }, + { dependency:'FINANCE', urlProperty:'baiyue.clients.finance.url', targetMethod:'GET', targetPath:`/internal/finance/v1/wallet/accounts/${customerId}`, + faultHttp:503, expectedHttp:200, expectedCode:'S0503', expectedMessages:['Finance account service is unavailable','Finance account response is invalid'] }, + ], + recovery:'Restore original Order image/config/network identity; submit each original frozen command once, then replay it; require one submission history and no duplicate Firstleg facts.' } +} + +export function assertOrderDependencyFaultObservation(plan, dependency, role, observation) { + const phase = plan?.phases?.find(item => item.dependency === dependency), command = plan?.commands?.[role] + const require = (ok, message) => { if (!ok) throw new Error(`Dependency fault evidence: ${message}`) } + require(Boolean(phase && command), 'unknown dependency or role') + const request = observation?.request, response = observation?.response + require(request?.method === command.method && request.path === command.path + && JSON.stringify(request.body) === JSON.stringify(command.body) + && request.headers?.['Idempotency-Key'] === command.headers['Idempotency-Key'], 'original frozen request/version/key changed') + const requestId = request.headers?.['X-Request-Id'] + require(typeof requestId === 'string' && requestId.length > 0, 'missing correlation ID') + require(response?.http === phase.expectedHttp && String(response.body?.code) === phase.expectedCode + && phase.expectedMessages.includes(response.body.message), 'must be the real Order dependency failure, not auth/version/business validation') + require(Array.isArray(observation.proxyRecords) && observation.proxyRecords.some(row => + row.requestId === requestId && row.method === phase.targetMethod && row.path === phase.targetPath + && row.matched === true && row.responseStatus === phase.faultHttp && row.forwarded === false + && row.internalTokenPresent === true), 'missing exact authenticated upstream fault hit') + require(observation.proxyRecords.filter(row => row.matched === true).every(row => row.method === phase.targetMethod + && row.path === phase.targetPath && row.responseStatus === phase.faultHttp && row.forwarded === false), 'fault escaped selected customer/path') + for (const label of ['before','after']) { + const snapshot = observation[label], detail = snapshot?.detail + require(snapshot?.inboundPageComplete === true && Array.isArray(snapshot.inbounds) + && ['cargoes','parties','internalNotes','statusHistory'].every(field => Array.isArray(detail?.[field])) + && id(detail?.order?.id) === command.orderId && id(detail.order.tenantId) === plan.customerId + && detail.order.status === 'DRAFT', `${label} incomplete/foreign order or Firstleg evidence`) + require(snapshot.task === null || snapshot.task && id(snapshot.task.orderId) === command.orderId + && id(snapshot.task.tenantId) === plan.customerId, `${label} task source mismatch`) + require(snapshot.inbounds.every(row => id(row.tenantId) === plan.customerId && row.orderNo === command.orderNo), `${label} inbound source mismatch`) + } + require(createHash('sha256').update(orderPersistenceFacts(observation.before.detail)).digest('hex') === command.initialOrderSha256, + 'baseline differs from prepared draft') + require(JSON.stringify(observation.before) === JSON.stringify(observation.after), 'order/version/history/Firstleg facts changed') + return { dependency, role, orderId:command.orderId, verified:true, requestId, + snapshotSha256:createHash('sha256').update(JSON.stringify(observation.after)).digest('hex') } +} export function requestedOrderDestinationType(context) { const value = String(context?.context?.orderDestinationType || 'FBA').trim().toUpperCase() return ['FBA', 'OVERSEAS', 'PRIVATE'].includes(value) ? value : 'FBA' @@ -848,11 +909,18 @@ async function waitForInbound(context, tenantId, orderId) { } export async function run(context) { + // Resolve per invocation: this module is cached across serial fresh-customer runs. + const fixture = context.context?.merchantQualificationFixture + const configured = context.options?.merchantUsername || context.options?.merchantPassword + const credentials = fixture ? { username: fixture.username, password: fixture.password } + : configured ? { username: context.options.merchantUsername, password: context.options.merchantPassword } + : { username: process.env.MERCHANT_TEST_USERNAME, password: process.env.MERCHANT_TEST_PASSWORD } + if (!credentials.username || !credentials.password) throw new Error('Client 当前运行缺少完整登录凭据') const loginResponse = await context.request('POST', '/api/auth/login', { - loginId: MERCHANT_USERNAME, password: MERCHANT_PASSWORD, + loginId: credentials.username, password: credentials.password, }, '') const login = context.apiSuccess(MODULE, 'Client 测试账号登录', loginResponse, - `username=${MERCHANT_USERNAME}`) + 'current-run customer credentials') const merchantToken = login?.accessToken if (!merchantToken) throw new Error('Client 登录未返回 accessToken') diff --git a/scripts/order-submit-dependency-fault-executor.mjs b/scripts/order-submit-dependency-fault-executor.mjs new file mode 100644 index 000000000..442037037 --- /dev/null +++ b/scripts/order-submit-dependency-fault-executor.mjs @@ -0,0 +1,251 @@ +// Explicit TEST-only runtime controller. Importing this module performs no HTTP. +import { readFileSync } from 'node:fs' +import { open, mkdir } from 'node:fs/promises' +import { createHash } from 'node:crypto' +import { resolve, dirname } from 'node:path' +import { pathToFileURL } from 'node:url' +import { assertOrderDependencyFaultObservation, orderPersistenceFacts } from './admin-test-env-modules/orders.mjs' + +const runtime = readFileSync(new URL('./order-submit-dependency-fault-runtime.py', import.meta.url), 'utf8') +const proxySource = readFileSync(new URL('./order-submit-dependency-fault-proxy.py', import.meta.url)).toString('base64') +const commonSource = readFileSync(new URL('./test-event-fault-runtime.py', import.meta.url)).toString('base64') +const encode = value => Buffer.from(value).toString('base64') +const sha = value => createHash('sha256').update(JSON.stringify(value)).digest('hex') +const assert = (ok, message) => { if (!ok) throw Error(message) } +const pause = ms => new Promise(resolve => setTimeout(resolve, ms)) +const safeError = error => /^(?:[A-Z][A-Z_0-9]{1,100}|Dependency fault evidence: [A-Za-z /:-]+)$/.test(error?.message || '') + ? error.message : 'ORDER_DEPENDENCY_VERIFICATION_FAILED' + +export function scopeFor(plan) { + assert(/^[A-Za-z0-9_-]{1,80}$/.test(plan?.marker || '') && /^[1-9]\d{0,19}$/.test(plan?.customerId || ''), 'Invalid TEST plan scope') + for (const role of ['merchant', 'platform']) { + const command = plan.commands?.[role] + assert(command?.method === 'POST' && /^[1-9]\d{0,19}$/.test(command.orderId || '') + && command.path === `/api/${role}/v1/orders/first-leg/${command.orderId}/submit${role === 'platform' ? `?tenantId=${plan.customerId}` : ''}` + && Number.isSafeInteger(command.body?.expectedVersion) && command.body.expectedVersion >= 0 + && Object.keys(command.body).length === 1 && /^[A-Za-z0-9_-]{1,128}$/.test(command.headers?.['Idempotency-Key'] || '') + && Object.keys(command.headers).length === 1 + && /^[0-9a-f]{64}$/.test(command.initialOrderSha256 || ''), 'Invalid frozen submission command') + } + assert(plan.commands.merchant.orderId !== plan.commands.platform.orderId, 'Two distinct new drafts required') + assert(plan.runtimeChangesExecuted === false, 'Only a reviewed unexecuted plan is accepted') + assert(Array.isArray(plan.phases) && plan.phases.length === 2, 'Exact two dependency phases required') + for (const [dependency, property, path, messages] of [ + ['IAM','baiyue.clients.iam.url',`/internal/iam/tenants/${plan.customerId}/order-eligibility`,['IAM eligibility service is unavailable','IAM 下单资格响应无效']], + ['FINANCE','baiyue.clients.finance.url',`/internal/finance/v1/wallet/accounts/${plan.customerId}`,['Finance account service is unavailable','Finance account response is invalid']], + ]) { + const rows = plan.phases.filter(row => row.dependency === dependency) + assert(rows.length === 1 && rows[0].urlProperty === property && rows[0].targetPath === path && rows[0].targetMethod === 'GET' + && rows[0].faultHttp === 503 && rows[0].expectedHttp === 200 && rows[0].expectedCode === 'S0503' + && JSON.stringify(rows[0].expectedMessages) === JSON.stringify(messages), 'Dependency failure contract may not be weakened') + } + return { marker: plan.marker, customerId: plan.customerId, + orderIds: Object.fromEntries(['merchant', 'platform'].map(role => [role, plan.commands[role].orderId])), planSha256: sha(plan) } +} + +export function renderOrderFaultAction(action, scope, agent, extra = {}) { + assert(['preflight','prepare','activate','health','mode','status','records','restore','reconcile','verify-restored','finish'].includes(action), 'Unknown runtime action') + assert(/^[A-Za-z0-9_.-]{1,128}$/.test(agent || '') && /^[A-Za-z0-9_-]{1,80}$/.test(scope.marker), 'Invalid agent/scope') + if (['prepare','preflight'].includes(action)) { + const routing = extra.routing + assert(routing && Object.keys(routing).every(key => ['configurationReviewed','sourceSha256','upstreams','nacos'].includes(key)) + && routing.configurationReviewed === true && routing.nacos && Object.keys(routing.nacos).every(key => ['baseUrl','loginPath','configPath'].includes(key)) + && routing.upstreams && Object.keys(routing.upstreams).sort().join(',') === 'FINANCE,IAM', 'Reviewed routing schema only; do not embed credentials') + for (const address of [routing.nacos.baseUrl,...Object.values(routing.upstreams)]) { + const url = new URL(address) + assert(['http:','https:'].includes(url.protocol) && !url.username && !url.password && !url.search && !url.hash + && ['', '/'].includes(url.pathname), 'Routing origins cannot contain credentials/path/query') + } + if (action === 'prepare') assert(/^sha256:[0-9a-f]{64}$/.test(extra.expectedImageId || '') + && /^[0-9a-f]{64}$/.test(routing.sourceSha256 || ''), 'Reviewed image and Nacos source fingerprint required') + } + const data = encode(JSON.stringify({ ...extra, action, scope, commonSource, ...(action === 'prepare' ? { proxySource } : {}) })) + const expression = value => '[' + encode(value).match(/.{1,4000}/g).map(part => `'${part}'`).join(',') + "].join('')" + const groovy = `def p=new ProcessBuilder(['python3','-c',new String(java.util.Base64.decoder.decode(${expression(runtime)}),'UTF-8'),'${data}']).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(${expression(groovy)}),'UTF-8'),c.channel))` +} + +export function createOrderFaultRuntime(env, scope, fetcher = globalThis.fetch) { + assert(env.CHONGCHONG_ENV === 'test', 'CHONGCHONG_ENV=test required') + assert(env.JENKINS_URL && env.JENKINS_USER && env.JENKINS_API_TOKEN && env.FBA_CORE_JENKINS_AGENT, 'Explicit TEST Jenkins credentials and agent required') + return async (action, 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: renderOrderFaultAction(action, scope, env.FBA_CORE_JENKINS_AGENT, extra) }), + }) + } catch { throw Error(`RUNTIME_${action.toUpperCase().replaceAll('-', '_')}_OUTCOME_UNKNOWN_USE_STATUS`) } + assert(response.ok, `RUNTIME_HTTP_${response.status}_NO_AUTO_RETRY`) + const rows = (await response.text()).split(/\r?\n/).filter(row => row.startsWith('ORDER_DEPENDENCY_JSON=')) + assert(rows.length === 1, 'RUNTIME_RESPONSE_AMBIGUOUS_NO_AUTO_RETRY') + const result = JSON.parse(rows[0].slice('ORDER_DEPENDENCY_JSON='.length)) + assert(result?.ok === true, /^[A-Z_0-9]{1,90}$/.test(result?.error || '') ? result.error : 'RUNTIME_ACTION_FAILED') + return result.data + } +} + +export async function runOrderFaultWindow({ plan, routing, expectedImageId, action, snapshot, submit, recovery, wait = pause }) { + const scope = scopeFor(plan), report = { scope, success: false, observations: [], restoration: null } + let prepareAttempted = false, failure + const health = async () => { + for (let i = 0; i < 180; i++) { + const value = await action('health') + if (value.healthy === true) return + await wait(1000) + } + throw Error('ORDER_HEALTH_TIMEOUT') + } + try { + for (const role of ['merchant','platform']) { + const before = await snapshot(role) + assert(createHash('sha256').update(orderPersistenceFacts(before.detail)).digest('hex') === plan.commands[role].initialOrderSha256 + && before.detail.order.status === 'DRAFT' && String(before.detail.order.tenantId) === scope.customerId + && String(before.detail.order.id) === scope.orderIds[role] + && ['cargoes','parties','internalNotes','statusHistory'].every(field => Array.isArray(before.detail[field])) + && before.inboundPageComplete === true, 'PREPARED_DRAFT_CHANGED') + } + prepareAttempted = true + report.prepared = await action('prepare', { routing, expectedImageId }) + for (let i = 0; i < 15; i++) { + if ((await action('status')).proxy?.mode === 'PASS') break + assert(i < 14, 'PROXY_STARTUP_NOT_READY'); await wait(1000) + } + await action('activate'); await health() + for (const dependency of ['IAM','FINANCE']) { + for (const role of ['merchant','platform']) { + const before = await snapshot(role), command = plan.commands[role] + const request = { method: command.method, path: command.path, body: structuredClone(command.body), + headers: { ...command.headers, 'X-Request-Id': `dep-${plan.marker}-${dependency}-${role}` } } + let response + await action('mode', { mode: dependency, ttl: 60 }) + try { response = await submit(role, request) } + finally { await action('mode', { mode: 'PASS' }) } + const after = await snapshot(role) + const proxyRecords = (await action('records')).records.filter(row => row.dependency === dependency) + report.observations.push(assertOrderDependencyFaultObservation(plan, dependency, role, { request, response, before, after, proxyRecords })) + } + } + } catch (error) { failure = error } + finally { + if (prepareAttempted) { + try { + await action('restore') // Never retry a mutation after an ambiguous outcome. + await health() + report.restoredRuntime = await action('verify-restored') + assert(report.restoredRuntime.originalImageId === expectedImageId + && report.restoredRuntime.currentImageId === expectedImageId + && report.restoredRuntime.originalFingerprint === report.restoredRuntime.currentFingerprint + && report.restoredRuntime.nacosSourcesUnchanged === true + && report.restoredRuntime.upstreamReads?.IAM === true && report.restoredRuntime.upstreamReads?.FINANCE === true, + 'RESTORED_IDENTITY_OR_ROUTES_NOT_PROVED') + const businessRecovery = await recovery() + report.businessRecovery = businessRecovery + report.restoration = await action('finish', { businessRecovery }) + assert(report.restoration.restored && report.restoration.proxyStopped && report.restoration.orderRoutesVerified, 'RESTORATION_NOT_COMPLETE') + } catch (error) { report.recoveryError = safeError(error); failure ||= error } + } + } + report.success = !failure && report.observations.length === 4 && Boolean(report.restoration?.restored) + if (failure) report.error = safeError(failure) + return report +} + +function businessClient(env, plan, fetcher = globalThis.fetch) { + const sessions = {} + const call = async (method, path, body, headers = {}, token) => { + const response = await fetcher(new URL(path, env.ADMIN_TEST_GATEWAY), { method, redirect:'error', signal:AbortSignal.timeout(20000), + headers:{'Content-Type':'application/json', ...headers, ...(token ? {Authorization:`Bearer ${token}`} : {})}, ...(body === undefined ? {} : {body:JSON.stringify(body)}) }) + return {http:response.status,body:await response.json()} + } + const ok = value => { assert(value.http === 200 && String(value.body?.code) === '200', 'BUSINESS_READ_OR_RECOVERY_FAILED'); return value.body.data } + const snapshot = async role => { + const command = plan.commands[role] + const detailPath = `/api/${role}/v1/orders/first-leg/${command.orderId}${role === 'platform' ? `?tenantId=${plan.customerId}` : ''}` + const detail = ok(await call('GET',detailPath,undefined,{},sessions[role].accessToken)) + assert(String(detail?.order?.id)===command.orderId && String(detail.order.tenantId)===plan.customerId + && ['cargoes','parties','internalNotes','statusHistory'].every(field=>Array.isArray(detail[field])), 'ORDER_SNAPSHOT_INCOMPLETE_OR_FOREIGN') + const taskResponse = await call('GET',`/api/platform/v1/first-leg/tasks/by-order/${command.orderId}?tenantId=${plan.customerId}`,undefined,{},sessions.platform.accessToken) + assert(taskResponse.http === 200 && ['200','A0404'].includes(String(taskResponse.body?.code)), 'TASK_SNAPSHOT_FAILED') + const task = String(taskResponse.body.code) === '200' ? taskResponse.body.data ?? null : null + const inbounds = []; let total + for (let page=1;;page++) { + const value=ok(await call('GET',`/api/platform/inbounds?tenantId=${plan.customerId}&orderNo=${encodeURIComponent(command.orderNo)}&page=${page}&size=200`,undefined,{},sessions.platform.accessToken)) + assert(Array.isArray(value.records) && Number(value.page)===page && Number.isSafeInteger(Number(value.total)), 'INBOUND_PAGE_INVALID') + total ??= Number(value.total); assert(Number(value.total)===total && value.records.length===Math.min(200,Math.max(0,total-inbounds.length)), 'INBOUND_PAGE_INCOMPLETE') + for(const row of value.records)assert(row.orderNo===command.orderNo && String(row.tenantId)===plan.customerId && !inbounds.some(before=>String(before.id)===String(row.id)), 'INBOUND_SCOPE_OR_DUPLICATE') + inbounds.push(...value.records);if(inbounds.length===total)break + } + return {detail,task,inbounds,inboundPageComplete:true} + } + const submit = (role, request) => call(request.method,request.path,request.body,request.headers,sessions[role].accessToken) + return { snapshot, submit, + async login() { + for(const [role,prefix] of [['platform','ADMIN_TEST'],['merchant','MERCHANT_TEST']]) { + assert(env[prefix+'_USERNAME'] && env[prefix+'_PASSWORD'], 'EXPLICIT_BUSINESS_CREDENTIALS_REQUIRED') + sessions[role]=ok(await call('POST','/api/auth/login',{loginId:env[prefix+'_USERNAME'],password:env[prefix+'_PASSWORD']})) + } + const profile=ok(await call('GET','/api/merchant/v1/iam/tenant/profile',undefined,{},sessions.merchant.accessToken)) + assert(String(profile.id)===plan.customerId && profile.status==='ACTIVE','REGISTERED_CUSTOMER_SESSION_MISMATCH') + }, + async recovery() { + const evidence={} + for(const role of ['merchant','platform']) { + const command=plan.commands[role] + ok(await submit(role,command)) + let before=await snapshot(role) + for(let i=0;i<10;i++) { await pause(500);const next=await snapshot(role);if(sha(before)===sha(next)){before=next;break}before=next } + assert(before.detail.order.status==='SUBMITTED' && before.detail.statusHistory.filter(row=>row.action==='SUBMITTED').length===1 + && before.inbounds.length<=1, 'RECOVERY_SINGLE_SUBMISSION_NOT_PROVED') + ok(await submit(role,command));const after=await snapshot(role) + assert(sha(before)===sha(after),'RECOVERY_REPLAY_DRIFT') + evidence[role]={orderId:command.orderId,status:'SUBMITTED',sameKeyReplayVerified:true,snapshotSha256:sha(after)} + } + return evidence + }, + async logout() { + let success=true + for(const session of Object.values(sessions))try {ok(await call('POST','/api/auth/logout',{refreshToken:session.refreshToken},{},session.accessToken))}catch{success=false} + return success + }, + } +} + +async function main(argv) { + const [mode,...rest]=argv, args={} + for(let i=0;i{console.error('ORDER_DEPENDENCY_ARGUMENT_OR_CONFIGURATION_ERROR');process.exitCode=1}) diff --git a/scripts/order-submit-dependency-fault-executor.test.mjs b/scripts/order-submit-dependency-fault-executor.test.mjs new file mode 100644 index 000000000..1e204dedf --- /dev/null +++ b/scripts/order-submit-dependency-fault-executor.test.mjs @@ -0,0 +1,209 @@ +import test from 'node:test' +import assert from 'node:assert/strict' +import { spawnSync } from 'node:child_process' +import { fileURLToPath } from 'node:url' +import { buildOrderDependencyFaultPlan } from './admin-test-env-modules/orders.mjs' +import { scopeFor, renderOrderFaultAction, createOrderFaultRuntime, runOrderFaultWindow } from './order-submit-dependency-fault-executor.mjs' + +const image = 'sha256:' + 'a'.repeat(64) +function fixture() { + const detail = id => ({order:{id,tenantId:'20',orderNo:`FL${id}`,version:2,status:'DRAFT'},cargoes:[],parties:[],internalNotes:[],statusHistory:[]}) + const raw={marker:'TEST-DEPENDENCY',customerId:'20',registeredCustomerId:'20',merchant:detail('101'),platform:detail('102')} + return {raw,plan:buildOrderDependencyFaultPlan(raw)} +} +const env={CHONGCHONG_ENV:'test',JENKINS_URL:'https://offline.invalid/',JENKINS_USER:'offline',JENKINS_API_TOKEN:'never-print-this',FBA_CORE_JENKINS_AGENT:'test-agent'} + +test('renderer binds TEST scope and emits no Jenkins credential; ambiguous remote mutation is never retried',async()=>{ + const {plan}=fixture(),scope=scopeFor(plan) + assert.ok(!renderOrderFaultAction('status',scope,env.FBA_CORE_JENKINS_AGENT).includes(env.JENKINS_API_TOKEN)) + assert.throws(()=>renderOrderFaultAction('delete-all',scope,'agent')) + assert.throws(()=>renderOrderFaultAction('prepare',scope,'agent',{routing:{configurationReviewed:true,password:'must-not-embed'}})) + assert.throws(()=>createOrderFaultRuntime({...env,CHONGCHONG_ENV:'prod'},scope)) + for(const mutate of [p=>p.phases[0].expectedCode='A0403',p=>p.phases[0].targetPath+='/other',p=>p.phases[0].expectedMessages.push('anything')]) { + const changed=structuredClone(plan);mutate(changed);assert.throws(()=>scopeFor(changed)) + } + let calls=0 + const runtime=createOrderFaultRuntime(env,scope,async()=>{calls++;throw Error('connection lost with secret')}) + await assert.rejects(runtime('activate'),/OUTCOME_UNKNOWN_USE_STATUS/) + assert.equal(calls,1) +}) + +function windowFixture(fault='') { + const {raw,plan}=fixture(),calls=[],records=[] + let mode='PASS' + const options={plan,routing:{configurationReviewed:true},expectedImageId:image,wait:async()=>{}, + async action(action,extra={}) { + calls.push(action) + if(action==='prepare'&&fault==='prepare-unknown')throw Error('PREPARE_OUTCOME_UNKNOWN') + if(action==='restore'&&['restore-unknown','prepare-unknown'].includes(fault))throw Error('UNCERTAIN_MUTATION_REQUIRES_RECONCILE') + if(action==='status')return {proxy:{mode:'PASS'}} + if(action==='health')return {healthy:true} + if(action==='mode'){mode=extra.mode;return {mode}} + if(action==='records')return {records} + if(action==='verify-restored')return {originalImageId:image,currentImageId:image,originalFingerprint:'same',currentFingerprint:'same',upstreamReads:{IAM:true,FINANCE:true},nacosSourcesUnchanged:true} + if(action==='finish')return {restored:true,healthy:true,proxyStopped:true,orderRoutesVerified:true} + return {} + }, + async snapshot(role){return {detail:structuredClone(raw[role]),task:null,inbounds:[],inboundPageComplete:true}}, + async submit(role,request){ + const phase=plan.phases.find(item=>item.dependency===mode) + records.push({requestId:request.headers['X-Request-Id'],dependency:mode,method:'GET',path:phase.targetPath, + matched:true,responseStatus:503,forwarded:false,internalTokenPresent:true}) + return {http:200,body:{code:fault==='wrong-business-code'?'A0403':'S0503',message:phase.expectedMessages[0]}} + }, + async recovery(){calls.push('business-recovery');if(fault==='business-recovery-failed')throw Error('RECOVERY_REPLAY_DRIFT') + return {merchant:{orderId:'101',status:'SUBMITTED',sameKeyReplayVerified:true},platform:{orderId:'102',status:'SUBMITTED',sameKeyReplayVerified:true}}}, + } + return {options,calls} +} + +test('four observations restore original routes and verify business recovery before stopping proxy',async()=>{ + const {options,calls}=windowFixture(),report=await runOrderFaultWindow(options) + assert.equal(report.success,true);assert.equal(report.observations.length,4) + assert.ok(calls.indexOf('restore'){ + const {options,calls}=windowFixture('wrong-business-code'),report=await runOrderFaultWindow(options) + assert.equal(report.success,false);assert.equal(report.restoration.restored,true) + assert.ok(calls.includes('finish')) +}) +for(const fault of ['prepare-unknown','restore-unknown','business-recovery-failed'])test(`unsafe cleanup blocks finish: ${fault}`,async()=>{ + const {options,calls}=windowFixture(fault),report=await runOrderFaultWindow(options) + assert.equal(report.success,false);assert.equal(calls.filter(action=>action==='prepare').length,1) + assert.equal(calls.filter(action=>action==='restore').length,1);assert.ok(!calls.includes('finish')) +}) + +test('offline runtime journals Docker transitions, reconciles observed outcomes and retains original configuration (no Docker/HTTP)',()=>{ + const script=String.raw` +import importlib.util,sys,copy,json,unittest +from pathlib import Path +spec=importlib.util.spec_from_file_location('runtime',sys.argv[1]);m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m) +common=m.common_tools(Path(sys.argv[2]).read_text(encoding='utf-8')) +class Fake: + def __init__(self): + self.current={'Id':'original-container-id','Name':'/firstleg-test-order','Image':'sha256:'+'a'*64, + 'Config':{'Image':'offline/order:tag','Env':['DEPLOY_ENV=test','SERVER_PORT=9204','INTERNAL_SERVICE_TOKEN=offline-secret-long'], + 'Entrypoint':copy.deepcopy(m.ENTRYPOINT),'Cmd':None,'Labels':{'com.docker.compose.project':'firstleg-test'}}, + 'HostConfig':{'AutoRemove':False,'ExtraHosts':['host.docker.internal:host-gateway']},'Mounts':[], + 'NetworkSettings':{'Networks':{'test':{'IPAddress':'172.20.0.10','NetworkID':'n1','Aliases':['order','original-con'],'MacAddress':'aa:bb:cc:dd:ee:ff'}}}, + 'State':{'Running':True,'Health':{'Status':'healthy'}}} + self.original=copy.deepcopy(self.current);self.calls=[];self.fail=None;self.failBefore=None;self.serial=0 + def inspect(self,name): + if not self.current or name not in ('firstleg-test-order',self.current['Id']):raise RuntimeError('DOCKER_404') + return copy.deepcopy(self.current) + def docker(self,method,path,body=None): + if method=='GET':return {'Id':self.original['Image']} + self.calls.append((method,path));result=None + if self.failBefore=='stop' and '/stop?' in path:raise RuntimeError('REQUEST_OUTCOME_UNKNOWN') + if '/stop?' in path:self.current['State']['Running']=False;op='stop' + elif method=='DELETE':self.current=None;op='delete' + elif '/create?' in path: + self.serial+=1;new_id='new-container-'+str(self.serial);op='create' + self.current={'Id':new_id,'Name':'/firstleg-test-order','Image':self.original['Image'], + 'Config':{k:copy.deepcopy(v) for k,v in body.items() if k not in ('HostConfig','NetworkingConfig')}, + 'HostConfig':copy.deepcopy(body['HostConfig']),'Mounts':[], + 'NetworkSettings':{'Networks':{}},'State':{'Running':False,'Health':{'Status':'healthy'}}} + for name,entry in body['NetworkingConfig']['EndpointsConfig'].items(): + self.current['NetworkSettings']['Networks'][name]={'IPAddress':entry['IPAMConfig']['IPv4Address'],'NetworkID':'n1', + 'Aliases':entry['Aliases']+[new_id[:12]],'MacAddress':entry['MacAddress'],'IPAMConfig':entry['IPAMConfig']} + result={'Id':new_id} + elif path.endswith('/start'):self.current['State']['Running']=True;op='start' + else:raise AssertionError(path) + if self.fail==op:self.fail=None;raise RuntimeError('RESPONSE_LOST') + return result +class RuntimeTests(unittest.TestCase): + def fixture(self): + f=Fake();tools=dict(common);tools.update(inspect=f.inspect,docker=f.docker) + scope={'marker':'OFFLINE','customerId':'7','orderIds':{'merchant':'10','platform':'11'},'planSha256':'b'*64} + r=m.Runtime(scope,tools,Path('offline-not-created'));saved=[];controls=[] + r.save=lambda state:saved.append(copy.deepcopy(state));r.control=lambda state,mode=None,ttl=0:(controls.append(mode) or {'mode':'PASS'}) + r.probes=lambda state:{'IAM':True,'FINANCE':True} + r.config_fingerprint=lambda container,routing:'reviewed-config' + state={'scope':scope,'phase':'PROXY_STARTED','originalId':f.original['Id'],'originalImageId':f.original['Image'], + 'originalFingerprint':common['fingerprint'](f.original),'originalBody':common['recreation'](f.original,common['env_of'](f.original),full=True), + 'proxyPort':12345,'controlPort':12346,'mutation':None,'routing':{},'routingSourceSha256':'reviewed-config'} + return f,r,state,saved,controls + def test_original_fingerprint_config_and_image_restored(self): + f,r,s,saved,controls=self.fixture();r.activate(s) + self.assertTrue(r.matches_injected(f.current,s));self.assertEqual(s['phase'],'INJECTED') + self.assertTrue(any(x.get('mutation',{}).get('step')=='CREATE_INTENT' for x in saved if x.get('mutation'))) + r.restore(s);result=r.verify_restored(s) + self.assertEqual(result['originalFingerprint'],result['currentFingerprint']);self.assertEqual(f.current['Config'],f.original['Config']) + def test_unknown_outcomes_never_auto_repeat_docker_and_can_be_explicitly_reconciled(self): + for operation in ['stop','delete','create','start']: + f,r,s,saved,controls=self.fixture();f.fail=operation + with self.assertRaisesRegex(RuntimeError,'UNCERTAIN'):r.activate(s) + count=len(f.calls) + with self.assertRaisesRegex(RuntimeError,'RECONCILE'):r.restore(s) + self.assertEqual(len(f.calls),count);self.assertEqual(controls[-1],'PASS') + r.reconcile(s);self.assertEqual(len(f.calls),count) + r.restore(s);self.assertTrue(r.verify_restored(s)['healthy']) + def test_unrecognized_configuration_is_never_overwritten(self): + f,r,s,saved,controls=self.fixture();r.activate(s);f.current['Config']['Env'].append('UNRELATED_CHANGE=1');count=len(f.calls) + with self.assertRaisesRegex(RuntimeError,'UNRECOGNIZED'):r.restore(s) + self.assertEqual(len(f.calls),count) + def test_unknown_stop_without_observed_stop_is_not_misreported_as_restored(self): + f,r,s,saved,controls=self.fixture();f.failBefore='stop' + with self.assertRaisesRegex(RuntimeError,'UNCERTAIN'):r.activate(s) + count=len(f.calls) + with self.assertRaisesRegex(RuntimeError,'NO_BLIND_RECREATE'):r.reconcile(s) + self.assertTrue(s['uncertain']);self.assertTrue(f.current['State']['Running']);self.assertEqual(len(f.calls),count) + def test_restore_still_runs_when_proxy_control_is_unreachable(self): + f,r,s,saved,controls=self.fixture();r.activate(s) + def unavailable(*args):raise RuntimeError('CONTROL_LOST') + r.control=unavailable;r.restore(s);self.assertTrue(r.verify_restored(s)['healthy']) + def test_finish_refuses_unproved_business_route_recovery(self): + f,r,s,saved,controls=self.fixture();r.activate(s);r.restore(s) + from unittest.mock import patch + with patch.object(Path,'read_text',return_value='OFFLINE'),patch.object(Path,'is_symlink',return_value=False): + with self.assertRaisesRegex(RuntimeError,'ROUTES_NOT_PROVED'):r.finish(s,{}) + def test_changed_nacos_configuration_prevents_recovery_claim(self): + f,r,s,saved,controls=self.fixture();r.activate(s);r.restore(s);r.config_fingerprint=lambda *args:'changed' + with self.assertRaisesRegex(RuntimeError,'NACOS_ROUTE_CONFIGURATION_CHANGED'):r.verify_restored(s) + def test_nacos_source_reads_are_explicit_authenticated_and_redacted(self): + from unittest.mock import patch + f,r,s,saved,controls=self.fixture() + f.current['Config']['Env']+=['NACOS_SERVER_ADDR=host.docker.internal:8848','NACOS_USERNAME=offline-user', + 'NACOS_PASSWORD=offline-password','NACOS_GROUP=BAIYUE','NACOS_NAMESPACE=test-only'] + routing={'nacos':{'baseUrl':'http://127.0.0.1:8848','loginPath':'/nacos/v1/auth/users/login','configPath':'/nacos/v1/cs/configs'}} + calls=[] + class Response: + status=200 + def __init__(self,value):self.value=value + def __enter__(self):return self + def __exit__(self,*args):pass + def read(self):return self.value + def read(req,timeout): + calls.append(req) + return Response(b'{"accessToken":"offline-token"}' if not isinstance(req,str) else b'spring:\n test: only\n') + with patch.object(m,'http_open',side_effect=read): + value=m.Runtime.config_fingerprint(r,f.current,routing) + self.assertEqual(len(value),64);self.assertEqual(len(calls),3) + self.assertIn(b'password=offline-password',calls[0].data) + self.assertTrue(all('accessToken=offline-token' in url for url in calls[1:])) + routing['nacos']['baseUrl']='http://other.invalid:8848' + with self.assertRaisesRegex(RuntimeError,'EXPLICIT_NACOS'):m.Runtime.config_fingerprint(r,f.current,routing) + self.assertEqual(len(calls),3) + def test_effective_source_extraction_excludes_all_event_and_sql_functions(self): + self.assertEqual(set(common),{'DockerConnection','docker','inspect','env_of','require_test','recreation','fingerprint'}) + def test_control_token_is_not_sent_to_an_unowned_listener(self): + from unittest.mock import patch + f,r,s,saved,controls=self.fixture();s.update(proxyPid=777,controlToken='private-control') + command=str(r.directory/'proxy.py').encode()+b'\0' + tcp='header\n 0: 00000000:3039 00000000:0000 0A 0:0 0:0 0 1000 0 123\n' + with patch.object(Path,'read_bytes',return_value=command),patch.object(Path,'iterdir',return_value=iter([Path('fd')])), \ + patch.object(m.os,'readlink',return_value='socket:[123]'),patch.object(Path,'read_text',return_value=tcp), \ + patch.object(m,'http_open') as request: + with self.assertRaisesRegex(RuntimeError,'LISTENERS_NOT_OWNED'):m.Runtime.control(r,s) + request.assert_not_called() +result=unittest.TextTestRunner().run(unittest.defaultTestLoader.loadTestsFromTestCase(RuntimeTests)) +sys.exit(0 if result.wasSuccessful() else 1) +` + const result=spawnSync(process.env.PYTHON||'python',['-c',script, + fileURLToPath(new URL('./order-submit-dependency-fault-runtime.py',import.meta.url)), + fileURLToPath(new URL('./test-event-fault-runtime.py',import.meta.url))],{encoding:'utf8',env:{...process.env,PYTHONDONTWRITEBYTECODE:'1'}}) + assert.equal(result.status,0,result.stderr || result.error?.message) + assert.match(result.stderr,/Ran 10 tests/) +}) diff --git a/scripts/order-submit-dependency-fault-proxy.py b/scripts/order-submit-dependency-fault-proxy.py new file mode 100644 index 000000000..240507e8d --- /dev/null +++ b/scripts/order-submit-dependency-fault-proxy.py @@ -0,0 +1,176 @@ +"""TEST-only scoped synchronous dependency proxy. Default/expired mode is PASS. + +Configuration and control token are read from a mode-0600 file, never argv values. +The runtime owns lifecycle; this file never invokes Docker, SQL or a shell. +""" +import argparse +import hmac +import http.client +import json +import os +import re +import threading +import time +import urllib.parse +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +HOP = {'connection', 'keep-alive', 'proxy-authenticate', 'proxy-authorization', + 'te', 'trailer', 'transfer-encoding', 'upgrade', 'host', 'content-length'} + + +def forward_headers(headers, length): + connection = {k.lower(): v for k, v in headers.items()}.get('connection', '') + excluded = HOP | {x.strip().lower() for x in connection.split(',')} + result = {k: v for k, v in headers.items() if k.lower() not in excluded} + result['Content-Length'] = str(length) + return result + + +def transport(upstream, method, path, headers, body): + url = urllib.parse.urlsplit(upstream) + cls = http.client.HTTPSConnection if url.scheme == 'https' else http.client.HTTPConnection + conn = cls(url.hostname, url.port, timeout=8) + try: + conn.request(method, path, body, forward_headers(headers, len(body))) + response = conn.getresponse() + return response.status, dict(response.getheaders()), response.read() + finally: + conn.close() + + +class Router: + def __init__(self, config, forward=transport, clock=time.monotonic, record=None): + self.config, self.forward, self.clock = config, forward, clock + self.mode, self.deadline, self.generation = 'PASS', 0, 0 + self.lock = threading.Lock() + self.record = record or (lambda row: None) + self.counts = {'forwarded': 0, 'faulted': 0} + customer = config['customerId'] + if not re.fullmatch(r'[1-9][0-9]{0,19}', customer): + raise ValueError('INVALID_CUSTOMER') + if not config.get('requestIds') or any(not re.fullmatch(r'[A-Za-z0-9_.:-]{1,128}', value) for value in config['requestIds']): + raise ValueError('BOUNDED_REQUEST_IDS_REQUIRED') + self.paths = {'IAM': '/internal/iam/tenants/' + customer + '/order-eligibility', + 'FINANCE': '/internal/finance/v1/wallet/accounts/' + customer} + + def status(self): + with self.lock: + if self.mode != 'PASS' and self.clock() >= self.deadline: + self.mode, self.deadline = 'PASS', 0 + return {'mode': self.mode, 'generation': self.generation, + 'remainingSeconds': max(0, self.deadline - self.clock()), **self.counts} + + 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: + # Even an authenticated malformed command disarms first. + self.mode, self.deadline = 'PASS', 0 + if mode not in ('PASS', 'IAM', 'FINANCE') 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): + prefix, separator, rest = path.lstrip('/').partition('/') + dependency = {'iam': 'IAM', 'finance': 'FINANCE'}.get(prefix) + if not dependency or not separator: + return 404, {}, b'Unknown dependency route' + actual_path = '/' + rest + 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']) + target = method == 'GET' and actual_path == self.paths[dependency] + request_id = normalized.get('x-request-id', '') + selected = target and peer in self.config['orderIps'] and authenticated and request_id in self.config['requestIds'] + mode = self.status()['mode'] + fault = selected and mode == dependency + # Only bounded correlation IDs and this customer's selected paths enter evidence. + safe_id = request_id if re.fullmatch(r'[A-Za-z0-9_.:-]{1,128}', request_id) else None + if fault: + result = (503, {'Content-Type': 'text/plain'}, b'TEST dependency temporarily unavailable') + else: + result = self.forward(self.config['upstreams'][dependency], method, actual_path, headers, body) + with self.lock: + self.counts['faulted' if fault else 'forwarded'] += 1 + if selected: + self.record({'requestId': safe_id, 'method': method, 'path': actual_path, + 'dependency': dependency, 'matched': fault, 'responseStatus': result[0], + 'forwarded': not fault, 'internalTokenPresent': True, + 'internalTokenValidated': True, 'mode': mode}) + return result + + +def main(path): + config_path = Path(path) + if config_path.is_symlink() or config_path.stat().st_mode & 0o077: + raise RuntimeError('PRIVATE_CONFIG_REQUIRED') + config = json.loads(config_path.read_text()) + evidence_path = config_path.parent / 'proxy-evidence.jsonl' + evidence_lock = threading.Lock() + + def record(row): + with evidence_lock: + fd = os.open(evidence_path, os.O_WRONLY | os.O_APPEND | os.O_CREAT, 0o600) + with os.fdopen(fd, 'w') as stream: + stream.write(json.dumps(row) + '\n'); stream.flush(); os.fsync(stream.fileno()) + + router = Router(config, record=record) + + class Handler(BaseHTTPRequestHandler): + def log_message(self, *_args): + pass + + def dispatch(self): + try: + length = int(self.headers.get('Content-Length', '0')) + if length < 0 or length > 2 * 1024 * 1024: + raise ValueError('BODY_LIMIT') + body = self.rfile.read(length) + if self.server.control: + token = self.headers.get('X-Order-Fault-Control', '') + if not hmac.compare_digest(token, config['controlToken']): + raise PermissionError('CONTROL_AUTH_REQUIRED') + if self.command == 'GET' and self.path == '/status': + value = router.status() + elif self.command == 'POST' and self.path == '/mode': + value = json.loads(body) + value = router.set_mode(token, value.get('mode'), value.get('ttl', 0)) + else: + raise ValueError('UNKNOWN_CONTROL') + status, headers, response = 200, {'Content-Type': 'application/json'}, json.dumps(value).encode() + else: + status, headers, response = router.dispatch(self.command, self.path, dict(self.headers), body, self.client_address[0]) + except PermissionError: + status, headers, response = 403, {}, b'Forbidden' + except (ValueError, KeyError): + if self.server.control and hmac.compare_digest(self.headers.get('X-Order-Fault-Control', ''), config['controlToken']): + router.set_mode(config['controlToken'], 'PASS') + status, headers, response = 400, {}, b'Invalid request' + except Exception: + status, headers, response = 502, {}, b'Upstream transport failed' + self.send_response(status) + for name, value in headers.items(): + if name.lower() not in HOP: + self.send_header(name, value) + self.send_header('Content-Length', str(len(response))); self.end_headers() + if self.command != 'HEAD': + self.wfile.write(response) + + do_GET = do_POST = do_PUT = do_PATCH = do_DELETE = do_HEAD = do_OPTIONS = dispatch + + business = ThreadingHTTPServer(('0.0.0.0', config['proxyPort']), Handler); business.control = False + control = ThreadingHTTPServer(('127.0.0.1', config['controlPort']), Handler); control.control = True + (config_path.parent / 'proxy.pid').write_text(str(os.getpid())) + os.chmod(config_path.parent / 'proxy.pid', 0o600) + threading.Thread(target=business.serve_forever, daemon=True).start() + control.serve_forever() + + +if __name__ == '__main__': + parser = argparse.ArgumentParser(); parser.add_argument('--config', required=True) + main(parser.parse_args().config) diff --git a/scripts/order-submit-dependency-fault-proxy.test.mjs b/scripts/order-submit-dependency-fault-proxy.test.mjs new file mode 100644 index 000000000..73b4c64f6 --- /dev/null +++ b/scripts/order-submit-dependency-fault-proxy.test.mjs @@ -0,0 +1,69 @@ +import test from 'node:test' +import assert from 'node:assert/strict' +import { spawnSync } from 'node:child_process' +import { fileURLToPath } from 'node:url' + +test('offline proxy routing/authentication/TTL/transparency and redaction (no sockets or HTTP)', () => { + const script = String.raw` +import importlib.util,json,sys,unittest +spec=importlib.util.spec_from_file_location('proxy',sys.argv[1]);p=importlib.util.module_from_spec(spec);spec.loader.exec_module(p) +class ProxyTests(unittest.TestCase): + def setUp(self): + self.now=[10.0];self.calls=[];self.logs=[] + self.config={'customerId':'7','controlToken':'private-control','internalToken':'private-internal-token', + 'orderIps':['172.20.0.10'],'requestIds':['fixture-request'],'upstreams':{'IAM':'http://iam.invalid:9201','FINANCE':'http://finance.invalid:9207'}} + def forward(*args): self.calls.append(args);return 201,{'X-Upstream':'kept'},b'original-response' + self.router=p.Router(self.config,forward=forward,clock=lambda:self.now[0],record=self.logs.append) + self.headers={'X-Internal-Service-Token':'private-internal-token','X-Request-Id':'fixture-request'} + self.path='/iam/internal/iam/tenants/7/order-eligibility' + def send(self,path=None,method='GET',headers=None,peer='172.20.0.10'): + return self.router.dispatch(method,path or self.path,headers or self.headers,b'original-body',peer) + def test_default_pass_and_transport_are_unchanged(self): + self.assertEqual(self.send(),(201,{'X-Upstream':'kept'},b'original-response')) + self.assertEqual(self.calls[0][2],'/internal/iam/tenants/7/order-eligibility') + self.assertEqual(self.calls[0][3]['X-Internal-Service-Token'],'private-internal-token') + self.assertEqual(self.calls[0][4],b'original-body') + def test_only_exact_authenticated_customer_get_from_order_faults(self): + self.router.set_mode('private-control','IAM',60) + self.assertEqual(self.send()[0],503);self.assertEqual(len(self.calls),0) + for path,method,headers,peer in [ + (self.path.replace('/7/','/8/'),'GET',self.headers,'172.20.0.10'), + (self.path+'?other=1','GET',self.headers,'172.20.0.10'), + (self.path,'POST',self.headers,'172.20.0.10'), + (self.path,'GET',{'X-Internal-Service-Token':'wrong'},'172.20.0.10'), + (self.path,'GET',{**self.headers,'X-Request-Id':'other-order-request'},'172.20.0.10'), + (self.path,'GET',self.headers,'172.20.0.11'), + ('/finance/internal/finance/v1/wallet/accounts/7','GET',self.headers,'172.20.0.10'), + ('/iam/internal/iam/finance-customers?page=1','GET',self.headers,'172.20.0.10')]: + self.assertEqual(self.send(path,method,headers,peer)[0],201) + def test_expiry_needs_no_control_request_and_returns_to_pass(self): + self.router.set_mode('private-control','FINANCE',5) + path='/finance/internal/finance/v1/wallet/accounts/7' + self.assertEqual(self.send(path)[0],503) + self.now[0]=15.0 + self.assertEqual(self.send(path)[0],201);self.assertEqual(self.router.status()['mode'],'PASS') + def test_control_auth_and_malformed_authenticated_command(self): + self.router.set_mode('private-control','IAM',10) + with self.assertRaises(PermissionError):self.router.set_mode('wrong','PASS') + for mode,ttl in [('UNKNOWN',10),('IAM',0),('IAM',61),('IAM',True)]: + with self.assertRaises(ValueError):self.router.set_mode('private-control',mode,ttl) + self.assertEqual(self.router.status()['mode'],'PASS') + def test_evidence_contains_no_secret_headers_or_payload(self): + self.router.set_mode('private-control','IAM',10);self.send() + serialized=json.dumps(self.logs) + for secret in ['private-control','private-internal-token','original-body']:self.assertNotIn(secret,serialized) + self.assertTrue(self.logs[0]['internalTokenValidated']);self.assertFalse(self.logs[0]['forwarded']) + self.assertEqual(self.logs[0]['requestId'],'fixture-request') + def test_hop_headers_removed_but_identity_preserved(self): + headers=p.forward_headers({'connection':'x-hop','x-hop':'drop','Host':'old','X-Internal-Service-Token':'keep','Content-Length':'99'},3) + self.assertNotIn('x-hop',headers);self.assertNotIn('Host',headers) + self.assertEqual(headers['X-Internal-Service-Token'],'keep');self.assertEqual(headers['Content-Length'],'3') +result=unittest.TextTestRunner().run(unittest.defaultTestLoader.loadTestsFromTestCase(ProxyTests)) +sys.exit(0 if result.wasSuccessful() else 1) +` + const result = spawnSync(process.env.PYTHON || 'python', ['-c', script, + fileURLToPath(new URL('./order-submit-dependency-fault-proxy.py', import.meta.url))], + { encoding:'utf8', env:{...process.env,PYTHONDONTWRITEBYTECODE:'1'} }) + assert.equal(result.status,0,result.stderr || result.error?.message) + assert.match(result.stderr,/Ran 6 tests/) +}) diff --git a/scripts/order-submit-dependency-fault-runtime.py b/scripts/order-submit-dependency-fault-runtime.py new file mode 100644 index 000000000..3823f2d2a --- /dev/null +++ b/scripts/order-submit-dependency-fault-runtime.py @@ -0,0 +1,445 @@ +"""Bounded remote TEST actions; injected by the JS executor, never invoked at import. + +Only Order is recreated. Every Docker mutation is journaled before sending; an +uncertain result blocks further mutations until explicit read-only reconciliation. +""" +import ast +import base64 +import copy +import hashlib +import http.client +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 + +ORDER = 'firstleg-test-order' +ENTRYPOINT = ['sh', '-c', 'exec java $JAVA_OPTS org.springframework.boot.loader.launch.JarLauncher'] +PROPERTIES = {'IAM': 'baiyue.clients.iam.url', 'FINANCE': 'baiyue.clients.finance.url'} +UPSTREAM_CONTAINERS = {'IAM': 'firstleg-test-iam', 'FINANCE': 'firstleg-test-finance'} + + +class NoRedirect(urllib.request.HTTPRedirectHandler): + def redirect_request(self, *_args, **_kwargs): return None + + +def http_open(request, timeout): + return urllib.request.build_opener(NoRedirect()).open(request, timeout=timeout) + + +def common_tools(source): + """Reuse only Docker read/config reconstruction primitives, never event/SQL functions.""" + tree = ast.parse(source) + allowed = {'DockerConnection', 'docker', 'inspect', 'env_of', 'require_test', 'recreation', 'fingerprint'} + nodes = [node for node in tree.body if isinstance(node, (ast.Import, ast.ImportFrom)) + or isinstance(node, (ast.FunctionDef, ast.ClassDef)) and node.name in allowed] + namespace = {'__name__': 'order_dependency_primitives'} + exec(compile(ast.Module(body=nodes, type_ignores=[]), '', 'exec'), namespace) + return {name: namespace[name] for name in allowed} + + +def digest(value): + return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(',', ':')).encode()).hexdigest() + + +def private_write(path, value): + temporary = path.with_name(path.name + '.next') + fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, 'w') as stream: + json.dump(value, stream); stream.flush(); os.fsync(stream.fileno()) + os.replace(temporary, path) + parent = os.open(path.parent, os.O_RDONLY) + try: os.fsync(parent) + finally: os.close(parent) + + +def validate_scope(scope): + if set(scope) != {'marker', 'customerId', 'orderIds', 'planSha256'} or not re.fullmatch(r'[A-Za-z0-9_-]{1,80}', scope['marker']): + raise RuntimeError('INVALID_SCOPE') + if not re.fullmatch(r'[1-9][0-9]{0,19}', scope['customerId']) or set(scope['orderIds']) != {'merchant', 'platform'}: + raise RuntimeError('INVALID_SCOPE') + if any(not re.fullmatch(r'[1-9][0-9]{0,19}', value) for value in scope['orderIds'].values()) or len(set(scope['orderIds'].values())) != 2: + raise RuntimeError('TWO_DISTINCT_ORDERS_REQUIRED') + if not re.fullmatch(r'[0-9a-f]{64}', scope['planSha256']): raise RuntimeError('INVALID_PLAN_HASH') + + +class Runtime: + def __init__(self, scope, tools, base=Path('/tmp')): + validate_scope(scope) + self.scope, self.tools = scope, tools + self.directory = base / ('chongchong-order-dependency-' + scope['marker']) + self.path = self.directory / 'state.json' + self.lock = base / 'chongchong-order-dependency.lock' + + def save(self, state): private_write(self.path, state) + + def load(self): + if self.directory.is_symlink() or self.directory.stat().st_uid != os.getuid() or self.directory.stat().st_mode & 0o077: + raise RuntimeError('UNSAFE_STATE_DIRECTORY') + if self.path.is_symlink() or self.path.stat().st_mode & 0o077: raise RuntimeError('UNSAFE_STATE_FILE') + state = json.loads(self.path.read_text()) + if state['scope'] != self.scope: raise RuntimeError('SCOPE_CHANGED') + if (state.get('mutation') or {}).get('step', '').endswith('_INTENT'): + state['uncertain'] = True # Includes process death between HTTP completion and journal commit. + return state + + def container(self): + value = self.tools['inspect'](ORDER) + self.tools['require_test'](value, ORDER) + if self.tools['env_of'](value).get('DEPLOY_ENV') != 'test': raise RuntimeError('NOT_TEST_ENV') + return value + + def control(self, state, mode=None, ttl=0): + pid = state.get('proxyPid') + if not isinstance(pid, int) or pid <= 1: raise RuntimeError('PROXY_PID_NOT_READY') + process = Path('/proc') / str(pid) + if str(self.directory / 'proxy.py').encode() not in (process / 'cmdline').read_bytes().split(b'\0'): + raise RuntimeError('PROXY_PROCESS_OWNERSHIP_MISMATCH') + socket_inodes = set() + for descriptor in (process / 'fd').iterdir(): + try: target = os.readlink(descriptor) + except FileNotFoundError: continue + match = re.fullmatch(r'socket:\[(\d+)\]', target) + if match: socket_inodes.add(match.group(1)) + owned_ports = set() + for line in Path('/proc/net/tcp').read_text().splitlines()[1:]: + values = line.split() + if len(values) > 9 and values[3] == '0A' and values[9] in socket_inodes: + owned_ports.add(int(values[1].split(':')[1], 16)) + if not {state['proxyPort'], state['controlPort']}.issubset(owned_ports): raise RuntimeError('PROXY_LISTENERS_NOT_OWNED') + req = urllib.request.Request('http://127.0.0.1:' + str(state['controlPort']) + ('/mode' if mode else '/status'), + None if mode is None else json.dumps({'mode': mode, 'ttl': ttl}).encode(), + {'X-Order-Fault-Control': state['controlToken'], 'Content-Type': 'application/json'}) + with http_open(req, timeout=5) as response: return json.load(response) + + def probes(self, state): + """Real authenticated reads, not manufactured business responses. No body/secret is returned.""" + customer = self.scope['customerId']; results = {} + targets = {'IAM': '/internal/iam/tenants/' + customer + '/order-eligibility', + 'FINANCE': '/internal/finance/v1/wallet/accounts/' + customer} + for dependency, path in targets.items(): + req = urllib.request.Request(state['upstreams'][dependency] + path, + headers={'X-Internal-Service-Token': state['internalToken']}) + with http_open(req, timeout=8) as response: + value = json.load(response) + if response.status != 200 or str(value.get('code')) != '200': raise RuntimeError('UPSTREAM_PROBE_FAILED') + data = value.get('data') or {} + if dependency == 'IAM' and data.get('eligible') is not True: raise RuntimeError('CUSTOMER_NOT_ELIGIBLE') + if dependency == 'FINANCE' and (data.get('actualBalance') is None or float(data['actualBalance']) < 0): + raise RuntimeError('CUSTOMER_BALANCE_NOT_READY') + results[dependency] = True + return results + + def config_fingerprint(self, container, routing): + """Read the two actual Nacos imports using explicitly reviewed v1 endpoints. + No endpoint guessing/fallback and no Nacos write operation is supported. + """ + env = self.tools['env_of'](container); config = routing.get('nacos', {}) + base = urllib.parse.urlsplit(config.get('baseUrl', '')) + configured = urllib.parse.urlsplit(env.get('NACOS_SERVER_ADDR', '') if '://' in env.get('NACOS_SERVER_ADDR', '') + else 'http://' + env.get('NACOS_SERVER_ADDR', '')) + allowed_hosts = {configured.hostname, '127.0.0.1', 'localhost'} | { + network.get('Gateway') for network in container['NetworkSettings']['Networks'].values()} + if (base.scheme not in ('http', 'https') or base.hostname not in allowed_hosts or base.port != configured.port + or base.path not in ('', '/') or base.username or base.password or base.query or base.fragment + or config.get('loginPath') not in ('/nacos/v1/auth/users/login', '/nacos/v1/auth/login') + or config.get('configPath') != '/nacos/v1/cs/configs'): + raise RuntimeError('EXPLICIT_NACOS_READ_ENDPOINTS_REQUIRED') + if any(not env.get(key) for key in ['NACOS_USERNAME', 'NACOS_PASSWORD', 'NACOS_GROUP', 'NACOS_NAMESPACE']): + raise RuntimeError('NACOS_READ_IDENTITY_UNAVAILABLE') + origin = config['baseUrl'].rstrip('/') + request = urllib.request.Request(origin + config['loginPath'], urllib.parse.urlencode({ + 'username': env['NACOS_USERNAME'], 'password': env['NACOS_PASSWORD']}).encode(), + {'Content-Type': 'application/x-www-form-urlencoded'}) + with http_open(request, timeout=8) as response: token = json.load(response).get('accessToken') + if not isinstance(token, str) or not token: raise RuntimeError('NACOS_READ_LOGIN_FAILED') + fingerprints = {} + for data_id in ('baiyue-common.yml', 'order-service.yml'): + query = urllib.parse.urlencode({'dataId': data_id, 'group': env['NACOS_GROUP'], 'tenant': env['NACOS_NAMESPACE'], 'accessToken': token}) + with http_open(origin + config['configPath'] + '?' + query, timeout=8) as response: + if response.status != 200: raise RuntimeError('NACOS_SOURCE_READ_FAILED') + content = response.read() + if not content or content.lstrip().startswith((b'{', b'<')): raise RuntimeError('NACOS_SOURCE_NOT_YAML') + fingerprints[data_id] = hashlib.sha256(content).hexdigest() + return digest({'imports': fingerprints, 'namespace': env['NACOS_NAMESPACE'], 'group': env['NACOS_GROUP']}) + + 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:[0-9a-f]{64}', original['Image']): + raise RuntimeError('ORIGINAL_IMAGE_MISMATCH') + if original['State'].get('Health', {}).get('Status') != 'healthy': raise RuntimeError('ORDER_NOT_HEALTHY') + if original['Config']['Entrypoint'] != 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.get('routing', {}) + if routing.get('configurationReviewed') is not True: + raise RuntimeError('UPSTREAM_CONFIGURATION_REVIEW_REQUIRED') + source_digest = self.config_fingerprint(original, routing) + if routing.get('sourceSha256') != source_digest: raise RuntimeError('REVIEWED_NACOS_SOURCE_DIGEST_CHANGED') + upstreams = routing.get('upstreams', {}) + if set(upstreams) != set(PROPERTIES): raise RuntimeError('EXPLICIT_UPSTREAMS_REQUIRED') + for dependency, upstream in upstreams.items(): + url = urllib.parse.urlsplit(upstream) + target = self.tools['inspect'](UPSTREAM_CONTAINERS[dependency]); self.tools['require_test'](target, UPSTREAM_CONTAINERS[dependency]) + target_env = self.tools['env_of'](target) + ips = {network['IPAddress'] for name, network in target['NetworkSettings']['Networks'].items() + if name in original['NetworkSettings']['Networks']} + if (url.scheme != 'http' or url.username or url.password or url.path not in ('', '/') or url.query or url.fragment + or url.hostname not in ips or str(url.port) != target_env.get('SERVER_PORT') + or target_env.get('DEPLOY_ENV') != 'test' or target['State'].get('Health', {}).get('Status') != 'healthy'): + raise RuntimeError('UPSTREAM_NOT_VERIFIED_TEST_CONTAINER') + lock_fd = os.open(self.lock, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(lock_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, + 'proxyPort': proxy_port, 'controlPort': control_port, 'controlToken': secrets.token_urlsafe(32), + 'upstreams': {k: v.rstrip('/') for k, v in upstreams.items()}, 'routingSourceSha256': source_digest, 'routing': routing, + 'orderIps': [x['IPAddress'] for x in original['NetworkSettings']['Networks'].values()], 'mutation': None, 'proxyLaunchIntent': False} + self.save(state) # Recovery info exists before any process or container mutation. + self.probes(state) + proxy = self.directory / 'proxy.py'; proxy.write_bytes(base64.b64decode(config['proxySource'])); os.chmod(proxy, 0o600) + private_write(self.directory / 'proxy-config.json', {k: state[k] for k in + ['internalToken', 'controlToken', 'proxyPort', 'controlPort', 'upstreams', 'orderIps']} | {'customerId': self.scope['customerId'], + 'requestIds': ['dep-' + self.scope['marker'] + '-' + dependency + '-' + role + for dependency in ('IAM', 'FINANCE') for role in ('merchant', 'platform')]}) + 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(proxy), '--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 summary(self, state): + return {'scope': self.scope, 'phase': state['phase'], 'originalImageId': state['originalImageId'], + 'originalFingerprint': state['originalFingerprint'], 'uncertain': bool(state.get('uncertain')), + 'mutationStep': (state.get('mutation') or {}).get('step'), 'restored': state.get('restored', False)} + + def mutation(self, state, step, method, path, body=None): + if state.get('uncertain'): raise RuntimeError('UNCERTAIN_MUTATION_REQUIRES_RECONCILE') + state['mutation'] = {'step': step, 'path': path, 'bodySha256': digest(body), 'intentAt': time.time()}; self.save(state) + try: + result = self.tools['docker'](method, path, body) + if step == 'CREATE_INTENT' and (not isinstance(result, dict) or not re.fullmatch(r'[A-Za-z0-9_-]{1,128}', result.get('Id', ''))): + raise RuntimeError('INVALID_CREATE_ACK') + except Exception: + state['uncertain'] = True; self.save(state) + raise RuntimeError('UNCERTAIN_DOCKER_MUTATION') + state['mutation']['step'] = step.replace('_INTENT', '_DONE') + if step == 'CREATE_INTENT': state['createdId'] = result['Id'] + self.save(state) + return result + + def replace(self, state, current, desired, purpose): + if state.get('uncertain'): raise RuntimeError('UNCERTAIN_MUTATION_REQUIRES_RECONCILE') + if self.container()['Id'] != current['Id']: raise RuntimeError('CONCURRENT_ORDER_CHANGE') + image = self.tools['docker']('GET', '/images/' + urllib.parse.quote(desired['Image'], safe='') + '/json') + if image['Id'] != state['originalImageId']: raise RuntimeError('IMAGE_TAG_MOVED_NO_RECREATE') + state['purpose'] = purpose; state['desiredBody'] = desired; state['replacingId'] = current['Id']; self.save(state) + if current['State']['Running']: + self.mutation(state, 'STOP_INTENT', 'POST', '/containers/' + current['Id'] + '/stop?t=15') + self.mutation(state, 'DELETE_INTENT', 'DELETE', '/containers/' + current['Id'] + '?v=false') + created = self.mutation(state, 'CREATE_INTENT', 'POST', '/containers/create?name=' + ORDER, desired) + self.mutation(state, 'START_INTENT', 'POST', '/containers/' + created['Id'] + '/start') + state['currentId'] = created['Id']; state['phase'] = purpose; state['mutation'] = None; self.save(state) + + def matches_injected(self, current, state): + expected = copy.deepcopy(state['originalBody']['Entrypoint']) + expected[2] += ''.join(' --' + prop + '=http://host.docker.internal:' + str(state['proxyPort']) + '/' + name.lower() + for name, prop in PROPERTIES.items()) + restored_view = copy.deepcopy(current) + restored_view['Config']['Entrypoint'] = state['originalBody']['Entrypoint'] + return (current['Image'] == state['originalImageId'] and current['Config']['Entrypoint'] == expected + and self.tools['fingerprint'](restored_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'][2] += ''.join(' --' + prop + '=http://host.docker.internal:' + str(state['proxyPort']) + '/' + name.lower() + for name, prop in PROPERTIES.items()) + self.replace(state, current, desired, 'INJECTED') + state['injectedFingerprint'] = self.tools['fingerprint'](self.container()); self.save(state) + return self.summary(state) + + def restore(self, state): + # PASS is attempted even if Docker state is ambiguous. Never stop the proxy here. + try: self.control(state, 'PASS') + except Exception: + state['passUnconfirmed'] = True; self.save(state) + if state.get('uncertain'): raise RuntimeError('UNCERTAIN_MUTATION_REQUIRES_RECONCILE') + if state['phase'] == 'ABSENT_CONFIRMED': + try: self.tools['inspect'](ORDER) + except RuntimeError as error: + if str(error) != 'DOCKER_404': raise + else: raise RuntimeError('ORDER_REAPPEARED_NO_RECREATE') + body = copy.deepcopy(state['originalBody']) + image = self.tools['docker']('GET', '/images/' + urllib.parse.quote(body['Image'], safe='') + '/json') + if image['Id'] != state['originalImageId']: raise RuntimeError('IMAGE_TAG_MOVED_NO_RECREATE') + state['purpose'] = 'ORIGINAL_RESTORED'; state['desiredBody'] = body; self.save(state) + created = self.mutation(state, 'CREATE_INTENT', 'POST', '/containers/create?name=' + ORDER, body) + self.mutation(state, 'START_INTENT', 'POST', '/containers/' + created['Id'] + '/start') + state['currentId'] = created['Id']; state['phase'] = 'ORIGINAL_RESTORED'; state['mutation'] = None; self.save(state) + return self.summary(state) + current = self.container() + if self.tools['fingerprint'](current) == state['originalFingerprint'] and current['Image'] == state['originalImageId']: + if not current['State']['Running']: + state['purpose'] = 'ORIGINAL_RESTORED'; state['createdId'] = current['Id']; self.save(state) + self.mutation(state, 'START_INTENT', 'POST', '/containers/' + current['Id'] + '/start') + state['phase'] = 'ORIGINAL_RESTORED'; self.save(state); return self.summary(state) + if current['Id'] != state.get('currentId') or not self.matches_injected(current, state): + raise RuntimeError('UNRECOGNIZED_ORDER_CONFIGURATION') + self.replace(state, current, copy.deepcopy(state['originalBody']), 'ORIGINAL_RESTORED') + return self.summary(state) + + def reconcile(self, state): + # This action is intentionally read-only with respect to Docker. It never resumes deletion/creation. + if not state.get('uncertain'): return self.summary(state) + try: current = self.container() + except RuntimeError as error: + if str(error) != 'DOCKER_404' or (state.get('mutation') or {}).get('step') != 'DELETE_INTENT': + raise RuntimeError('MANUAL_DOCKER_RECOVERY_REQUIRED_NO_BLIND_RECREATE') + try: self.tools['inspect'](state['replacingId']) + except RuntimeError as by_id: + if str(by_id) != 'DOCKER_404': raise + else: raise RuntimeError('ORIGINAL_ID_STILL_PRESENT') + state['uncertain'] = False; state['phase'] = 'ABSENT_CONFIRMED'; state['mutation'] = None; self.save(state) + return self.summary(state) + step = (state.get('mutation') or {}).get('step') + original = current['Image'] == state['originalImageId'] and self.tools['fingerprint'](current) == state['originalFingerprint'] + injected = self.matches_injected(current, state) + desired_matches = injected if state.get('purpose') == 'INJECTED' else original + if step == 'STOP_INTENT' and current['Id'] == state.get('replacingId') and not current['State']['Running'] and (original or injected): + phase = 'ORIGINAL_RESTORED' if original else 'INJECTED' + elif step == 'CREATE_INTENT' and current['Id'] != state.get('replacingId') and desired_matches: + phase = state['purpose'] + elif step == 'START_INTENT' and current['Id'] == state.get('createdId') and current['State']['Running'] and desired_matches: + phase = state['purpose'] + else: + raise RuntimeError('MANUAL_DOCKER_RECOVERY_REQUIRED_NO_BLIND_RECREATE') + state['uncertain'] = False; state['currentId'] = current['Id']; state['phase'] = phase; state['mutation'] = None + if phase == 'INJECTED': state['injectedFingerprint'] = self.tools['fingerprint'](current) + self.save(state); return self.summary(state) + + def verify_restored(self, state): + current = self.container() + if state.get('uncertain') or current['Image'] != state['originalImageId'] or self.tools['fingerprint'](current) != state['originalFingerprint']: + raise RuntimeError('RESTORED_CONFIG_OR_IMAGE_MISMATCH') + if current['State'].get('Health', {}).get('Status') != 'healthy': raise RuntimeError('RESTORED_NOT_HEALTHY') + if self.config_fingerprint(current, state['routing']) != state['routingSourceSha256']: + raise RuntimeError('NACOS_ROUTE_CONFIGURATION_CHANGED') + checks = self.probes(state) + if checks != {'IAM': True, 'FINANCE': True}: raise RuntimeError('RESTORED_ROUTE_PROBE_INVALID') + state['routeReadsVerified'] = checks; self.save(state) + return {'healthy': True, 'originalFingerprint': state['originalFingerprint'], 'currentFingerprint': self.tools['fingerprint'](current), + 'originalImageId': state['originalImageId'], 'currentImageId': current['Image'], 'upstreamReads': checks, + 'nacosSourcesUnchanged': True} + + def finish(self, state, business): + self.verify_restored(state) + if self.lock.is_symlink() or self.lock.read_text() != self.scope['marker']: raise RuntimeError('RUNTIME_LOCK_OWNER_CHANGED') + if set(business or {}) != {'merchant', 'platform'} or any( + business[role].get('orderId') != order_id or business[role].get('status') != 'SUBMITTED' + or business[role].get('sameKeyReplayVerified') is not True for role, order_id in self.scope['orderIds'].items()): + raise RuntimeError('RESTORED_ORDER_ROUTES_NOT_PROVED') + pid = state.get('proxyPid') + if not pid and state.get('proxyLaunchIntent'): + pid_file = self.directory / 'proxy.pid' + if pid_file.is_symlink() or not pid_file.exists() or not pid_file.read_text().isdigit(): raise RuntimeError('PROXY_PID_UNCERTAIN') + pid = int(pid_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(0.1) + else: raise RuntimeError('PROXY_NOT_STOPPED') + state['restored'] = True; state['phase'] = 'FINISHED' + # Recovery snapshots remain protected until completion; erase secrets only after verified restoration. + for field in ['originalBody', 'desiredBody', 'internalToken', 'controlToken']: state.pop(field, None) + self.save(state) + config = self.directory / 'proxy-config.json' + if config.exists(): config.unlink() + self.lock.unlink() + return {'restored': True, 'healthy': True, 'proxyStopped': True, 'originalImageId': state['originalImageId'], + 'originalFingerprint': state['originalFingerprint'], 'orderRoutesVerified': True} + + +def execute(config): + runtime = Runtime(config['scope'], common_tools(base64.b64decode(config['commonSource']).decode())) + action = config['action'] + if action == 'preflight': + current = runtime.container() + return {'originalImageId': current['Image'], 'originalFingerprint': runtime.tools['fingerprint'](current), + 'sourceSha256': runtime.config_fingerprint(current, config['routing']), 'runtimeChanged': False} + 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 action not in ('records', 'health') and (runtime.lock.is_symlink() or runtime.lock.read_text() != runtime.scope['marker']): + raise RuntimeError('RUNTIME_LOCK_OWNER_CHANGED') + if action == 'records': + path = runtime.directory / 'proxy-evidence.jsonl' + return {'records': [json.loads(line) for line in path.read_text().splitlines()] if path.exists() else []} + if action == 'activate': return runtime.activate(state) + if action == 'mode': + if state['phase'] != 'INJECTED' or state.get('uncertain'): raise RuntimeError('FAULT_WINDOW_NOT_READY') + return runtime.control(state, config['mode'], config.get('ttl', 0)) + if action == 'health': return {'healthy': runtime.container()['State'].get('Health', {}).get('Status') == 'healthy'} + if action == 'restore': return runtime.restore(state) + if action == 'reconcile': return runtime.reconcile(state) + if action == 'verify-restored': return runtime.verify_restored(state) + if action == 'finish': return runtime.finish(state, config.get('businessRecovery')) + raise RuntimeError('UNKNOWN_ACTION') + + +def main(config): + # A lost Jenkins response must not allow a second process to race the first Docker mutation. + import fcntl + validate_scope(config['scope']) + path = Path('/tmp') / ('chongchong-order-dependency-' + config['scope']['marker'] + '.action.lock') + if path.is_symlink(): raise RuntimeError('UNSAFE_ACTION_LOCK') + fd = os.open(path, 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: + result = main(json.loads(base64.b64decode(sys.argv[1]))) + print('ORDER_DEPENDENCY_JSON=' + json.dumps({'ok': True, 'data': result})) + except Exception as error: + message = str(error) + print('ORDER_DEPENDENCY_JSON=' + json.dumps({'ok': False, 'error': message if re.fullmatch(r'[A-Z_0-9]{1,90}', message) else 'ORDER_DEPENDENCY_ACTION_FAILED'}))