diff --git a/scripts/fleet-test-env-acceptance.mjs b/scripts/fleet-test-env-acceptance.mjs index 0bf83fb87..30ed00d3a 100644 --- a/scripts/fleet-test-env-acceptance.mjs +++ b/scripts/fleet-test-env-acceptance.mjs @@ -2,6 +2,7 @@ import { mkdirSync, writeFileSync, openSync, closeSync, unlinkSync, readFileSync import { resolve, dirname, basename, relative, isAbsolute, join } from 'node:path' import { fileURLToPath, pathToFileURL } from 'node:url' import { createHash, randomUUID } from 'node:crypto' +import { AsyncLocalStorage } from 'node:async_hooks' import { AcceptanceRun, parseArguments, validateExecutionOptions, buildAcceptanceSummary } from './admin-test-env-acceptance.mjs' import { redactEvidence } from './fba-core-regression-test-env.mjs' import { createTestEvidenceReaders } from './fba-core-regression-evidence.mjs' @@ -24,17 +25,35 @@ const success = response => response?.http === 200 && id(response?.body?.code) = const key = () => `fleet-${randomUUID()}` /** Install before login/prerequisites. A caught unknown write must not permit another request. */ -export function guardFleetWrites(context, checkpoint) { +export function guardFleetWrites(context, checkpoint, { guardFetch = false } = {}) { let stopped = null, sequence = 0 - for (const name of ['request', 'requestWithBody', 'requestBinary']) { - const original = context[name] + const scope = new AsyncLocalStorage(), originalFetch = globalThis.fetch + const direct = { async requestWithBody(method, path, body, token, headers, request) { + const nativeResponse = await originalFetch(request), copy = nativeResponse.clone() + const content = await copy.text() + let payload = null + try { payload = content ? JSON.parse(content) : null } catch { /* Keep bounded diagnostic only. */ } + return { nativeResponse, http: nativeResponse.status, body: payload, raw: payload ? null : content.slice(0, 500), + requestId: payload?.requestId || nativeResponse.headers.get('x-request-id'), + traceId: payload?.traceId || nativeResponse.headers.get('x-trace-id') } + } } + const targets = ['request', 'requestWithBody', 'requestBinary'].map(name => [context, name]) + if (guardFetch) targets.push([direct, 'requestWithBody']) + for (const [owner, name] of targets) { + const original = owner[name] if (typeof original !== 'function') continue - context[name] = async function (method, path, body, token, headers) { + owner[name] = async function (method, path, body, token, headers) { if (stopped) throw stopped if (['GET', 'HEAD', 'OPTIONS'].includes(String(method).toUpperCase())) - return original.apply(this, arguments) + return scope.run(true, () => original.apply(this, arguments)) const entry = { sequence: ++sequence, method, path, status: 'SENDING', startedAt: new Date().toISOString(), headers, body } + let session = owner === direct ? String(headers?.authorization || '').replace(/^Bearer\s+/i, '') + : token === undefined ? this.token : token + if (owner !== direct && this.pdaToken && /\/pda-loading\/tasks\/\d+/.test(String(path))) session = this.pdaToken + entry.actor = session && session === context.pdaToken ? { kind: 'PDA', source: 'CONTEXT_PDA_SESSION' } + : session && session === context.token && context.options?.username ? { kind: 'PLATFORM', source: 'CONTEXT_ADMIN_SESSION' } + : { kind: 'UNKNOWN', source: session ? 'UNRESOLVED_SESSION' : 'NO_SESSION' } // Multipart delegates to requestWithBody, so it is journaled exactly once, after construction. if (body instanceof FormData) { entry.body = { multipart: [] } @@ -44,31 +63,60 @@ export function guardFleetWrites(context, checkpoint) { } try { await checkpoint(entry) } catch (error) { stopped = error; throw error } try { - const response = await original.apply(this, arguments) + const response = await scope.run(true, () => original.apply(this, arguments)) const http = response?.http, payload = response?.body, code = id(payload?.code) + const binary = name === 'requestBinary' && response?.bytes instanceof Uint8Array + const pdf = binary && http === 200 && /^application\/pdf(?:\s*;|$)/i.test(response.contentType || '') + && response.bytes.length > 4 && Buffer.from(response.bytes.subarray(0, 5)).toString('ascii') === '%PDF-' const definite = Number.isInteger(http) && http >= 200 && http < 500 - && payload && typeof payload === 'object' && !Array.isArray(payload) && response.requestId && response.traceId - && (http === 200 && code === '200' || /^[A-Z]\d{4}$/.test(code) && (http === 200 || http >= 400)) + && response.requestId && response.traceId + && (pdf || payload && typeof payload === 'object' && !Array.isArray(payload) + && (name !== 'requestBinary' && http === 200 && code === '200' || /^[A-Z]\d{4}$/.test(code) && (http === 200 || http >= 400))) Object.assign(entry, { status: definite ? 'RESPONDED' : 'UNKNOWN_RESULT', http, code, - requestId: response?.requestId, traceId: response?.traceId, finishedAt: new Date().toISOString() }) + requestId: response?.requestId, traceId: response?.traceId, raw: boundedFleetDiagnostic(response?.raw), + responseBody: redactFleetWrite(response?.body), finishedAt: new Date().toISOString() }) + if (binary) entry.binary = { contentType: response.contentType, bytes: response.bytes.length, + sha256: createHash('sha256').update(response.bytes).digest('hex'), validPdf: Boolean(pdf) } if (!definite) stopped = Error(`Unknown write outcome: ${method} ${path}; preserve original request, do not resend`) await checkpoint(entry) if (stopped) throw stopped return response } catch (error) { - stopped ||= error + stopped ||= Error(boundedFleetDiagnostic(error instanceof Error ? error.message : String(error)) || 'Fleet write failed') if (entry.status === 'SENDING') { entry.status = 'UNKNOWN_RESULT'; entry.finishedAt = new Date().toISOString() + entry.raw = boundedFleetDiagnostic(error instanceof Error ? error.message : String(error)) await checkpoint(entry) } throw stopped } } } + if (guardFetch) globalThis.fetch = async (input, init) => { + if (scope.getStore()) return originalFetch(input, init) + if (stopped) throw stopped + const request = new Request(input, init), url = new URL(request.url) + if (url.origin !== new URL(context.options.gateway).origin) return originalFetch(input, init) + if (['GET', 'HEAD', 'OPTIONS'].includes(request.method)) return originalFetch(input, init) + const type = request.headers.get('content-type') || '' + let body + if (request.body) body = type.includes('multipart/form-data') ? await request.clone().formData() + : type.includes('application/json') ? await request.clone().json() : await request.clone().text() + const result = await direct.requestWithBody(request.method, url.pathname + url.search, body, undefined, + Object.fromEntries(request.headers), request) + return result.nativeResponse + } + return () => { if (guardFetch) globalThis.fetch = originalFetch } +} + +function boundedFleetDiagnostic(value) { + return typeof value === 'string' ? value.replace(/Bearer\s+[^\s"'<>]+/gi, 'Bearer [REDACTED]') + .replace(/((?:password|accessToken|refreshToken|authorization)["']?\s*[=:]\s*["']?)[^\s,;"'<>]+/gi, '$1[REDACTED]') + .replace(/https?:\/\/[^\s"'<>]+/gi, '[REDACTED_URL]').slice(0, 500) : null } export function redactFleetWrite(value, secrets = []) { - const sensitive = name => name !== 'draftToken' && /password|authorization|cookie|privateKey|token$|^(?:loginId|username)$/i.test(name) + const sensitive = name => name !== 'draftToken' && /password|authorization|cookie|privateKey|api.?key|secret|token$|^(?:loginId|username)$/i.test(name) if (Array.isArray(value)) return value.map(item => redactFleetWrite(item, secrets)) if (value && typeof value === 'object' && typeof value.field === 'string' && sensitive(value.field)) return { field: value.field, value: '[REDACTED]' } @@ -531,7 +579,7 @@ export async function runFleetAcceptance(config, { env = process.env, readers, c if (!rel || rel.startsWith('..') || isAbsolute(rel)) throw Error('Report scope invalid') mkdirSync(resolve(root, 'reports'), { recursive: true }) const lockPath = resolve(root, 'reports/.fleet-acceptance.lock'), lock = openSync(lockPath, 'wx') - let context, initialized = false, report + let context, initialized = false, report, restoreFetch const secrets = Object.entries(env).filter(([name]) => /PASSWORD|TOKEN|USERNAME|JENKINS_USER/.test(name)).map(([,value]) => value) const save = (name, value) => writeFileSync(join(output, name), JSON.stringify(redactEvidence(value, [...secrets, context?.token, context?.pdaToken, context?.refreshToken, context?.context?.merchantQualificationFixture?.password]), null, 2) + '\n') @@ -556,13 +604,13 @@ export async function runFleetAcceptance(config, { env = process.env, readers, c markerNamespace: createHash('sha256').update(config.runId).digest('hex').slice(0, 12) }) context = createContext ? createContext(options) : new AcceptanceRun(options) let writeCheckpoint = 0 - guardFleetWrites(context, entry => { + restoreFetch = guardFleetWrites(context, entry => { const payload = JSON.stringify(redactFleetWrite({ marker: context.marker, runId: config.runId, request: entry }, [...secrets, context.token, context.pdaToken, context.refreshToken]), null, 2) + '\n' const descriptor = openSync(join(output, 'evidence', `write-${String(++writeCheckpoint).padStart(6, '0')}.json`), 'wx') try { writeFileSync(descriptor, payload); fsyncSync(descriptor) } finally { closeSync(descriptor) } - }) + }, { guardFetch: true }) context.context.fleetRunId = config.runId context.context.fleetScenario = config.scenario || 'handover' context.context.saveFleetHandoverFixture = fixture => save('handover-fixture.json', fixture) @@ -616,6 +664,7 @@ export async function runFleetAcceptance(config, { env = process.env, readers, c summary: context ? buildAcceptanceSummary(context.steps, context.modules, context.coverageGaps) : { failed: 1 }, steps: context?.steps || [], modules: context?.modules || [], createdResources: context?.createdResources || {} } } finally { + restoreFetch?.() if (initialized) save('report.json', report || { success: false, fatal: 'RUN_INTERRUPTED' }) closeSync(lock); unlinkSync(lockPath) } diff --git a/scripts/fleet-test-env-acceptance.test.mjs b/scripts/fleet-test-env-acceptance.test.mjs index 6ed19e498..0e28c93ba 100644 --- a/scripts/fleet-test-env-acceptance.test.mjs +++ b/scripts/fleet-test-env-acceptance.test.mjs @@ -3,6 +3,8 @@ import assert from 'node:assert/strict' import { mkdtempSync, mkdirSync, readFileSync, existsSync, rmSync } from 'node:fs' import { tmpdir } from 'node:os' import { join, dirname, resolve } from 'node:path' +import { AcceptanceRun } from './admin-test-env-acceptance.mjs' +import { run as runInboundMeasurement } from './admin-test-env-modules/inbound-measurement.mjs' import { parseFleetArguments, fleetExecutionPlan, runFleetAcceptance, reviewedFullCheckpoint, verifyFleetTerminalState, runFleetHandoverStage, fleetCostShares, prepareFleetCostActors, guardFleetWrites, redactFleetWrite } from './fleet-test-env-acceptance.mjs' for (const response of [ @@ -35,6 +37,93 @@ test('formal correlated rejection is recorded and remains available for expected await ctx.request('GET', '/readback'); assert.equal(saved.length, 2) }) +test('response evidence preserves business messages and returned resource IDs with explicit session provenance and no tokens', async () => { + const saved = [], ctx = { token: 'admin-session', pdaToken: 'pda-session', options: { username: 'test-admin' }, + async request() { return { http: 200, body: { code: '200', message: 'created', data: { + id: '123', businessNo: 'FIXTURE-123', accessToken: 'new-secret', refreshToken: 'refresh-secret' } }, requestId: 'r', traceId: 't' } } } + guardFleetWrites(ctx, entry => saved.push(redactFleetWrite(structuredClone(entry)))) + await ctx.request('POST', '/create', { expectedVersion: 3 }) + assert.deepEqual(saved[0].actor, { kind: 'PLATFORM', source: 'CONTEXT_ADMIN_SESSION' }) + assert.equal(saved[1].responseBody.message, 'created'); assert.equal(saved[1].responseBody.data.id, '123') + await ctx.request('POST', '/pda', {}, 'pda-session'); assert.equal(saved[2].actor.kind, 'PDA') + await ctx.request('POST', '/other', {}, 'unresolved-session'); assert.equal(saved[4].actor.kind, 'UNKNOWN') + await ctx.request('POST', '/login', {}, ''); assert.equal(saved[6].actor.source, 'NO_SESSION') + for (const secret of ['admin-session', 'pda-session', 'new-secret', 'refresh-secret', 'unresolved-session']) + assert.ok(!JSON.stringify(saved).includes(secret)) +}) + +test('POST carton-label PDF is a definite binary response with durable hash and permits same-fixture readback', async () => { + const saved = [], calls = [], response = { http: 200, body: null, bytes: Buffer.from('%PDF-1.7\nfixture'), + contentType: 'application/pdf', requestId: 'r', traceId: 't' } + const ctx = { async requestBinary(method, path) { + assert.equal(saved.at(-1).status, 'SENDING'); calls.push(path); return response + }, async request() { calls.push('readback') } } + guardFleetWrites(ctx, entry => saved.push(structuredClone(entry))) + const path = '/api/platform/inbounds/1/carton-label-pdf?tenantId=2&cartonIndex=1' + assert.equal(await ctx.requestBinary('POST', path), response) + assert.equal(saved[0].path, path); assert.equal(saved[1].status, 'RESPONDED') + assert.equal(saved[1].binary.bytes, response.bytes.length); assert.match(saved[1].binary.sha256, /^[a-f0-9]{64}$/) + assert.equal(saved[1].binary.validPdf, true) + await ctx.request('GET', '/api/platform/inbounds/1'); assert.deepEqual(calls, [path, 'readback']) +}) + +test('real inbound prerequisite and AcceptanceRun binary transport accept three PDF writes then stop raw multipart retries', async t => { + const calls = [], saved = [], summary = { id: '1', tenantId: '2', processStatus: 'WAIT_MEASURE', inboundCode: 'FIXTURE-INBOUND', totalPackageCount: 2 } + const native = t.mock.method(globalThis, 'fetch', async (input, init) => { + const req = new Request(input, init), path = new URL(req.url).pathname + calls.push({ method: req.method, path }) + const headers = { 'x-request-id': 'r', 'x-trace-id': 't' } + if (req.method === 'GET') return new Response(JSON.stringify({ code: '200', data: { summary } }), { headers: { ...headers, 'content-type': 'application/json' } }) + assert.equal(saved.at(-1).status, 'SENDING') + if (path.endsWith('/carton-label-pdf')) return new Response('%PDF-1.7\nfixture', { headers: { ...headers, 'content-type': 'application/pdf' } }) + assert.ok(path.endsWith('/measurement/carton-scans')) + throw Error('transport failure Bearer sensitive-session https://example.invalid/private') + }) + const context = new AcceptanceRun({ gateway: 'https://test.invalid' }) + context.createdResources.orders = { marker: context.marker, inboundId: '1', tenantId: '2' } + const restore = guardFleetWrites(context, entry => saved.push(redactFleetWrite(structuredClone(entry))), { guardFetch: true }) + try { + await runInboundMeasurement(context) + assert.equal(calls.filter(row => row.path.endsWith('/carton-label-pdf')).length, 3) + assert.equal(calls.filter(row => row.path.endsWith('/measurement/carton-scans')).length, 1, 'raw helper retries must not reach transport') + const pdfs = saved.filter(row => row.binary?.validPdf) + assert.equal(pdfs.length, 3); assert.ok(pdfs.every(row => row.status === 'RESPONDED')) + assert.equal(context.steps.find(row => row.name === '生成箱唛 PDF 并冻结稳定箱码').passed, true) + const last = saved.at(-1); assert.equal(last.status, 'UNKNOWN_RESULT') + assert.ok(last.body.multipart.some(row => row.field === 'data' && row.base64)) + assert.ok(last.raw.includes('transport failure')); assert.ok(!last.raw.includes('sensitive-session')); assert.ok(!last.raw.includes('example.invalid')) + await assert.rejects(context.request('POST', '/subsequent'), /transport failure/) + } finally { restore(); assert.equal(globalThis.fetch, native); t.mock.restoreAll() } +}) + +test('raw JSON prerequisite writes retain headers, original version and bounded diagnostics without duplicate journaling', async t => { + let calls = 0 + t.mock.method(globalThis, 'fetch', async () => { calls++; return new Response('x'.repeat(600), { status: 502 }) }) + const saved = [], context = new AcceptanceRun({ gateway: 'https://test.invalid' }) + const restore = guardFleetWrites(context, entry => saved.push(redactFleetWrite(structuredClone(entry))), { guardFetch: true }) + try { + await assert.rejects(fetch('https://test.invalid/approve', { method: 'PUT', headers: { 'content-type': 'application/json', + Authorization: 'Bearer secret', 'Idempotency-Key': 'same-key' }, body: JSON.stringify({ expectedVersion: 6 }) }), /Unknown write outcome/) + assert.equal(saved.length, 2); assert.equal(saved[0].body.expectedVersion, 6) + assert.equal(saved[0].headers['idempotency-key'], 'same-key'); assert.equal(saved[0].headers.authorization, '[REDACTED]') + assert.equal(saved[1].raw.length, 500) + await assert.rejects(fetch('https://test.invalid/retry', { method: 'POST' }), /Unknown write outcome/) + assert.equal(calls, 1) + } finally { restore(); t.mock.restoreAll() } +}) + +for (const override of [{ http: 0 }, { http: 503 }, { bytes: Buffer.from('proxy') }, + { bytes: new Uint8Array() }, { contentType: 'text/html' }, { traceId: null }, + { contentType: 'application/json', bytes: Buffer.from('{"code":"200"}'), body: { code: '200' } }]) + test(`binary write still stops on unknown or invalid response: ${JSON.stringify(override)}`, async () => { + const saved = [], ctx = { async requestBinary() { return { http: 200, body: null, bytes: Buffer.from('%PDF-1.7'), + contentType: 'application/pdf', requestId: 'r', traceId: 't', ...override } }, async request() { assert.fail('must stop') } } + guardFleetWrites(ctx, entry => saved.push(structuredClone(entry))) + await assert.rejects(ctx.requestBinary('POST', '/carton-label-pdf'), /Unknown write outcome/) + assert.equal(saved[1].status, 'UNKNOWN_RESULT') + await assert.rejects(ctx.request('GET', '/readback'), /Unknown write outcome/) + }) + test('checkpoint failure stops before dispatch; transport throw stops even after caller catches', async () => { let calls = 0 const ctx = { async request() { calls++; throw Error('transport failure') } }