247 lines
13 KiB
TypeScript
247 lines
13 KiB
TypeScript
import { mkdtemp, rm } from 'node:fs/promises';
|
|
import { tmpdir } from 'node:os';
|
|
import { join } from 'node:path';
|
|
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
|
|
import { FileSystemBackendArtifactStore } from '../../server/artifacts/backend-artifact-store';
|
|
import { GpoBackupJobProcessor } from '../../server/gpo/gpo-backup-job-processor';
|
|
import { GpoBackupManagementService } from '../../server/gpo/gpo-backup-management-service';
|
|
import { GpoBackupSubmissionService } from '../../server/gpo/gpo-backup-submission-service';
|
|
import { directGpoBackupWorkerExecutor, directGpoInventoryExecutor } from '../../server/gpo/gpo-inventory-executor';
|
|
import type { IngestionJob } from '../../src/domain/ingestion-persistence';
|
|
import { LocalAssessorRepository } from '../../src/persistence/local-assessor-repository';
|
|
import { RepositoryConflictError } from '../../src/persistence/repository-errors';
|
|
import { syntheticGpoBackupZip } from './gpo-synthetic-fixture';
|
|
|
|
const initialTime = Date.parse('2026-07-21T10:00:00.000Z');
|
|
const project = {
|
|
id: 'gpo-job-project', name: 'Synthetic GPO job project',
|
|
createdAt: new Date(initialTime).toISOString(), updatedAt: new Date(initialTime).toISOString(), attributes: {}
|
|
};
|
|
|
|
class CompletionFailsOnceRepository extends LocalAssessorRepository {
|
|
private failNextCompletion = true;
|
|
|
|
override completeIngestionJob(jobId: string, workerId: string, completedAt: string): Promise<IngestionJob> {
|
|
if (this.failNextCompletion) {
|
|
this.failNextCompletion = false;
|
|
return Promise.reject(new RepositoryConflictError('synthetic completion interruption'));
|
|
}
|
|
return super.completeIngestionJob(jobId, workerId, completedAt);
|
|
}
|
|
}
|
|
|
|
describe('lease-based GPO backup job processor', () => {
|
|
let directory: string;
|
|
let repository: LocalAssessorRepository;
|
|
let store: FileSystemBackendArtifactStore;
|
|
let currentTime: number;
|
|
|
|
beforeEach(async () => {
|
|
directory = await mkdtemp(join(tmpdir(), 'assessor-gpo-job-'));
|
|
repository = new LocalAssessorRepository();
|
|
store = new FileSystemBackendArtifactStore(directory);
|
|
currentTime = initialTime;
|
|
await repository.saveProject(project);
|
|
});
|
|
|
|
afterEach(async () => {
|
|
await rm(directory, { recursive: true, force: true });
|
|
});
|
|
|
|
async function submit(target = repository): Promise<void> {
|
|
const service = new GpoBackupSubmissionService(target, store, directGpoInventoryExecutor, 1024 * 1024);
|
|
await service.submit({ projectId: project.id, fileName: 'backup.zip', bytes: syntheticGpoBackupZip(), receivedAt: new Date(currentTime).toISOString() });
|
|
}
|
|
|
|
function processor(target = repository): GpoBackupJobProcessor {
|
|
return new GpoBackupJobProcessor(target, store, directGpoBackupWorkerExecutor, {
|
|
maximumArtifactBytes: 1024 * 1024,
|
|
leaseDurationMs: 60_000,
|
|
retryDelayMs: 1_000,
|
|
now: () => new Date(currentTime)
|
|
});
|
|
}
|
|
|
|
it('reads the project artifact with integrity verification and atomically persists run and backup instance before completion', async () => {
|
|
await submit();
|
|
const result = await processor().processNext('worker-one');
|
|
|
|
expect(result).toMatchObject({ status: 'succeeded', backupInstanceCount: 1, processingRunId: expect.any(String) });
|
|
const artifacts = await repository.listImportArtifacts(project.id);
|
|
await expect(repository.listProcessingRuns(artifacts[0].id)).resolves.toMatchObject([{
|
|
id: result?.processingRunId, projectId: project.id, artifactId: artifacts[0].id, status: 'partial',
|
|
processors: [
|
|
{ kind: 'artifact-parser', id: 'gpo-backup-ingestion', version: '2.0.0' },
|
|
{ kind: 'normalizer', id: 'gpo-technical-registry-target', version: '1.0.0' }
|
|
]
|
|
}]);
|
|
await expect(repository.listGpoBackupInstances(project.id)).resolves.toMatchObject([{
|
|
projectId: project.id, artifactId: artifacts[0].id, processingRunId: result?.processingRunId,
|
|
instance: {
|
|
projectId: project.id,
|
|
sourceArtifactId: artifacts[0].id,
|
|
policyObservations: expect.arrayContaining([
|
|
expect.objectContaining({
|
|
deliveryMechanism: 'registry-policy',
|
|
state: 'instruction',
|
|
identity: expect.objectContaining({
|
|
section: 'HKEY_LOCAL_MACHINE',
|
|
key: 'Software\\Policies\\Synthetic\\Mode'
|
|
}),
|
|
registryPolicy: { sequence: 0, operation: 'set-value' },
|
|
values: [{
|
|
kind: 'registry-policy',
|
|
operation: 'set-value',
|
|
key: 'Software\\Policies\\Synthetic',
|
|
valueName: 'Mode',
|
|
typeCode: 4,
|
|
data: { kind: 'uint32', value: 1 }
|
|
}],
|
|
sourceRecords: [{ index: 0, byteOffset: 8 }],
|
|
effectiveState: 'unknown'
|
|
}), expect.objectContaining({
|
|
deliveryMechanism: 'security-template',
|
|
state: 'configured',
|
|
identity: expect.objectContaining({ section: 'system access', key: 'MinimumPasswordLength' }),
|
|
values: [{ kind: 'scalar', raw: '14' }],
|
|
effectiveState: 'unknown'
|
|
}),
|
|
expect.objectContaining({
|
|
deliveryMechanism: 'advanced-audit-policy',
|
|
state: 'configured',
|
|
identity: expect.objectContaining({ key: '0cce9210-69ae-11d9-bed3-505054503030' }),
|
|
values: [{ kind: 'audit-flags', code: 3, success: true, failure: true }],
|
|
effectiveState: 'unknown'
|
|
}),
|
|
expect.objectContaining({
|
|
deliveryMechanism: 'group-policy-preference',
|
|
state: 'instruction',
|
|
identity: expect.objectContaining({ section: 'registry' }),
|
|
preference: expect.objectContaining({ cse: 'registry', operation: 'update', sequence: 0 }),
|
|
values: [expect.objectContaining({
|
|
kind: 'registry-preference', hive: 'HKEY_LOCAL_MACHINE', key: 'Software\Synthetic',
|
|
target: 'value', valueName: 'Mode', valueType: 'REG_SZ', rawData: 'Strict'
|
|
})],
|
|
effectiveState: 'unknown'
|
|
}),
|
|
expect.objectContaining({
|
|
deliveryMechanism: 'group-policy-preference',
|
|
state: 'instruction',
|
|
identity: expect.objectContaining({ section: 'local user', key: 'syntheticoperator' }),
|
|
preference: expect.objectContaining({ cse: 'local-users-groups', operation: 'update', sequence: 0 }),
|
|
values: [expect.objectContaining({
|
|
kind: 'local-user-preference', userName: 'SyntheticOperator',
|
|
passwordAttributePresent: true, accountDisabled: true
|
|
})],
|
|
diagnosticCodes: ['gpo-secret-indicator-present'],
|
|
effectiveState: 'unknown'
|
|
}),
|
|
expect.objectContaining({
|
|
deliveryMechanism: 'group-policy-preference',
|
|
state: 'instruction',
|
|
identity: expect.objectContaining({ section: 'service', key: 'syntheticservice' }),
|
|
preference: expect.objectContaining({ cse: 'services', operation: 'configure', sequence: 0 }),
|
|
values: [expect.objectContaining({
|
|
kind: 'service-preference', serviceName: 'SyntheticService', serviceAction: 'restart-if-required',
|
|
startupType: 'automatic', accountName: 'Synthetic\\Service', passwordAttributePresent: true,
|
|
thirdFailure: 'restart', restartServiceDelay: 900000
|
|
})],
|
|
diagnosticCodes: ['gpo-secret-indicator-present'],
|
|
effectiveState: 'unknown'
|
|
}),
|
|
expect.objectContaining({
|
|
deliveryMechanism: 'group-policy-preference',
|
|
scope: 'user',
|
|
state: 'instruction',
|
|
identity: expect.objectContaining({ section: 'drive map', key: 'S:exact' }),
|
|
preference: expect.objectContaining({ cse: 'drive-maps', operation: 'replace', sequence: 0 }),
|
|
values: [expect.objectContaining({
|
|
kind: 'drive-map-preference', path: '\\\\synthetic-server\\share', label: 'Synthetic Share',
|
|
persistent: true, userName: 'SYNTHETIC\\DriveUser', passwordAttributePresent: true,
|
|
letter: 'S', useLetter: true, selectionMode: 'exact', thisDrive: 'show', allDrives: 'no-change'
|
|
})],
|
|
diagnosticCodes: ['gpo-secret-indicator-present'],
|
|
effectiveState: 'unknown'
|
|
})
|
|
])
|
|
}
|
|
}]);
|
|
const policyAnalyses = await repository.listGpoPolicyAnalyses(project.id);
|
|
expect(policyAnalyses).toHaveLength(1);
|
|
expect(policyAnalyses[0]).toMatchObject({
|
|
run: {
|
|
sourceProcessingRunId: result?.processingRunId,
|
|
analyzer: { id: 'gpo-policy-identity-analysis', version: 'gpo-backup-policy-analysis:2.0.0' },
|
|
identityResolver: { id: 'gpo-technical-registry-target', version: '1.0.0' },
|
|
observationCount: expect.any(Number),
|
|
resolvedCount: 1,
|
|
status: 'partial'
|
|
},
|
|
resolutions: expect.arrayContaining([
|
|
expect.objectContaining({ resolution: expect.objectContaining({ status: 'resolved' }) }),
|
|
expect.objectContaining({ resolution: expect.objectContaining({ status: 'unresolved' }) })
|
|
])
|
|
});
|
|
const exported = await repository.exportProject(project.id);
|
|
expect(exported).toMatchObject({
|
|
gpoBackupInstances: [{ processingRunId: result?.processingRunId }],
|
|
gpoPolicyAnalyses: [{ run: { sourceProcessingRunId: result?.processingRunId } }]
|
|
});
|
|
expect(JSON.stringify(exported)).not.toContain('synthetic-cpassword-fixture');
|
|
expect(JSON.stringify(exported)).not.toContain('synthetic-service-cpassword-fixture');
|
|
expect(JSON.stringify(exported)).not.toContain('synthetic-drive-cpassword-fixture');
|
|
});
|
|
|
|
it('recovers idempotently when completion fails after the atomic analysis commit', async () => {
|
|
repository = new CompletionFailsOnceRepository();
|
|
await repository.saveProject(project);
|
|
await submit(repository);
|
|
const value = processor(repository);
|
|
|
|
const first = await value.processNext('worker-one');
|
|
expect(first).toMatchObject({ status: 'queued', errorCode: 'gpo-analysis-failed' });
|
|
expect(await repository.listGpoBackupInstances(project.id)).toHaveLength(1);
|
|
const revisionAfterCommit = repository.revision;
|
|
|
|
currentTime += 1_000;
|
|
const second = await value.processNext('worker-two');
|
|
expect(second).toMatchObject({ status: 'succeeded', processingRunId: first?.processingRunId, backupInstanceCount: 1 });
|
|
expect(await repository.listGpoBackupInstances(project.id)).toHaveLength(1);
|
|
expect(repository.revision).toBeGreaterThan(revisionAfterCommit);
|
|
});
|
|
|
|
it('creates one idempotent superseding run for an explicit reprocessing request', async () => {
|
|
await submit();
|
|
const first = await processor().processNext('worker-one');
|
|
if (!first?.processingRunId) throw new Error('Synthetic first processing run missing.');
|
|
const artifact = (await repository.listImportArtifacts(project.id))[0];
|
|
currentTime += 1_000;
|
|
const management = new GpoBackupManagementService(repository, store, () => new Date(currentTime));
|
|
const requested = await management.reprocess(project.id, artifact.id, first.processingRunId);
|
|
const repeated = await management.reprocess(project.id, artifact.id, first.processingRunId);
|
|
expect(repeated.jobId).toBe(requested.jobId);
|
|
|
|
const second = await processor().processNext('worker-two');
|
|
expect(second).toMatchObject({ status: 'succeeded', processingRunId: expect.any(String) });
|
|
expect(second?.processingRunId).not.toBe(first.processingRunId);
|
|
const runs = await repository.listProcessingRuns(artifact.id);
|
|
expect(runs.find(item => item.id === second?.processingRunId)).toMatchObject({ supersedesRunId: first.processingRunId });
|
|
const analyses = await repository.listGpoPolicyAnalyses(project.id);
|
|
const firstAnalysis = analyses.find(item => item.run.sourceProcessingRunId === first.processingRunId);
|
|
const secondAnalysis = analyses.find(item => item.run.sourceProcessingRunId === second?.processingRunId);
|
|
expect(secondAnalysis?.run.supersedesAnalysisRunId).toBe(firstAnalysis?.run.id);
|
|
await expect(management.reprocess(project.id, artifact.id, first.processingRunId)).rejects.toBeInstanceOf(RepositoryConflictError);
|
|
});
|
|
it('claims only dedicated GPO analysis jobs', async () => {
|
|
await repository.enqueueIngestionJob({
|
|
version: 1, id: 'unrelated-job', projectId: project.id, kind: 'parse-artifact', idempotencyKey: 'unrelated',
|
|
status: 'queued', attempts: 0, maxAttempts: 1, availableAt: new Date(currentTime).toISOString(),
|
|
createdAt: new Date(currentTime).toISOString(), updatedAt: new Date(currentTime).toISOString()
|
|
});
|
|
await expect(processor().processNext('gpo-worker')).resolves.toBeNull();
|
|
await expect(repository.listIngestionJobs(project.id)).resolves.toMatchObject([{ id: 'unrelated-job', status: 'queued' }]);
|
|
});
|
|
});
|
|
|
|
|