test(order): add isolated dependency failure and recovery verification
This commit is contained in:
@@ -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；环境执行仍需本轮写入授权。
|
||||
|
||||
@@ -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 = []
|
||||
|
||||
@@ -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')
|
||||
|
||||
|
||||
@@ -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<rest.length;i+=2){assert(rest[i]?.startsWith('--') && rest[i+1] && !rest[i+1].startsWith('--'),'Use explicit --name value arguments');args[rest[i].slice(2)]=rest[i+1]}
|
||||
assert(['plan','preflight','run','status','reconcile','restore'].includes(mode),'Use plan/preflight/run/status/reconcile/restore')
|
||||
assert(args.plan,'--plan required');const plan=JSON.parse(readFileSync(args.plan,'utf8')),scope=scopeFor(plan)
|
||||
if(mode==='plan'){console.log(JSON.stringify({scope,actions:['prepare','activate','IAM x 2','FINANCE x 2','finally restore','verify original routes','finish proxy'],requiresExclusiveOrderWindow:true},null,2));return}
|
||||
assert(args['confirm-test-runtime']==='yes','--confirm-test-runtime yes required')
|
||||
const action=createOrderFaultRuntime(process.env,scope)
|
||||
if(mode==='preflight'){assert(args.routing,'--routing required');console.log(JSON.stringify(await action('preflight',{routing:JSON.parse(readFileSync(args.routing,'utf8'))}),null,2));return}
|
||||
if(['status','reconcile'].includes(mode)){console.log(JSON.stringify(await action(mode),null,2));return}
|
||||
assert(args.report && process.env.ADMIN_TEST_GATEWAY,'--report and ADMIN_TEST_GATEWAY required')
|
||||
await mkdir(dirname(resolve(args.report)),{recursive:true})
|
||||
const reportFile=await open(resolve(args.report),'wx',0o600) // Never overwrite a previous run or recovery report.
|
||||
const business=businessClient(process.env,plan);let result
|
||||
try {
|
||||
await business.login()
|
||||
if(mode==='run') {
|
||||
assert(args.routing && /^sha256:[0-9a-f]{64}$/.test(args['expected-image']||''),'--routing and immutable --expected-image required')
|
||||
result=await runOrderFaultWindow({plan,routing:JSON.parse(readFileSync(args.routing,'utf8')),expectedImageId:args['expected-image'],action,...business})
|
||||
} else {
|
||||
await action('restore')
|
||||
for(let i=0;i<180;i++){if((await action('health')).healthy)break;assert(i<179,'RESTORE_HEALTH_TIMEOUT');await pause(1000)}
|
||||
const runtime=await action('verify-restored'),businessRecovery=await business.recovery()
|
||||
result={success:true,runtime,businessRecovery,restoration:await action('finish',{businessRecovery})}
|
||||
}
|
||||
} catch(error){result={success:false,error:/^[A-Z_0-9 :.-]+$/.test(error.message)?error.message:'EXECUTION_FAILED_USE_RUNTIME_STATUS'}}
|
||||
finally {
|
||||
const sessionsClosed=await business.logout();result ??={success:false};result.scope=scope;result.sessionsClosed=sessionsClosed
|
||||
if(!sessionsClosed)result.success=false
|
||||
try {await reportFile.writeFile(JSON.stringify(result,null,2)+'\n');await reportFile.sync()}
|
||||
finally {await reportFile.close()}
|
||||
}
|
||||
console.log(JSON.stringify({success:result.success,report:resolve(args.report),restored:result.restoration?.restored===true}))
|
||||
if(!result.success)process.exitCode=1
|
||||
}
|
||||
|
||||
if(process.argv[1] && import.meta.url===pathToFileURL(resolve(process.argv[1])).href)main(process.argv.slice(2)).catch(()=>{console.error('ORDER_DEPENDENCY_ARGUMENT_OR_CONFIGURATION_ERROR');process.exitCode=1})
|
||||
@@ -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')<calls.indexOf('verify-restored'))
|
||||
assert.ok(calls.indexOf('verify-restored')<calls.indexOf('business-recovery'))
|
||||
assert.ok(calls.indexOf('business-recovery')<calls.indexOf('finish'))
|
||||
})
|
||||
test('failed business assertion still restores, but never reports fault verification success',async()=>{
|
||||
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/)
|
||||
})
|
||||
@@ -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)
|
||||
@@ -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/)
|
||||
})
|
||||
@@ -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=[]), '<reviewed-docker-primitives>', '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'}))
|
||||
Reference in New Issue
Block a user