forked from wangziqi/gongxue-base
472 lines
19 KiB
JavaScript
472 lines
19 KiB
JavaScript
import assert from 'node:assert/strict';
|
|
import pg from 'pg';
|
|
import { spawn } from 'node:child_process';
|
|
import http from 'node:http';
|
|
|
|
const databaseUrl = process.env.DATABASE_URL || 'postgresql://postgres:postgres@127.0.0.1:54322/postgres';
|
|
const tenantId = '00000000-0000-0000-0000-000000000001';
|
|
|
|
const ids = {
|
|
okAsset: '20000000-0000-0000-0000-000000000801',
|
|
badAsset: '20000000-0000-0000-0000-000000000802',
|
|
scanBadAsset: '20000000-0000-0000-0000-000000000803',
|
|
httpPassAsset: '20000000-0000-0000-0000-000000000804',
|
|
httpFailAsset: '20000000-0000-0000-0000-000000000805',
|
|
httpUnavailableAsset: '20000000-0000-0000-0000-000000000806',
|
|
};
|
|
|
|
const checksumA = 'a'.repeat(64);
|
|
const checksumB = 'b'.repeat(64);
|
|
|
|
async function runWorkerOnce(extraEnv = {}) {
|
|
const child = spawn(process.execPath, ['apps/worker/dist/apps/worker/src/index.js', '--once', '--job', 'assets'], {
|
|
cwd: process.cwd(),
|
|
env: {
|
|
...process.env,
|
|
DATABASE_URL: databaseUrl,
|
|
WORKER_ASSET_BATCH_SIZE: '20',
|
|
WORKER_ASSET_MIN_AGE_SECONDS: '0',
|
|
WORKER_ASSET_RECHECK_INTERVAL_SECONDS: '86400',
|
|
WORKER_ASSET_REQUEST_TIMEOUT_MS: '5000',
|
|
WORKER_ASSET_SECURITY_SCANNER: 'metadata_rules',
|
|
WORKER_ASSET_SECURITY_SCAN_HTTP_ENDPOINT: '',
|
|
WORKER_ASSET_SECURITY_SCAN_HTTP_TOKEN: '',
|
|
WORKER_ASSET_SECURITY_SCAN_HTTP_TIMEOUT_MS: '10000',
|
|
WORKER_ASSET_SECURITY_SCAN_FAIL_OPEN: 'false',
|
|
STORAGE_REQUIRE_TENANT_PREFIX: 'true',
|
|
...extraEnv,
|
|
},
|
|
stdio: ['ignore', 'pipe', 'pipe'],
|
|
windowsHide: true,
|
|
});
|
|
let output = '';
|
|
child.stdout.on('data', chunk => {
|
|
output += chunk.toString();
|
|
});
|
|
child.stderr.on('data', chunk => {
|
|
output += chunk.toString();
|
|
});
|
|
const code = await new Promise(resolve => child.on('exit', resolve));
|
|
assert.equal(code, 0, `worker should exit 0\n${output}`);
|
|
assert.match(output, /assets batch processed=\d+/, 'worker output should include assets summary');
|
|
return output;
|
|
}
|
|
|
|
async function startScannerServer() {
|
|
const requests = [];
|
|
const server = http.createServer((request, response) => {
|
|
let body = '';
|
|
request.on('data', chunk => {
|
|
body += chunk.toString();
|
|
});
|
|
request.on('end', () => {
|
|
const payload = JSON.parse(body || '{}');
|
|
requests.push({
|
|
authorization: request.headers.authorization,
|
|
payload,
|
|
});
|
|
response.setHeader('content-type', 'application/json');
|
|
if (payload.assetId === ids.httpFailAsset) {
|
|
response.end(JSON.stringify({
|
|
status: 'failed',
|
|
riskLevel: 'critical',
|
|
issueCodes: ['virus_detected'],
|
|
details: {
|
|
engine: 'fake-av',
|
|
signature: 'EICAR-Test-File',
|
|
accessToken: 'must-be-redacted',
|
|
},
|
|
}));
|
|
return;
|
|
}
|
|
response.end(JSON.stringify({
|
|
status: 'passed',
|
|
riskLevel: 'none',
|
|
issueCodes: [],
|
|
details: {
|
|
engine: 'fake-av',
|
|
scannedBytes: payload.fileSizeBytes,
|
|
},
|
|
}));
|
|
});
|
|
});
|
|
await new Promise(resolve => server.listen(0, '127.0.0.1', resolve));
|
|
const address = server.address();
|
|
assert.ok(address && typeof address === 'object', 'scanner server should listen on a TCP port');
|
|
return {
|
|
endpoint: `http://127.0.0.1:${address.port}/scan`,
|
|
requests,
|
|
close: () => new Promise(resolve => server.close(resolve)),
|
|
};
|
|
}
|
|
|
|
function countFromWorkerOutput(output, name) {
|
|
const match = output.match(new RegExp(`${name}=(\\d+)`));
|
|
return match ? Number(match[1]) : 0;
|
|
}
|
|
|
|
async function cleanup(pool) {
|
|
await pool.query(
|
|
`
|
|
delete from public.audit_logs
|
|
where tenant_id = $1
|
|
and target_type = 'content_asset'
|
|
and target_id = any($2::text[])
|
|
`,
|
|
[tenantId, Object.values(ids)],
|
|
);
|
|
await pool.query(
|
|
`
|
|
delete from public.content_asset_security_scan_events
|
|
where tenant_id = $1 and asset_id = any($2::uuid[])
|
|
`,
|
|
[tenantId, Object.values(ids)],
|
|
);
|
|
await pool.query(
|
|
`
|
|
delete from public.content_assets
|
|
where tenant_id = $1 and id = any($2::uuid[])
|
|
`,
|
|
[tenantId, Object.values(ids)],
|
|
);
|
|
}
|
|
|
|
async function seed(pool) {
|
|
await pool.query(
|
|
`
|
|
insert into public.content_assets (
|
|
id, tenant_id, asset_key, title, asset_type, storage_provider,
|
|
bucket, object_key, file_name, mime_type, file_size_bytes, checksum_sha256,
|
|
visibility, status, upload_status, verified_at, verified_size_bytes,
|
|
verified_checksum_sha256, verification_details,
|
|
security_scan_status, security_scan_provider, security_scan_summary,
|
|
security_flags, source,
|
|
created_at, updated_at
|
|
)
|
|
values
|
|
(
|
|
$1, $2, 'asset-worker-ok', '资源复检正常 PDF', 'pdf', 'local_dev',
|
|
'tenant-assets', $3, 'ok.pdf', 'application/pdf', 4096, $4,
|
|
'tenant', 'active', 'verified', now() - interval '2 days', 4096,
|
|
$4, '{"source":"asset-worker-test"}'::jsonb,
|
|
'pending', 'metadata_rules', '{}'::jsonb,
|
|
'{}'::jsonb, 'integration-test',
|
|
now() - interval '2 days', now() - interval '2 days'
|
|
),
|
|
(
|
|
$5, $2, 'asset-worker-bad', '资源复检异常 PDF', 'pdf', 'local_dev',
|
|
'tenant-assets', $6, 'bad.pdf', 'application/pdf', 1024, $7,
|
|
'tenant', 'active', 'verified', now() - interval '2 days', 2048,
|
|
$7, '{"source":"asset-worker-test"}'::jsonb,
|
|
'pending', 'metadata_rules', '{}'::jsonb,
|
|
'{}'::jsonb, 'integration-test',
|
|
now() - interval '2 days', now() - interval '2 days'
|
|
),
|
|
(
|
|
$8, $2, 'asset-worker-scan-bad', '资源安全扫描异常 PDF', 'pdf', 'local_dev',
|
|
'tenant-assets', $9, 'scan-bad.pdf', 'application/pdf', 4096, $4,
|
|
'tenant', 'active', 'verified', now() - interval '2 days', 4096,
|
|
$4, '{"source":"asset-worker-test"}'::jsonb,
|
|
'pending', 'metadata_rules', '{"securityScanForceFail":true}'::jsonb,
|
|
'{}'::jsonb, 'integration-test',
|
|
now() - interval '2 days', now() - interval '2 days'
|
|
),
|
|
(
|
|
$10, $2, 'asset-worker-http-pass', '外部扫描正常 PDF', 'pdf', 'local_dev',
|
|
'tenant-assets', $11, 'http-pass.pdf', 'application/pdf', 4096, $4,
|
|
'tenant', 'active', 'verified', now() - interval '2 days', 4096,
|
|
$4, '{"source":"asset-worker-test"}'::jsonb,
|
|
'pending', 'metadata_rules', '{}'::jsonb,
|
|
'{}'::jsonb, 'integration-test',
|
|
now() - interval '2 days', now() - interval '2 days'
|
|
),
|
|
(
|
|
$12, $2, 'asset-worker-http-fail', '外部扫描异常 PDF', 'pdf', 'local_dev',
|
|
'tenant-assets', $13, 'http-fail.pdf', 'application/pdf', 4096, $4,
|
|
'tenant', 'active', 'verified', now() - interval '2 days', 4096,
|
|
$4, '{"source":"asset-worker-test"}'::jsonb,
|
|
'pending', 'metadata_rules', '{}'::jsonb,
|
|
'{}'::jsonb, 'integration-test',
|
|
now() - interval '2 days', now() - interval '2 days'
|
|
),
|
|
(
|
|
$14, $2, 'asset-worker-http-unavailable', '外部扫描不可用 PDF', 'pdf', 'local_dev',
|
|
'tenant-assets', $15, 'http-unavailable.pdf', 'application/pdf', 4096, $4,
|
|
'tenant', 'active', 'verified', now() - interval '2 days', 4096,
|
|
$4, '{"source":"asset-worker-test"}'::jsonb,
|
|
'pending', 'metadata_rules', '{}'::jsonb,
|
|
'{}'::jsonb, 'integration-test',
|
|
now() - interval '2 days', now() - interval '2 days'
|
|
)
|
|
`,
|
|
[
|
|
ids.okAsset,
|
|
tenantId,
|
|
`${tenantId}/assets/worker-ok.pdf`,
|
|
checksumA,
|
|
ids.badAsset,
|
|
`${tenantId}/assets/worker-bad.pdf`,
|
|
checksumB,
|
|
ids.scanBadAsset,
|
|
`${tenantId}/assets/worker-scan-bad.pdf`,
|
|
ids.httpPassAsset,
|
|
`${tenantId}/assets/worker-http-pass.pdf`,
|
|
ids.httpFailAsset,
|
|
`${tenantId}/assets/worker-http-fail.pdf`,
|
|
ids.httpUnavailableAsset,
|
|
`${tenantId}/assets/worker-http-unavailable.pdf`,
|
|
],
|
|
);
|
|
}
|
|
|
|
async function resetAssetsForHttp(pool) {
|
|
await pool.query(
|
|
`
|
|
update public.content_assets
|
|
set status = 'active',
|
|
upload_status = 'verified',
|
|
verified_at = now() - interval '2 days',
|
|
verification_details = '{"source":"asset-worker-test"}'::jsonb,
|
|
security_scan_status = 'pending',
|
|
security_scan_provider = 'metadata_rules',
|
|
security_scan_summary = '{}'::jsonb,
|
|
security_flags = '{}'::jsonb,
|
|
updated_at = now() - interval '2 days'
|
|
where tenant_id = $1
|
|
and id = any($2::uuid[])
|
|
`,
|
|
[tenantId, [ids.httpPassAsset, ids.httpFailAsset, ids.httpUnavailableAsset]],
|
|
);
|
|
await pool.query(
|
|
`
|
|
delete from public.content_asset_security_scan_events
|
|
where tenant_id = $1 and asset_id = any($2::uuid[])
|
|
`,
|
|
[tenantId, [ids.httpPassAsset, ids.httpFailAsset, ids.httpUnavailableAsset]],
|
|
);
|
|
}
|
|
|
|
async function main() {
|
|
const pool = new pg.Pool({ connectionString: databaseUrl });
|
|
let seeded = false;
|
|
try {
|
|
await pool.query('begin');
|
|
await cleanup(pool);
|
|
await seed(pool);
|
|
await pool.query('commit');
|
|
seeded = true;
|
|
|
|
const output = await runWorkerOnce();
|
|
assert.ok(countFromWorkerOutput(output, 'verified') >= 3, 'worker should verify metadata-safe assets');
|
|
assert.ok(countFromWorkerOutput(output, 'failed') >= 2, 'worker should fail metadata and security scan assets');
|
|
|
|
const assets = await pool.query(
|
|
`
|
|
select id, status, upload_status, verification_details,
|
|
security_scan_status, security_scan_provider, security_scan_summary,
|
|
security_flags
|
|
from public.content_assets
|
|
where tenant_id = $1 and id = any($2::uuid[])
|
|
order by id
|
|
`,
|
|
[tenantId, [ids.okAsset, ids.badAsset, ids.scanBadAsset, ids.httpPassAsset, ids.httpFailAsset, ids.httpUnavailableAsset]],
|
|
);
|
|
const okAsset = assets.rows.find(row => row.id === ids.okAsset);
|
|
const badAsset = assets.rows.find(row => row.id === ids.badAsset);
|
|
const scanBadAsset = assets.rows.find(row => row.id === ids.scanBadAsset);
|
|
assert.equal(okAsset?.status, 'active', 'verified asset should remain active');
|
|
assert.equal(okAsset?.upload_status, 'verified', 'verified asset should remain verified');
|
|
assert.equal(okAsset?.security_scan_status, 'passed', 'verified asset should pass security scan');
|
|
assert.equal(okAsset?.security_scan_provider, 'metadata_rules', 'verified asset should record scan provider');
|
|
assert.equal(okAsset?.verification_details?.assetWorker?.lastResult, 'verified', 'verified asset should record worker result');
|
|
assert.equal(okAsset?.security_flags?.assetRecheckFailed, undefined, 'verified asset should not keep recheck failure flag');
|
|
|
|
assert.equal(badAsset?.status, 'draft', 'mismatched asset should be unpublished');
|
|
assert.equal(badAsset?.upload_status, 'failed', 'mismatched asset should be marked failed');
|
|
assert.equal(badAsset?.security_scan_status, 'skipped', 'mismatched asset should skip security scan');
|
|
assert.equal(badAsset?.security_flags?.assetRecheckFailed, true, 'mismatched asset should record security flag');
|
|
assert.deepEqual(
|
|
badAsset?.verification_details?.assetWorker?.issues,
|
|
['file_size_mismatch'],
|
|
'mismatched asset should record exact issue',
|
|
);
|
|
assert.equal(scanBadAsset?.status, 'draft', 'security failed asset should be unpublished');
|
|
assert.equal(scanBadAsset?.upload_status, 'verified', 'security failed asset should keep upload verification evidence');
|
|
assert.equal(scanBadAsset?.security_scan_status, 'failed', 'security failed asset should record failed scan status');
|
|
assert.equal(scanBadAsset?.security_flags?.assetSecurityScanFailed, true, 'security failed asset should record security flag');
|
|
assert.ok(
|
|
scanBadAsset?.security_scan_summary?.issueCodes?.includes('security_scan_forced_failure'),
|
|
'security failed asset should record issue code',
|
|
);
|
|
|
|
const scanEvents = await pool.query(
|
|
`
|
|
select asset_id, scan_status, risk_level, issue_codes, provider, details
|
|
from public.content_asset_security_scan_events
|
|
where tenant_id = $1 and asset_id in ($2::uuid, $3::uuid)
|
|
order by created_at asc
|
|
`,
|
|
[tenantId, ids.okAsset, ids.scanBadAsset],
|
|
);
|
|
assert.ok(
|
|
scanEvents.rows.some(row => row.asset_id === ids.okAsset && row.scan_status === 'passed' && row.provider === 'metadata_rules'),
|
|
'worker should write passed security scan event',
|
|
);
|
|
assert.ok(
|
|
scanEvents.rows.some(row => row.asset_id === ids.scanBadAsset && row.scan_status === 'failed' && row.issue_codes.includes('security_scan_forced_failure')),
|
|
'worker should write failed security scan event',
|
|
);
|
|
|
|
const skippedScanEvents = await pool.query(
|
|
`
|
|
select asset_id, scan_status, risk_level, issue_codes, provider, details
|
|
from public.content_asset_security_scan_events
|
|
where tenant_id = $1 and asset_id = $2::uuid
|
|
order by created_at asc
|
|
`,
|
|
[tenantId, ids.badAsset],
|
|
);
|
|
assert.ok(
|
|
skippedScanEvents.rows.some(row => row.scan_status === 'skipped' && row.issue_codes.includes('file_size_mismatch')),
|
|
'worker should write skipped scan event when metadata verification fails',
|
|
);
|
|
|
|
const audits = await pool.query(
|
|
`
|
|
select action, details
|
|
from public.audit_logs
|
|
where tenant_id = $1
|
|
and target_type = 'content_asset'
|
|
and target_id in ($2, $3, $4)
|
|
order by created_at asc
|
|
`,
|
|
[tenantId, ids.okAsset, ids.badAsset, ids.scanBadAsset],
|
|
);
|
|
assert.ok(
|
|
audits.rows.some(row => row.action === 'content.asset.rechecked' && row.details?.result === 'verified'),
|
|
'worker should write verified audit log',
|
|
);
|
|
assert.ok(
|
|
audits.rows.some(row => row.action === 'content.asset.recheck_failed' && row.details?.issues?.includes('file_size_mismatch')),
|
|
'worker should write failed audit log',
|
|
);
|
|
assert.ok(
|
|
audits.rows.some(row => row.action === 'content.asset.security_scan_failed' && row.details?.issueCodes?.includes('security_scan_forced_failure')),
|
|
'worker should write failed security scan audit log',
|
|
);
|
|
|
|
await resetAssetsForHttp(pool);
|
|
const scanner = await startScannerServer();
|
|
try {
|
|
const httpOutput = await runWorkerOnce({
|
|
WORKER_ASSET_SECURITY_SCANNER: 'metadata_rules,http',
|
|
WORKER_ASSET_SECURITY_SCAN_HTTP_ENDPOINT: scanner.endpoint,
|
|
WORKER_ASSET_SECURITY_SCAN_HTTP_TOKEN: 'scanner-test-token',
|
|
WORKER_ASSET_SECURITY_SCAN_HTTP_TIMEOUT_MS: '5000',
|
|
WORKER_ASSET_SECURITY_SCAN_FAIL_OPEN: 'false',
|
|
});
|
|
assert.ok(countFromWorkerOutput(httpOutput, 'verified') >= 1, 'worker should pass externally clean assets');
|
|
assert.ok(countFromWorkerOutput(httpOutput, 'failed') >= 1, 'worker should fail externally rejected assets');
|
|
assert.equal(scanner.requests.length, 3, 'external scanner should receive metadata-clean assets only');
|
|
assert.ok(
|
|
scanner.requests.every(item => item.authorization === 'Bearer scanner-test-token'),
|
|
'external scanner should receive bearer token',
|
|
);
|
|
|
|
const httpAssets = await pool.query(
|
|
`
|
|
select id, status, security_scan_status, security_scan_provider, security_scan_summary, security_flags
|
|
from public.content_assets
|
|
where tenant_id = $1 and id = any($2::uuid[])
|
|
order by id
|
|
`,
|
|
[tenantId, [ids.httpPassAsset, ids.httpFailAsset, ids.httpUnavailableAsset]],
|
|
);
|
|
const httpPassAsset = httpAssets.rows.find(row => row.id === ids.httpPassAsset);
|
|
const httpFailAsset = httpAssets.rows.find(row => row.id === ids.httpFailAsset);
|
|
const httpUnavailableAsset = httpAssets.rows.find(row => row.id === ids.httpUnavailableAsset);
|
|
|
|
assert.equal(httpPassAsset?.status, 'active', 'external scan pass asset should stay active');
|
|
assert.equal(httpPassAsset?.security_scan_status, 'passed', 'external scan pass asset should pass');
|
|
assert.equal(
|
|
httpPassAsset?.security_scan_provider,
|
|
'metadata_rules+http_security_scan',
|
|
'external scan pass asset should record merged provider',
|
|
);
|
|
assert.equal(httpFailAsset?.status, 'draft', 'external scan failed asset should be unpublished');
|
|
assert.equal(httpFailAsset?.security_scan_status, 'failed', 'external scan failed asset should fail');
|
|
assert.ok(
|
|
httpFailAsset?.security_scan_summary?.issueCodes?.includes('virus_detected'),
|
|
'external scan failed asset should include provider issue code',
|
|
);
|
|
const scannerDetails = httpFailAsset?.security_scan_summary?.details?.results?.find(
|
|
result => result.provider === 'http_security_scan',
|
|
)?.details?.scannerDetails;
|
|
assert.equal(scannerDetails?.accessToken, '[redacted]', 'scanner details should redact secret-like keys');
|
|
assert.equal(httpUnavailableAsset?.security_scan_status, 'passed', 'available scanner should pass clean third asset');
|
|
|
|
const httpScanEvents = await pool.query(
|
|
`
|
|
select asset_id, scan_status, provider, issue_codes, details
|
|
from public.content_asset_security_scan_events
|
|
where tenant_id = $1 and asset_id = any($2::uuid[])
|
|
order by created_at asc
|
|
`,
|
|
[tenantId, [ids.httpPassAsset, ids.httpFailAsset]],
|
|
);
|
|
assert.ok(
|
|
httpScanEvents.rows.some(row => row.asset_id === ids.httpPassAsset && row.provider === 'metadata_rules+http_security_scan' && row.scan_status === 'passed'),
|
|
'worker should write external pass event',
|
|
);
|
|
assert.ok(
|
|
httpScanEvents.rows.some(row => row.asset_id === ids.httpFailAsset && row.issue_codes.includes('virus_detected')),
|
|
'worker should write external failure event',
|
|
);
|
|
} finally {
|
|
await scanner.close();
|
|
}
|
|
|
|
await resetAssetsForHttp(pool);
|
|
const unavailableOutput = await runWorkerOnce({
|
|
WORKER_ASSET_SECURITY_SCANNER: 'metadata_rules,http',
|
|
WORKER_ASSET_SECURITY_SCAN_HTTP_ENDPOINT: 'http://127.0.0.1:9/scan',
|
|
WORKER_ASSET_SECURITY_SCAN_HTTP_TOKEN: 'scanner-test-token',
|
|
WORKER_ASSET_SECURITY_SCAN_HTTP_TIMEOUT_MS: '300',
|
|
WORKER_ASSET_SECURITY_SCAN_FAIL_OPEN: 'false',
|
|
});
|
|
assert.ok(countFromWorkerOutput(unavailableOutput, 'failed') >= 3, 'fail-closed scanner outage should fail assets');
|
|
const unavailableAssets = await pool.query(
|
|
`
|
|
select id, status, security_scan_status, security_scan_summary, security_flags
|
|
from public.content_assets
|
|
where tenant_id = $1 and id = any($2::uuid[])
|
|
`,
|
|
[tenantId, [ids.httpPassAsset, ids.httpFailAsset, ids.httpUnavailableAsset]],
|
|
);
|
|
for (const row of unavailableAssets.rows) {
|
|
assert.equal(row.status, 'draft', 'fail-closed scanner outage should unpublish active asset');
|
|
assert.equal(row.security_scan_status, 'failed', 'fail-closed scanner outage should fail scan');
|
|
assert.ok(
|
|
row.security_scan_summary?.issueCodes?.includes('external_security_scan_unavailable'),
|
|
'fail-closed scanner outage should record unavailable issue',
|
|
);
|
|
assert.equal(row.security_flags?.assetSecurityScanFailed, true, 'fail-closed scanner outage should record security flag');
|
|
}
|
|
|
|
console.log('Asset worker integration test complete.');
|
|
} catch (error) {
|
|
await pool.query('rollback').catch(() => {});
|
|
throw error;
|
|
} finally {
|
|
if (seeded) {
|
|
await cleanup(pool).catch(() => {});
|
|
}
|
|
await pool.end();
|
|
}
|
|
}
|
|
|
|
main().catch(error => {
|
|
console.error(error);
|
|
process.exit(1);
|
|
});
|