diff --git a/lib/cso/cache.ts b/lib/cso/cache.ts index 2c639f45a..e9a3f6ecb 100644 --- a/lib/cso/cache.ts +++ b/lib/cso/cache.ts @@ -9,6 +9,7 @@ export const DEFAULT_PUBLIC_ARCHIVE_CACHE_BYTES = 10 * 1024 * 1024 * 1024; const METADATA_VERSION = 1; const METADATA_LIMIT = 4096; const COPY_BUFFER_BYTES = 64 * 1024; +const MAX_CACHE_DIRECTORY_ENTRIES = 100_000; const CACHE_LOCK_PROTOCOL = 'immutable-cache-lease-set-v3'; const SHA256 = /^[a-f0-9]{64}$/; const RELATIVE_STAGE_PATH = /^(?!\/)(?!.*(?:^|\/)\.\.?(?:\/|$))(?!.*\\)[^\0-\x1f\x7f]+$/; @@ -26,6 +27,17 @@ export interface PublicArchiveCacheOptions { now?: () => number; } +/** Optional cooperative bounds for synchronous cache work. */ +export interface CacheOperationControl { + /** Absolute Unix timestamp in milliseconds. */ + deadline?: number; + /** Checked between bounded filesystem operations. */ + signal?: AbortSignal; +} + +export type CacheOperationInput = CacheOperationControl | number | undefined; +type NormalizedCacheOperationControl = Readonly<{ deadline?: number; signal?: AbortSignal }>; + export interface PublicArchiveCacheEntry { sha256: string; path: string; @@ -69,6 +81,49 @@ function fail(code: ConstructorParameters[0], message: string): throw new CsoError(code, message); } +function operationControl(input?: CacheOperationInput): NormalizedCacheOperationControl { + const value = typeof input === 'number' ? { deadline: input } : input ?? {}; + if (!value || typeof value !== 'object' || Array.isArray(value)) fail('INVALID_ARGUMENT', 'Cache operation control must be an object or absolute deadline'); + if (value.deadline !== undefined && (!Number.isSafeInteger(value.deadline) || value.deadline <= 0)) + fail('INVALID_ARGUMENT', 'Cache deadline must be an absolute millisecond timestamp'); + if (value.signal !== undefined && typeof value.signal.aborted !== 'boolean') + fail('INVALID_ARGUMENT', 'Cache cancellation signal is invalid'); + const control = Object.freeze({ ...(value.deadline === undefined ? {} : { deadline: value.deadline }), + ...(value.signal === undefined ? {} : { signal: value.signal }) }); + checkOperation(control); + return control; +} + +function checkOperation(control: NormalizedCacheOperationControl): void { + if (control.signal?.aborted) fail('CANCELLED', 'Archive-cache operation was cancelled'); + if (control.deadline !== undefined && Date.now() >= control.deadline) + fail('DEADLINE', 'Archive-cache operation reached its deadline'); +} + +function boundedDirectoryNames(path: string, label: string, control: NormalizedCacheOperationControl): string[] { + checkOperation(control); + const directory = fs.opendirSync(path), names: string[] = []; + try { + for (;;) { + checkOperation(control); + const entry = directory.readSync(); + if (!entry) break; + if (names.length >= MAX_CACHE_DIRECTORY_ENTRIES) fail('INSUFFICIENT_CAPACITY', `${label} exceeds the cache entry limit`); + names.push(entry.name); + } + } finally { directory.closeSync(); } + checkOperation(control); + return names; +} + +function assertEmptyDirectory(path: string, control: NormalizedCacheOperationControl): void { + checkOperation(control); + const directory = fs.opendirSync(path); + try { if (directory.readSync()) fail('UNSAFE_PATH', 'Archive materialization directory must be empty'); } + finally { directory.closeSync(); } + checkOperation(control); +} + function boundedPositiveInteger(value: number, name: string): number { if (!Number.isSafeInteger(value) || value <= 0) fail('INVALID_ARGUMENT', `${name} must be a positive safe integer`); return value; @@ -130,10 +185,11 @@ function assertExistingDirectory(path: string, label: string): string { return canonical!; } -function assertContainedAncestors(root: string, relativePath: string): string { +function assertContainedAncestors(root: string, relativePath: string, control: NormalizedCacheOperationControl): string { const parts = relativePath.split('/'); let cursor = root; for (const part of parts.slice(0, -1)) { + checkOperation(control); cursor = join(cursor, part); let stat: fs.Stats; try { stat = fs.lstatSync(cursor); } @@ -154,15 +210,19 @@ function openNoFollow(path: string, flags: number, mode?: number): number { catch { fail('UNSAFE_PATH', 'Archive file could not be opened without following links'); } } -function readMetadata(path: string, digest: string): Metadata { +function readMetadata(path: string, digest: string, control: NormalizedCacheOperationControl): Metadata { + checkOperation(control); let stat: fs.Stats; try { stat = fs.lstatSync(path); } catch { fail('INCOMPATIBLE_INPUT', `Cache metadata is missing for ${digest}`); } assertOwnedRegular(stat!, 'Cache metadata', METADATA_LIMIT); if ((stat!.mode & 0o077) !== 0) fail('INCOMPATIBLE_INPUT', 'Cache metadata permissions are not private'); let value: unknown; - try { value = JSON.parse(fs.readFileSync(path, 'utf8')); } - catch { fail('INCOMPATIBLE_INPUT', `Cache metadata is invalid for ${digest}`); } + try { checkOperation(control); value = JSON.parse(fs.readFileSync(path, 'utf8')); checkOperation(control); } + catch (error) { + if (error instanceof CsoError) throw error; + fail('INCOMPATIBLE_INPUT', `Cache metadata is invalid for ${digest}`); + } const record = value as Partial; if (!value || typeof value !== 'object' || Array.isArray(value) || Object.keys(value).sort().join(',') !== 'bytes,createdAt,lastAccessedAt,sha256,version' || record.version !== METADATA_VERSION || record.sha256 !== digest || !Number.isSafeInteger(record.bytes) || Number(record.bytes) < 0 || @@ -226,12 +286,14 @@ export class PublicArchiveCache { } /** Promote a verified staging file. The staging file is never deleted. */ - promote(stagedPath: string, sha256: string): PublicArchiveCacheEntry { + promote(stagedPath: string, sha256: string, operation?: CacheOperationInput): PublicArchiveCacheEntry { + const control = operationControl(operation); const digest = expectedDigest(sha256), relativePath = stagedRelativePath(stagedPath); - const source = assertContainedAncestors(this.stagingRoot, relativePath); + const source = assertContainedAncestors(this.stagingRoot, relativePath, control); return this.withLock(() => { - this.cleanIncoming(); - this.recoverInterruptedOperations(); + checkOperation(control); + this.cleanIncoming(control); + this.recoverInterruptedOperations(control); let initial: fs.Stats; try { initial = fs.lstatSync(source); } catch { fail('MISSING_INPUT', 'Staged archive is missing'); } @@ -243,24 +305,25 @@ export class PublicArchiveCache { if (targetExists !== metadataExists) fail('INCOMPATIBLE_INPUT', `Cache entry is incomplete for ${digest}`); if (targetExists) { - const staged = this.hashFile(source, this.maxEntryBytes, false, stableStat(initial!)); + const staged = this.hashFile(source, this.maxEntryBytes, false, stableStat(initial!), control); if (staged.digest !== digest) fail('INCOMPATIBLE_INPUT', 'Staged archive does not match its caller-provided SHA-256'); - const metadata = this.verifiedEntry(digest); - return this.touch(metadata); + const metadata = this.verifiedEntry(digest, control); + return this.touch(metadata, control); } // Authenticate the complete staged object before it is allowed to // displace any already-verified cache entry. Copying below hashes it a // second time so a staging race still fails closed. - const authenticated = this.hashFile(source, this.maxEntryBytes, false, stableStat(initial!)); + const authenticated = this.hashFile(source, this.maxEntryBytes, false, stableStat(initial!), control); if (authenticated.digest !== digest) fail('INCOMPATIBLE_INPUT', 'Staged archive does not match its caller-provided SHA-256'); - this.evictToFit(initial!.size); + this.evictToFit(initial!.size, control); const incoming = join(this.incomingDir, `.incoming-${process.pid}-${randomBytes(12).toString('hex')}`); let promoted = false; try { - const staged = this.copyAndHash(source, incoming, this.maxEntryBytes, stableStat(initial!)); + const staged = this.copyAndHash(source, incoming, this.maxEntryBytes, stableStat(initial!), control); if (staged.digest !== digest) fail('INCOMPATIBLE_INPUT', 'Staged archive does not match its caller-provided SHA-256'); if (staged.bytes !== initial!.size) fail('SNAPSHOT_RACE', 'Staged archive changed during promotion'); + checkOperation(control); fs.chmodSync(incoming, 0o400); // A hard-link followed by unlink is an atomic no-replace publication on // the cache filesystem. rename(2) would silently replace a raced target. @@ -269,7 +332,7 @@ export class PublicArchiveCache { promoted = true; const now = this.timestamp(); const metadata: Metadata = { version: 1, sha256: digest, bytes: staged.bytes, createdAt: now, lastAccessedAt: now }; - try { writeMetadata(metadataPath, metadata, true); } + try { checkOperation(control); writeMetadata(metadataPath, metadata, true); } catch (error) { try { this.discardObject(target, 'entry', digest); } catch {} throw error; @@ -281,30 +344,36 @@ export class PublicArchiveCache { if (stat.isFile() && !stat.isSymbolicLink()) fs.unlinkSync(incoming); } } - }); + }, control); } /** Return a cache hit only after hashing every byte and validating metadata. */ - get(sha256: string): PublicArchiveCacheEntry | undefined { + get(sha256: string, operation?: CacheOperationInput): PublicArchiveCacheEntry | undefined { + const control = operationControl(operation); const digest = expectedDigest(sha256); return this.withLock(() => { - this.cleanIncoming(); - this.recoverInterruptedOperations(); + checkOperation(control); + this.cleanIncoming(control); + this.recoverInterruptedOperations(control); const targetExists = existsNoFollow(this.entryPath(digest)), metadataExists = existsNoFollow(this.metadataPath(digest)); if (!targetExists && !metadataExists) return undefined; if (targetExists !== metadataExists) fail('INCOMPATIBLE_INPUT', `Cache entry is incomplete for ${digest}`); - return this.touch(this.verifiedEntry(digest)); - }); + return this.touch(this.verifiedEntry(digest, control), control); + }, control); } /** Inspect capacity without treating entries as execution-ready cache hits. */ - stats(): PublicArchiveCacheStats { + stats(operation?: CacheOperationInput): PublicArchiveCacheStats { + const control = operationControl(operation); return this.withLock(() => { - this.cleanIncoming(); - this.recoverInterruptedOperations(); - const entries = this.inventory(); - return { entries: entries.length, bytes: entries.reduce((sum, item) => sum + item.bytes, 0), maxBytes: this.maxBytes }; - }); + checkOperation(control); + this.cleanIncoming(control); + this.recoverInterruptedOperations(control); + const entries = this.inventory(control); + let bytes = 0; + for (const entry of entries) { checkOperation(control); bytes += entry.bytes; } + return { entries: entries.length, bytes, maxBytes: this.maxBytes }; + }, control); } /** @@ -313,35 +382,49 @@ export class PublicArchiveCache { * cache paths. Every source and every copy is fully hashed in the same * critical section. */ - materialize(digests: string[], destinationRoot: string): MaterializedArchive[] { - if (!Array.isArray(digests) || !digests.length || digests.some(value => typeof value !== 'string')) + materialize(digests: string[], destinationRoot: string, operation?: CacheOperationInput): MaterializedArchive[] { + const control = operationControl(operation); + if (!Array.isArray(digests) || !digests.length) fail('INVALID_ARGUMENT', 'Archive materialization requires at least one SHA-256 digest'); - const selected = [...new Set(digests.map(expectedDigest))].sort(), destination = assertExistingDirectory(destinationRoot, 'Archive materialization directory'); - if (fs.readdirSync(destination).length) fail('UNSAFE_PATH', 'Archive materialization directory must be empty'); + if (digests.length > MAX_CACHE_DIRECTORY_ENTRIES) + fail('INSUFFICIENT_CAPACITY', 'Archive materialization exceeds the cache entry limit'); + const selectedSet=new Set();for(const digest of digests){checkOperation(control);if(typeof digest!=='string')fail('INVALID_ARGUMENT', 'Archive materialization requires SHA-256 digest strings');selectedSet.add(expectedDigest(digest));} + checkOperation(control);const selected=[...selectedSet].sort();checkOperation(control); + const destination = assertExistingDirectory(destinationRoot, 'Archive materialization directory'); + assertEmptyDirectory(destination, control); if (destination === this.root || destination.startsWith(`${this.root}${sep}`) || this.root.startsWith(`${destination}${sep}`) || destination === this.stagingRoot || destination.startsWith(`${this.stagingRoot}${sep}`) || this.stagingRoot.startsWith(`${destination}${sep}`)) fail('UNSAFE_PATH', 'Archive materialization directory must be separate from cache and staging roots'); return this.withLock(() => { - this.cleanIncoming(); - this.recoverInterruptedOperations(); + checkOperation(control); + this.cleanIncoming(control); + this.recoverInterruptedOperations(control); const created: string[] = [], result: MaterializedArchive[] = []; try { for (const digest of selected) { - const metadata = this.verifiedEntry(digest), source = this.entryPath(digest), initial = fs.lstatSync(source); + checkOperation(control); + const metadata = this.verifiedEntry(digest, control), source = this.entryPath(digest), initial = fs.lstatSync(source); assertOwnedRegular(initial, 'Cached archive', this.maxEntryBytes, true); - const target = join(destination, digest), copied = this.copyAndHash(source, target, this.maxEntryBytes, stableStat(initial)); + const target = join(destination, digest); + let copied: { digest: string; bytes: number }; + try { copied = this.copyAndHash(source, target, this.maxEntryBytes, stableStat(initial), control); } + catch (error) { + if (existsNoFollow(target)) try { removeRegular(target, 'Incomplete run-owned archive copy'); } catch {} + throw error; + } + created.push(target); if (copied.digest !== digest || copied.bytes !== metadata.bytes) fail('SNAPSHOT_RACE', 'Cached archive changed while its run-owned copy was materialized'); fs.chmodSync(target, 0o400); - const verified = this.hashFile(target, this.maxEntryBytes, true); + const verified = this.hashFile(target, this.maxEntryBytes, true, undefined, control); if (verified.digest !== digest || verified.bytes !== metadata.bytes) fail('SNAPSHOT_RACE', 'Run-owned archive copy failed verification'); - created.push(target); result.push(Object.freeze({ sha256: digest, path: target, bytes: verified.bytes })); + result.push(Object.freeze({ sha256: digest, path: target, bytes: verified.bytes })); } return result; } catch (error) { for (const path of created.reverse()) { try { removeRegular(path, 'Incomplete run-owned archive copy'); } catch {} } throw error; } - }); + }, control); } private timestamp(): number { @@ -358,20 +441,25 @@ export class PublicArchiveCache { createdAt: metadata.createdAt, lastAccessedAt: metadata.lastAccessedAt }); } - private touch(metadata: Metadata): PublicArchiveCacheEntry { + private touch(metadata: Metadata, control: NormalizedCacheOperationControl): PublicArchiveCacheEntry { + checkOperation(control); const updated: Metadata = { ...metadata, lastAccessedAt: Math.max(metadata.lastAccessedAt, this.timestamp()) }; + checkOperation(control); writeMetadata(this.metadataPath(metadata.sha256), updated); return this.entry(updated); } - private verifiedEntry(digest: string): Metadata { - const metadata = readMetadata(this.metadataPath(digest), digest); - const result = this.hashFile(this.entryPath(digest), this.maxEntryBytes, true); + private verifiedEntry(digest: string, control: NormalizedCacheOperationControl): Metadata { + checkOperation(control); + const metadata = readMetadata(this.metadataPath(digest), digest, control); + const result = this.hashFile(this.entryPath(digest), this.maxEntryBytes, true, undefined, control); if (result.digest !== digest || result.bytes !== metadata.bytes) fail('INCOMPATIBLE_INPUT', `Cached archive failed SHA-256 verification: ${digest}`); return metadata; } - private hashFile(path: string, maxBytes: number, immutable: boolean, expected?: StableStat): { digest: string; bytes: number } { + private hashFile(path: string, maxBytes: number, immutable: boolean, expected: StableStat | undefined, + control: NormalizedCacheOperationControl): { digest: string; bytes: number } { + checkOperation(control); const fd = openNoFollow(path, fs.constants.O_RDONLY); try { const beforeStat = fs.fstatSync(fd); @@ -380,19 +468,24 @@ export class PublicArchiveCache { if (expected && !sameStat(expected, before)) fail('SNAPSHOT_RACE', 'Archive changed before it could be verified'); let bytes = 0; for (;;) { + checkOperation(control); const read = fs.readSync(fd, buffer, 0, buffer.length, null); if (!read) break; bytes += read; if (bytes > maxBytes) fail('INSUFFICIENT_CAPACITY', 'Archive exceeded its byte limit while being read'); hash.update(buffer.subarray(0, read)); + checkOperation(control); } + checkOperation(control); const after = stableStat(fs.fstatSync(fd)); if (!sameStat(before, after) || bytes !== before.size) fail('SNAPSHOT_RACE', 'Archive changed while it was being verified'); return { digest: hash.digest('hex'), bytes }; } finally { fs.closeSync(fd); } } - private copyAndHash(source: string, destination: string, maxBytes: number, expected: StableStat): { digest: string; bytes: number } { + private copyAndHash(source: string, destination: string, maxBytes: number, expected: StableStat, + control: NormalizedCacheOperationControl): { digest: string; bytes: number } { + checkOperation(control); const sourceFd = openNoFollow(source, fs.constants.O_RDONLY); let destinationFd: number | undefined; try { @@ -403,6 +496,7 @@ export class PublicArchiveCache { destinationFd = openNoFollow(destination, fs.constants.O_WRONLY | fs.constants.O_CREAT | fs.constants.O_EXCL, 0o600); let bytes = 0; for (;;) { + checkOperation(control); const read = fs.readSync(sourceFd, buffer, 0, buffer.length, null); if (!read) break; bytes += read; @@ -410,9 +504,17 @@ export class PublicArchiveCache { if (bytes > maxBytes) fail('INSUFFICIENT_CAPACITY', 'Archive exceeded its byte limit during promotion'); hash.update(buffer.subarray(0, read)); let offset = 0; - while (offset < read) offset += fs.writeSync(destinationFd, buffer, offset, read - offset); + while (offset < read) { + checkOperation(control); + const written = fs.writeSync(destinationFd, buffer, offset, read - offset); + if (written <= 0) fail('PERSISTENCE_FAILED', 'Archive copy stopped before the current chunk was written'); + offset += written; + checkOperation(control); + } } + checkOperation(control); fs.fsyncSync(destinationFd); + checkOperation(control); const after = stableStat(fs.fstatSync(sourceFd)); if (!sameStat(before, after) || bytes !== before.size) fail('SNAPSHOT_RACE', 'Staged archive changed during promotion'); return { digest: hash.digest('hex'), bytes }; @@ -422,28 +524,39 @@ export class PublicArchiveCache { } } - private inventory(): Metadata[] { - const entryNames = fs.readdirSync(this.entriesDir), metadataNames = fs.readdirSync(this.metadataDir); - for (const name of entryNames) if (!SHA256.test(name)) fail('INCOMPATIBLE_INPUT', 'Cache entries directory contains an unexpected object'); - for (const name of metadataNames) if (!/^[a-f0-9]{64}\.json$/.test(name)) fail('INCOMPATIBLE_INPUT', 'Cache metadata directory contains an unexpected object'); - const entrySet = new Set(entryNames), metadataSet = new Set(metadataNames.map(name => name.slice(0, -5))); - if (entrySet.size !== metadataSet.size || [...entrySet].some(name => !metadataSet.has(name))) + private inventory(control: NormalizedCacheOperationControl): Metadata[] { + checkOperation(control); + const entryNames = boundedDirectoryNames(this.entriesDir, 'Cache entries directory', control), + metadataNames = boundedDirectoryNames(this.metadataDir, 'Cache metadata directory', control); + const entrySet=new Set(),metadataSet=new Set(); + for (const name of entryNames) {checkOperation(control);if (!SHA256.test(name)) fail('INCOMPATIBLE_INPUT', 'Cache entries directory contains an unexpected object');entrySet.add(name);} + for (const name of metadataNames) {checkOperation(control);if (!/^[a-f0-9]{64}\.json$/.test(name)) fail('INCOMPATIBLE_INPUT', 'Cache metadata directory contains an unexpected object');metadataSet.add(name.slice(0,-5));} + if (entrySet.size !== metadataSet.size) fail('INCOMPATIBLE_INPUT', 'Cache entries and metadata are inconsistent'); + for(const name of entrySet){checkOperation(control);if(!metadataSet.has(name)) fail('INCOMPATIBLE_INPUT', 'Cache entries and metadata are inconsistent'); - return entryNames.map(digest => { + } + const inventory = entryNames.map(digest => { + checkOperation(control); const stat = fs.lstatSync(this.entryPath(digest)); assertOwnedRegular(stat, 'Cached archive', this.maxEntryBytes, true); - const metadata = readMetadata(this.metadataPath(digest), digest); + const metadata = readMetadata(this.metadataPath(digest), digest, control); if (metadata.bytes !== stat.size) fail('INCOMPATIBLE_INPUT', `Cache size metadata is inconsistent for ${digest}`); return metadata; }); + checkOperation(control); + return inventory; } - private evictToFit(incomingBytes: number): void { + private evictToFit(incomingBytes: number, control: NormalizedCacheOperationControl): void { if (!Number.isSafeInteger(incomingBytes) || incomingBytes < 0 || incomingBytes > this.maxBytes) fail('INSUFFICIENT_CAPACITY', 'Archive cannot fit within the public-cache limit'); - const entries = this.inventory().sort((a, b) => a.lastAccessedAt - b.lastAccessedAt || a.createdAt - b.createdAt || a.sha256.localeCompare(b.sha256)); - let total = entries.reduce((sum, item) => sum + item.bytes, 0); + checkOperation(control); + const entries = this.inventory(control);checkOperation(control);entries.sort((a, b) => a.lastAccessedAt - b.lastAccessedAt || a.createdAt - b.createdAt || a.sha256.localeCompare(b.sha256)); + checkOperation(control); + let total = 0; + for (const item of entries) { checkOperation(control); total += item.bytes; } for (const item of entries) { + checkOperation(control); if (total + incomingBytes <= this.maxBytes) break; const archive = this.moveToRecovery(this.entryPath(item.sha256), 'entry', item.sha256); const metadata = this.moveToRecovery(this.metadataPath(item.sha256), 'metadata', item.sha256); @@ -454,18 +567,26 @@ export class PublicArchiveCache { if (total + incomingBytes > this.maxBytes) fail('INSUFFICIENT_CAPACITY', 'Archive cache could not free enough verified capacity'); } - private cleanIncoming(): void { - for (const name of fs.readdirSync(this.incomingDir)) { + private cleanIncoming(control: NormalizedCacheOperationControl): void { + const incomingNames = boundedDirectoryNames(this.incomingDir, 'Cache incoming directory', control); + let publishedByInode:Map|undefined; + for (const name of incomingNames) { + checkOperation(control); if (!/^\.incoming-\d+-[a-f0-9]{24}$/.test(name)) fail('INCOMPATIBLE_INPUT', 'Cache incoming directory contains an unexpected object'); const incoming = join(this.incomingDir, name), stat = fs.lstatSync(incoming); if (!stat.isFile() || stat.isSymbolicLink() || ![1, 2].includes(stat.nlink) || stat.size > this.maxEntryBytes || (process.getuid && stat.uid !== process.getuid())) fail('UNSAFE_PATH', 'Incomplete cache archive is not a bounded regular file'); if (stat.nlink === 2) { - const matches = fs.readdirSync(this.entriesDir).filter(entry => { - if (!SHA256.test(entry)) fail('INCOMPATIBLE_INPUT', 'Cache entries directory contains an unexpected object'); - const candidate = fs.lstatSync(this.entryPath(entry)); - return candidate.dev === stat.dev && candidate.ino === stat.ino; - }); + if(!publishedByInode){ + publishedByInode=new Map(); + for(const entry of boundedDirectoryNames(this.entriesDir, 'Cache entries directory', control)){ + checkOperation(control); + if (!SHA256.test(entry)) fail('INCOMPATIBLE_INPUT', 'Cache entries directory contains an unexpected object'); + const candidate=fs.lstatSync(this.entryPath(entry)),key=`${candidate.dev}:${candidate.ino}`,matches=publishedByInode.get(key)??[]; + matches.push(entry);publishedByInode.set(key,matches); + } + } + const matches=publishedByInode.get(`${stat.dev}:${stat.ino}`)??[]; if (matches.length !== 1) fail('UNSAFE_PATH', 'Incoming archive hard link does not match one published cache entry'); const target = fs.lstatSync(this.entryPath(matches[0])); if (!target.isFile() || target.isSymbolicLink() || target.nlink !== 2 || target.size > this.maxEntryBytes || @@ -480,23 +601,29 @@ export class PublicArchiveCache { * Recover only artifacts whose names and inode types prove they belong to an * interrupted cache transaction. Unknown objects remain a hard failure. */ - private recoverInterruptedOperations(): void { - for (const name of fs.readdirSync(this.recoveryDir)) { + private recoverInterruptedOperations(control: NormalizedCacheOperationControl): void { + for (const name of boundedDirectoryNames(this.recoveryDir, 'Cache recovery directory', control)) { + checkOperation(control); if (!/^\.recovery-(?:entry|metadata)-[a-f0-9]{64}-\d+-[a-f0-9]{24}$/.test(name)) fail('INCOMPATIBLE_INPUT', 'Cache recovery directory contains an unexpected object'); removeRegular(join(this.recoveryDir, name), 'Interrupted cache transaction'); } - const entries = fs.readdirSync(this.entriesDir), metadataObjects = fs.readdirSync(this.metadataDir); + const entries = boundedDirectoryNames(this.entriesDir, 'Cache entries directory', control), + metadataObjects = boundedDirectoryNames(this.metadataDir, 'Cache metadata directory', control); for (const name of metadataObjects) { - if (/^[a-f0-9]{64}\.json\.tmp\.\d+\.[a-f0-9]{8}$/.test(name)) this.recoverMetadataTemp(name); + checkOperation(control); + if (/^[a-f0-9]{64}\.json\.tmp\.\d+\.[a-f0-9]{8}$/.test(name)) this.recoverMetadataTemp(name, control); else if (!/^[a-f0-9]{64}\.json$/.test(name)) fail('INCOMPATIBLE_INPUT', 'Cache metadata directory contains an unexpected object'); } - const metadata = fs.readdirSync(this.metadataDir); + const metadata = boundedDirectoryNames(this.metadataDir, 'Cache metadata directory', control); for (const name of entries) if (!SHA256.test(name)) fail('INCOMPATIBLE_INPUT', 'Cache entries directory contains an unexpected object'); for (const name of metadata) if (!/^[a-f0-9]{64}\.json$/.test(name)) fail('INCOMPATIBLE_INPUT', 'Cache metadata directory contains an unexpected object'); - const entrySet = new Set(entries), metadataSet = new Set(metadata.map(name => name.slice(0, -5))); - for (const digest of new Set([...entrySet, ...metadataSet])) { + const entrySet=new Set(),metadataSet=new Set(),digests=new Set(); + for(const name of entries){checkOperation(control);entrySet.add(name);digests.add(name);} + for(const name of metadata){checkOperation(control);const digest=name.slice(0,-5);metadataSet.add(digest);digests.add(digest);} + for (const digest of digests) { + checkOperation(control); if (entrySet.has(digest) === metadataSet.has(digest)) continue; if (entrySet.has(digest)) this.discardObject(this.entryPath(digest), 'entry', digest); else this.discardObject(this.metadataPath(digest), 'metadata', digest); @@ -526,7 +653,8 @@ export class PublicArchiveCache { removeRegular(quarantined, 'Interrupted cache transaction'); } - private recoverMetadataTemp(name: string): void { + private recoverMetadataTemp(name: string, control: NormalizedCacheOperationControl): void { + checkOperation(control); const path = join(this.metadataDir, name), stat = fs.lstatSync(path); if (!stat.isFile() || stat.isSymbolicLink() || ![1, 2].includes(stat.nlink) || stat.size > METADATA_LIMIT || (process.getuid && stat.uid !== process.getuid()) || (stat.mode & 0o077) !== 0) @@ -541,14 +669,16 @@ export class PublicArchiveCache { try { fs.unlinkSync(path); } catch { fail('PERSISTENCE_FAILED', 'Interrupted cache metadata write could not be removed'); } } - private withLock(callback: () => T): T { + private withLock(callback: () => T, control: NormalizedCacheOperationControl): T { + checkOperation(control); const marker = `${JSON.stringify({ protocol: CACHE_LOCK_PROTOCOL })}\n`; const options={label:'Cache lock protocol',maxBytes:METADATA_LIMIT,validate:(value:unknown)=>{ if(!value||typeof value!=='object'||Array.isArray(value)||Object.keys(value).join(',')!=='protocol'||(value as any).protocol!==CACHE_LOCK_PROTOCOL) fail('INCOMPATIBLE_INPUT','Archive-cache lock protocol is invalid'); }}; const tempPattern=/^\.lock\.tmp\.(\d{1,10})\.[a-f0-9]{8}$/; - for(const name of fs.readdirSync(this.root)){ + for(const name of boundedDirectoryNames(this.root, 'Cache root directory', control)){ + checkOperation(control); const match=name.match(tempPattern);if(!match)continue; const temporary=join(this.root,name),publisherPid=Number(match[1]); if(existsNoFollow(this.lockDir))recoverAtomicNoReplaceJson(this.lockDir,options); @@ -566,7 +696,7 @@ export class PublicArchiveCache { let protocol:unknown;try{protocol=JSON.parse(fs.readFileSync(this.lockDir,'utf8')).protocol;}catch{} if(protocol!==CACHE_LOCK_PROTOCOL)fail('INCOMPATIBLE_INPUT','Archive-cache lock protocol is invalid'); } - const result=withStateLock(this.root,callback); + const result=withStateLock(this.root,()=>{checkOperation(control);return callback();}); if(result&&typeof (result as any).then==='function')fail('PERSISTENCE_FAILED','Archive-cache operation unexpectedly became asynchronous'); return result as T; } diff --git a/lib/cso/cli.ts b/lib/cso/cli.ts index 4ecb8da95..9c9bf71e4 100644 --- a/lib/cso/cli.ts +++ b/lib/cso/cli.ts @@ -26,6 +26,7 @@ import { historyForPath } from './history'; import { catalogImageProvisioningPolicy, inspectCatalogImages, openLocalCatalogImageSession, provisionCatalogImages, qualifiedCatalogImages, type CatalogImageSessionFactory } from './image-provisioning'; const VERSION = '3.0.0'; +const RETENTION_MAINTENANCE_MS=1_000,RETENTION_MAX_ENTRIES=100_000,REPLAY_LOOKUP_MAX_ENTRIES=10_000; const productionCatalogImageSession:CatalogImageSessionFactory=deadline=>openLocalCatalogImageSession(process.env,deadline); /** Internal qualification seam. The public entrypoint below always supplies committed dependencies. */ export interface CsoCliDependencies { @@ -145,8 +146,8 @@ function parseStart(args:string[]):{repo:string;policy:RunPolicy;comparisonBase? const scope=named?`domain:${string(named,'scope',100)}`:scopeFlags[0]?.slice(2)??'default'; return {repo,policy:{mode:comprehensive?'comprehensive':'daily',scope,diff,base,offline,budgetSeconds,maxWorkers:3,maxRepairs:3},...(diff||explicitBase?{comparisonBase:base}:{})}; } -async function start(args:string[],dependencies:CsoCliDependencies,parent?:RunReportV3['parent'],requiredAncestor?:string):Promise{ - const {repo,policy,comparisonBase}=parseStart(args);assertStateOutside(repo);if(!parent)retention();const createdAt=new Date(),run=newRun(repo); +async function start(args:string[],dependencies:CsoCliDependencies,parent?:RunReportV3['parent'],requiredAncestor?:string,startedAt=new Date()):Promise{ + const {repo,policy,comparisonBase}=parseStart(args),createdAt=startedAt;assertStateOutside(repo);if(!parent)retention(createdAt.getTime(),{deadlineMs:Math.min(createdAt.getTime()+RETENTION_MAINTENANCE_MS,createdAt.getTime()+policy.budgetSeconds*1000-60_000),maxEntries:RETENTION_MAX_ENTRIES});const run=newRun(repo); let manifest:Awaited>;try{manifest=await capture(repo,run.dir,comparisonBase,requiredAncestor,{deadlineMs:createdAt.getTime()+policy.budgetSeconds*1000-60_000});}catch(error){fs.rmSync(run.dir,{recursive:true,force:true});throw error;} const report:RunReportV3={schemaVersion:3,runId:run.runId,repoId:run.repoId,createdAt:createdAt.toISOString(),deadline:new Date(createdAt.getTime()+policy.budgetSeconds*1000).toISOString(),status:'running',completeness:'not assessed',policy, source:{root:repo,snapshotHash:manifest.executionHash,originalHash:manifest.originalHash,baseCommit:manifest.baseCommit, @@ -362,11 +363,11 @@ async function scanner(args:string[],sarif=false){ }); } function scannerOutcome(args:string[]){const {dir}=run(args);if(args.length!==1||!/^[a-z0-9-]{1,40}-[a-f0-9]{16}-[a-f0-9]{16}$/.test(args[0]))throw new CsoError('INVALID_ARGUMENT','scanner-outcome requires one immutable scanner artifact ID');return readJson(join(dir,'scanner-outcomes',`${args[0]}.json`));} -async function recheck(args:string[],dependencies:CsoCliDependencies){if(!args.length)throw new CsoError('INVALID_ARGUMENT','recheck requires a finding ID');retention();const findingId=args.shift()!,runId=need(args,'--run'),originalDir=runDirectory(runId);return await withLock(originalDir,async()=>{const original=loadReport(originalDir),finding=original.findings.find(f=>f.id===findingId);if(!finding)throw new CsoError('MISSING_INPUT','Original finding does not exist');if(original.status!=='finished')throw new CsoError('INVALID_SCHEMA','Recheck requires a finished original audit'); +async function recheck(args:string[],dependencies:CsoCliDependencies){const startedAt=new Date();if(!args.length)throw new CsoError('INVALID_ARGUMENT','recheck requires a finding ID');retention(startedAt.getTime(),{deadlineMs:startedAt.getTime()+RETENTION_MAINTENANCE_MS,maxEntries:RETENTION_MAX_ENTRIES});const findingId=args.shift()!,runId=need(args,'--run'),originalDir=runDirectory(runId);return await withLock(originalDir,async()=>{const original=loadReport(originalDir),finding=original.findings.find(f=>f.id===findingId);if(!finding)throw new CsoError('MISSING_INPUT','Original finding does not exist');if(original.status!=='finished')throw new CsoError('INVALID_SCHEMA','Recheck requires a finished original audit'); // Keep the original immutable while the fresh snapshot is captured and // until its child lineage report has been durably published. const repo=callerPath(need(args,'--repo'));if(args.length)throw new CsoError('INVALID_ARGUMENT',`Unknown recheck argument: ${args[0]}`);if(repoId(repo)!==original.repoId)throw new CsoError('INCOMPATIBLE_INPUT','Recheck must use the original audited repository');const oldManifest=readJson(join(originalDir,'snapshot.json')); - const preserveBase=original.policy.diff||Boolean(original.source.baseCommit),report=await start(['--repo',repo,...(original.policy.mode==='comprehensive'?['--comprehensive']:[]),...(original.policy.diff?['--diff']:[]),...(preserveBase?['--base',original.policy.base]:[]),'--budget',String(original.policy.budgetSeconds),...(original.policy.offline?['--offline']:[]),...(original.policy.scope==='default'?[]:original.policy.scope.startsWith('domain:')?['--scope',original.policy.scope.slice(7)]:[`--${original.policy.scope}`])],dependencies,{runId,findingId,kind:'recheck'},oldManifest.headCommit); + const preserveBase=original.policy.diff||Boolean(original.source.baseCommit),report=await start(['--repo',repo,...(original.policy.mode==='comprehensive'?['--comprehensive']:[]),...(original.policy.diff?['--diff']:[]),...(preserveBase?['--base',original.policy.base]:[]),'--budget',String(original.policy.budgetSeconds),...(original.policy.offline?['--offline']:[]),...(original.policy.scope==='default'?[]:original.policy.scope.startsWith('domain:')?['--scope',original.policy.scope.slice(7)]:[`--${original.policy.scope}`])],dependencies,{runId,findingId,kind:'recheck'},oldManifest.headCommit,startedAt); return {runId:report.runId,parent:report.parent};});} function recordReview(args:string[]){ @@ -453,11 +454,16 @@ async function verify(args:string[],dependencies:CsoCliDependencies){const {dir} finding.reproduction=result.manifest.before.security==='intended_failure'&&result.manifest.before.booted&&result.manifest.before.legitimate?'reproduced':result.manifest.before.security==='pass'?'disproved':result.manifest.before.booted?'inconclusive':'blocked'; finding.repair=result.manifest.result==='tested'?'tested':result.manifest.result==='runtime_tested'?'runtime_tested':'failed';if(['tested','runtime_tested'].includes(result.manifest.result)){finding.verificationId=result.manifest.id;finding.verificationAssurance={assertions:result.manifest.assertionAssurance!,testCompletion:result.manifest.testCompletionAssurance,review:result.manifest.reviewAssurance};delete finding.reproductionAttemptId;}event(report,'verification',`${result.manifest.id}: ${result.manifest.result}; assertion assurance ${result.manifest.assertionAssurance}; test completion assurance ${result.manifest.testCompletionAssurance}; review assurance ${result.manifest.reviewAssurance}`);saveReport(dir,report);return{result:result.manifest.result,verification:result.manifest,bundle:`bundles/${result.bundle.id}.json`};});} function replayBundle(stored:{path:string;dir:string},id:string):any{const bundle=validateRepairBundle(readJson(stored.path),id),report=loadReport(stored.dir),finding=report.findings.find(f=>f.verificationId===id&&['tested','runtime_tested'].includes(f.repair));if(!['tested','runtime_tested'].includes(bundle.verification.result)||!finding)throw new CsoError('INCOMPATIBLE_INPUT','Only a helper-recorded runtime-tested or host-reviewed repair bundle can be replayed');return bundle;} -async function withReplayBundle(id:string,fn:(stored:{path:string;dir:string},bundle:any)=>Promise):Promise{if(!/^[a-f0-9]{32}$/.test(id))throw new CsoError('INVALID_ARGUMENT','Bundle identifier must be the 32-character ID returned by verify');const root=privateRoot();for(const repo of fs.readdirSync(root).filter(x=>/^[a-f0-9]{24}$/.test(x)))for(const run of fs.readdirSync(join(root,repo)).filter(x=>/^\d{13}-[a-f0-9]{16}$/.test(x))){const stored={dir:join(root,repo,run),path:join(root,repo,run,'bundles',`${id}.json`)};if(!fs.existsSync(stored.path))continue;let matched=false,value!:T;await withLock(stored.dir,async()=>{if(!fs.existsSync(stored.path))return;const bundle=replayBundle(stored,id);matched=true;value=await fn(stored,bundle);});if(matched)return value;}throw new CsoError('MISSING_INPUT','Repair bundle was not found or expired');} +async function withReplayBundle(id:string,deadline:number,fn:(stored:{path:string;dir:string},bundle:any)=>Promise):Promise{ + if(!/^[a-f0-9]{32}$/.test(id))throw new CsoError('INVALID_ARGUMENT','Bundle identifier must be the 32-character ID returned by verify'); + let visited=0,stored:{path:string;dir:string}|undefined;const admit=()=>{if(Date.now()>=deadline)throw new CsoError('DEADLINE','Replay exhausted its five-minute budget while locating the recorded bundle');if(++visited>REPLAY_LOOKUP_MAX_ENTRIES)throw new CsoError('INSUFFICIENT_CAPACITY',`Replay bundle lookup exceeded ${REPLAY_LOOKUP_MAX_ENTRIES} private state entries`);}; + const root=privateRoot(),repos=fs.opendirSync(root);try{let repo:fs.Dirent|null;search:while((repo=repos.readSync())!==null){admit();if(!repo.isDirectory()||!/^[a-f0-9]{24}$/.test(repo.name))continue;const repoDir=join(root,repo.name),runs=fs.opendirSync(repoDir);try{let run:fs.Dirent|null;while((run=runs.readSync())!==null){admit();if(!run.isDirectory()||!/^\d{13}-[a-f0-9]{16}$/.test(run.name))continue;const candidate={dir:join(repoDir,run.name),path:join(repoDir,run.name,'bundles',`${id}.json`)};if(fs.existsSync(candidate.path)){stored=candidate;break search;}}}finally{runs.closeSync();}}}finally{repos.closeSync();} + if(!stored)throw new CsoError('MISSING_INPUT','Repair bundle was not found or expired');let matched=false,value!:T;await withLock(stored.dir,async()=>{if(!fs.existsSync(stored!.path))return;const bundle=replayBundle(stored!,id);matched=true;value=await fn(stored!,bundle);});if(matched)return value;throw new CsoError('MISSING_INPUT','Repair bundle was not found or expired'); +} function replayManifestValue(manifest:any):unknown{const {id:_,createdAt:__,witnessHash:___,before,after,...stable}=manifest,observation=(value:any)=>{const{output:_,...rest}=value;return rest;};return{...stable,before:observation(before),after:observation(after)};} -async function replay(args:string[],dependencies:CsoCliDependencies){if(!args.length)throw new CsoError('INVALID_ARGUMENT','replay requires a bundle ID');const id=args.shift()!,source=args.includes('--source')?callerPath(need(args,'--source')):undefined;if(args.length)throw new CsoError('INVALID_ARGUMENT',`Unknown replay argument: ${args[0]}`);return await withReplayBundle(id,async(stored,bundle)=>{ +async function replay(args:string[],dependencies:CsoCliDependencies){const replayStarted=Date.now(),replayDeadline=replayStarted+300_000;retention(replayStarted,{deadlineMs:replayStarted+RETENTION_MAINTENANCE_MS,maxEntries:RETENTION_MAX_ENTRIES});if(!args.length)throw new CsoError('INVALID_ARGUMENT','replay requires a bundle ID');const id=args.shift()!,source=args.includes('--source')?callerPath(need(args,'--source')):undefined;if(args.length)throw new CsoError('INVALID_ARGUMENT',`Unknown replay argument: ${args[0]}`);return await withReplayBundle(id,replayDeadline,async(stored,bundle)=>{ if(bundle.requiredInputs.archives?.length&&!bundle.requiredInputs.dependencyClosures)throw new CsoError('MISSING_INPUT','Replay bundle predates retained dependency closures and cannot substitute current dependency state'); - let workDir=stored.dir,manifest:any,temporary:string|undefined;const retained=join(stored.dir,'snapshot'),replayDeadline=Date.now()+300_000; + let workDir=stored.dir,manifest:any,temporary:string|undefined;const retained=join(stored.dir,'snapshot'); try{ if(fs.existsSync(retained)){manifest=readJson(join(stored.dir,'snapshot.json'));assertSnapshot(stored.dir,manifest);}else if(source){const temp=newRun(source);temporary=temp.dir;workDir=temp.dir;manifest=await capture(source,workDir,undefined,undefined,{deadlineMs:replayDeadline});if(manifest.executionHash!==bundle.requiredInputs.sourceHash||manifest.originalHash!==bundle.requiredInputs.originalHash)throw new CsoError('INCOMPATIBLE_INPUT','Supplied source does not match the bundle input hashes');}else throw new CsoError('MISSING_INPUT','Retained source expired; supply explicitly matching source'); if(manifest.executionHash!==bundle.requiredInputs.sourceHash||manifest.originalHash!==bundle.requiredInputs.originalHash)throw new CsoError('INCOMPATIBLE_INPUT','Retained source hashes do not match the bundle');validateRepairBundle(bundle,id,join(workDir,'snapshot'),manifest);if(bundle.verification.policyHash!==ISOLATION_POLICY_HASH)throw new CsoError('INCOMPATIBLE_INPUT','Current helper isolation policy does not match the recorded bundle');const targetPlatform=bundle.requiredInputs.platform as 'linux/amd64'|'linux/arm64';let runtime;try{runtime=selectRuntime(bundle.verification.runtime.profile,targetPlatform,dependencies.runtimeCatalog);}catch(error:any){throw new CsoError('PREREQUISITE',error?.message||'Qualified replay runtime is unavailable');}const verifier=runtime;if(runtime.image!==bundle.requiredInputs.runtimeImage)throw new CsoError('INCOMPATIBLE_INPUT','Qualified runtime digest does not match the bundle'); @@ -469,7 +475,7 @@ async function replay(args:string[],dependencies:CsoCliDependencies){if(!args.le } export async function dispatchCsoCommand(command:string,args:string[],dependencies:CsoCliDependencies):Promise{ - args=[...args];if(!['start','recheck','doctor','provision-images'].includes(command))retention(); + args=[...args];if(!['start','recheck','replay','doctor','provision-images'].includes(command)){const started=Date.now();retention(started,{deadlineMs:started+RETENTION_MAINTENANCE_MS,maxEntries:RETENTION_MAX_ENTRIES});} let result:unknown; switch(command){case'start':result=await start(args,dependencies);break;case'doctor':result=await doctor(args,dependencies);break;case'provision-images':result=await provisionImages(args,dependencies);break;case'resume':result=resume(args);break;case'inspect':await inspect(args);return;case'read':await read(args);return;case'history':await history(args);return;case'submit':result=submit(args);break;case'finish':result=finish(args);break;case'import-v2':result=importV2(args);break;case'scan':result=await scanner(args);break;case'scanner-outcome':result=scannerOutcome(args);break;case'import-sarif':result=await scanner(args,true);break;case'record-review':result=recordReview(args);break;case'test-plan':result=testPlan(args);break;case'runtime-plan':result=runtimePlan(args);break;case'recheck':result=await recheck(args,dependencies);break; case'verify':result=await verify(args,dependencies);break;case'replay':result=await replay(args,dependencies);break;case'patch-hash':if(args.length!==1)throw new CsoError('INVALID_ARGUMENT','patch-hash requires one request JSON file');result={patchHash:patchHash(validateVerificationRequest(readInput(args[0])))};break;default:throw new CsoError('INVALID_ARGUMENT',`Unknown command: ${command}`);} diff --git a/lib/cso/preparation-executor.ts b/lib/cso/preparation-executor.ts index 26c96afde..ffa20067d 100644 --- a/lib/cso/preparation-executor.ts +++ b/lib/cso/preparation-executor.ts @@ -22,6 +22,7 @@ import { const SHA256 = /^[a-f0-9]{64}$/; const HOST = /^(?=.{1,253}$)(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)*[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$/; const RELATIVE_ARCHIVE = /^(?!\/)(?!.*(?:^|\/)\.\.?(?:\/|$))(?!.*\\)[A-Za-z0-9@._+\/-]{1,1024}$/; +const MAX_ARCHIVES = 25_000; const MAX_TREE_FILES = 200_000; const MAX_SOURCE_BYTES = 64 * 1024 * 1024; // The smallest automatic execution role is the tests container: 768 MiB at @@ -46,7 +47,8 @@ function deepFreeze(value: T): T { } return value; } -function checkDeadline(deadline: number): void { +function checkDeadline(deadline: number, signal?: AbortSignal): void { + if (signal?.aborted) fail('CANCELLED', 'Preparation was cancelled before the operation completed'); if (!Number.isSafeInteger(deadline) || deadline <= 0) fail('INVALID_ARGUMENT', 'Preparation deadline must be an absolute millisecond timestamp'); if (Date.now() >= deadline) fail('DEADLINE', 'Preparation deadline was reached before the operation completed'); } @@ -376,10 +378,13 @@ function relativeArchivePath(value: string, label: string): string { return value; } -function stagedFileHashes(root: string, relativePath: string, expectedBytes: number): { sha256: string; sha512: string } { +function stagedFileHashes(root: string, relativePath: string, expectedBytes: number, deadline: number, + signal?: AbortSignal): { sha256: string; sha512: string } { + checkDeadline(deadline, signal); const parts = relativeArchivePath(relativePath, 'Staged archive').split('/'); let cursor = root; for (const part of parts.slice(0, -1)) { + checkDeadline(deadline, signal); cursor = join(cursor, part); let stat: fs.Stats; try { stat = fs.lstatSync(cursor); } catch { fail('MISSING_INPUT', 'Acquisition staging directory is missing'); } @@ -399,12 +404,15 @@ function stagedFileHashes(root: string, relativePath: string, expectedBytes: num const h256 = createHash('sha256'), h512 = createHash('sha512'), buffer = Buffer.allocUnsafe(64 * 1024); let bytes = 0; for (;;) { + checkDeadline(deadline, signal); const count = fs.readSync(fd!, buffer, 0, buffer.length, null); if (!count) break; bytes += count; if (bytes > expectedBytes) fail('SNAPSHOT_RACE', 'Staged archive grew while it was verified'); h256.update(buffer.subarray(0, count)); h512.update(buffer.subarray(0, count)); + checkDeadline(deadline, signal); } + checkDeadline(deadline, signal); const after = fs.fstatSync(fd!); if (bytes !== expectedBytes || before.dev !== after.dev || before.ino !== after.ino || before.size !== after.size || before.mtimeMs !== after.mtimeMs || before.ctimeMs !== after.ctimeMs || before.mode !== after.mode || before.nlink !== after.nlink) @@ -423,18 +431,23 @@ function closureIdentity(closure: Omit): strin function logicalInput(input: PreparationInput): string { return `${input.name}\0${input.version}`; } function validateClosure(plan: PreparationPlan, admission: PreparationRuntimeAdmission, closure: DependencyClosure, - cache: PublicArchiveCache, materializationBase?: string): { mounts: CachedArchiveMount[]; materializedRoot?: string } { + cache: PublicArchiveCache, deadline: number, signal?: AbortSignal, + materializationBase?: string): { mounts: CachedArchiveMount[]; materializedRoot?: string } { + checkDeadline(deadline, signal); const runtime = admission.runtime, expectedPlanHash = planHash(plan); const { closureHash, ...closureBody } = closure ?? ({} as DependencyClosure); if (!closure || closure.schemaVersion !== 1 || closure.stack !== plan.stack || closure.planHash !== expectedPlanHash || closure.catalogRevision !== admission.catalogRevision || closure.runtimeId !== runtime.id || closure.runtimeImage !== runtime.image || - closure.platform !== runtime.platform || !Array.isArray(closure.archives) || !SHA256.test(closureHash) || + closure.platform !== runtime.platform || !Array.isArray(closure.archives) || closure.archives.length > MAX_ARCHIVES || !SHA256.test(closureHash) || closureHash !== closureIdentity(closureBody)) fail('INCOMPATIBLE_INPUT', 'Dependency closure does not match the admitted plan and runtime'); const publicInputs = plan.inputs.map((input, index) => ({ input, index })).filter(item => item.input.kind === 'public'); + if (publicInputs.length > MAX_ARCHIVES) fail('INSUFFICIENT_CAPACITY', 'Dependency closure exceeds the archive-count limit'); + const publicInputsByIndex = new Map(publicInputs.map(item => [item.index, item])); const covered = new Set(), paths = new Set(), pending: Array<{ archive: DependencyArchive; input: PreparationInput }> = []; for (const archive of closure.archives) { - const selected = publicInputs.find(item => item.index === archive.inputIndex); + checkDeadline(deadline, signal); + const selected = publicInputsByIndex.get(archive.inputIndex); if (!selected || archive.name !== selected.input.name || archive.version !== selected.input.version || archive.declaredIntegrity !== (selected.input.integrity ?? 'registry-on-acquisition') || !SHA256.test(archive.sha256) || !Number.isSafeInteger(archive.bytes) || archive.bytes < 0 || !plan.registryHosts.includes(archive.requestedHost)) @@ -446,8 +459,11 @@ function validateClosure(plan: PreparationPlan, admission: PreparationRuntimeAdm paths.add(archive.installPath); covered.add(logicalInput(selected.input)); pending.push({ archive, input: selected.input }); } - for (const { input } of publicInputs) if (!covered.has(logicalInput(input))) - fail('MISSING_INPUT', `Dependency closure does not contain a compatible public archive for ${input.name}@${input.version}`); + for (const { input } of publicInputs) { + checkDeadline(deadline, signal); + if (!covered.has(logicalInput(input))) + fail('MISSING_INPUT', `Dependency closure does not contain a compatible public archive for ${input.name}@${input.version}`); + } if (!pending.length) return { mounts: [] }; let base: string; if (materializationBase) { @@ -459,8 +475,11 @@ function validateClosure(plan: PreparationPlan, admission: PreparationRuntimeAdm const materializedRoot = fs.mkdtempSync(join(base, 'gstack-cso-archives-')); fs.chmodSync(materializedRoot, 0o700); try { - const copies = cache.materialize(pending.map(item => item.archive.sha256), materializedRoot), byDigest = new Map(copies.map(copy => [copy.sha256, copy])); + checkDeadline(deadline, signal); + const copies = cache.materialize(pending.map(item => item.archive.sha256), materializedRoot, { deadline, signal }), + byDigest = new Map(copies.map(copy => [copy.sha256, copy])); const mounts = pending.map(({ archive }) => { + checkDeadline(deadline, signal); const copy = byDigest.get(archive.sha256); if (!copy || copy.bytes !== archive.bytes) fail('MISSING_INPUT', `Verified public archive is missing from the offline cache: ${archive.name}@${archive.version}`); return { inputIndex: archive.inputIndex, name: archive.name, version: archive.version, @@ -475,7 +494,26 @@ function validateClosure(plan: PreparationPlan, admission: PreparationRuntimeAdm } type TreeEntry = { path: string; kind: 'file' | 'symlink'; mode: number; bytes: number; sha256: string }; -function treeManifest(rootPath: string, maxBytes: number, deadline: number, allowContainedSymlinks = false): TreeEntry[] { +function boundedTreeNames(directory: string, deadline: number, signal?: AbortSignal): string[] { + checkDeadline(deadline, signal); + const handle = fs.opendirSync(directory), names: string[] = []; + try { + for (;;) { + checkDeadline(deadline, signal); + const entry = handle.readSync(); + if (!entry) break; + if (names.length >= MAX_TREE_FILES) fail('INSUFFICIENT_CAPACITY', 'Preparation directory exceeds its bounded manifest limit'); + names.push(entry.name); + } + } finally { handle.closeSync(); } + checkDeadline(deadline, signal); + names.sort(); + checkDeadline(deadline, signal); + return names; +} +function treeManifest(rootPath: string, maxBytes: number, deadline: number, allowContainedSymlinks = false, + signal?: AbortSignal): TreeEntry[] { + checkDeadline(deadline, signal); const root = resolve(rootPath); let rootStat: fs.Stats, canonicalRoot: string; try { rootStat = fs.lstatSync(root); canonicalRoot = fs.realpathSync(root); } @@ -486,9 +524,10 @@ function treeManifest(rootPath: string, maxBytes: number, deadline: number, allo const entries: TreeEntry[] = []; let total = 0, nodes = 0; const walk = (directory: string, prefix = ''): void => { - checkDeadline(deadline); - const before = fs.readdirSync(directory).sort(); + checkDeadline(deadline, signal); + const before = boundedTreeNames(directory, deadline, signal); for (const name of before) { + checkDeadline(deadline, signal); const path = join(directory, name), relativePath = prefix ? `${prefix}/${name}` : name; nodes++; if (nodes > MAX_TREE_FILES || Buffer.byteLength(name) > 255 || Buffer.byteLength(relativePath) > 4096) @@ -525,7 +564,8 @@ function treeManifest(rootPath: string, maxBytes: number, deadline: number, allo try { const initial = fs.fstatSync(fd!), hash = createHash('sha256'), buffer = Buffer.allocUnsafe(64 * 1024); let readBytes = 0; - for (;;) { const count = fs.readSync(fd!, buffer, 0, buffer.length, null); if (!count) break; readBytes += count; hash.update(buffer.subarray(0, count)); checkDeadline(deadline); } + for (;;) { checkDeadline(deadline, signal); const count = fs.readSync(fd!, buffer, 0, buffer.length, null); if (!count) break; readBytes += count; hash.update(buffer.subarray(0, count)); checkDeadline(deadline, signal); } + checkDeadline(deadline, signal); const final = fs.fstatSync(fd!); if (readBytes !== initial.size || initial.dev !== final.dev || initial.ino !== final.ino || initial.size !== final.size || initial.mode !== final.mode || initial.mtimeMs !== final.mtimeMs || initial.ctimeMs !== final.ctimeMs) @@ -533,13 +573,15 @@ function treeManifest(rootPath: string, maxBytes: number, deadline: number, allo entries.push({ path: relativePath, kind: 'file', mode: initial.mode & 0o777, bytes: initial.size, sha256: hash.digest('hex') }); } finally { fs.closeSync(fd!); } } - if (canonical(before) !== canonical(fs.readdirSync(directory).sort())) fail('SNAPSHOT_RACE', 'Preparation tree membership changed while hashing'); + checkDeadline(deadline, signal); + if (canonical(before) !== canonical(boundedTreeNames(directory, deadline, signal))) fail('SNAPSHOT_RACE', 'Preparation tree membership changed while hashing'); }; walk(root); return entries; } -function treeHash(rootPath: string, maxBytes: number, deadline: number, allowContainedSymlinks = false): string { - return sha256(canonical(treeManifest(rootPath, maxBytes, deadline, allowContainedSymlinks))); +function treeHash(rootPath: string, maxBytes: number, deadline: number, allowContainedSymlinks = false, + signal?: AbortSignal): string { + return sha256(canonical(treeManifest(rootPath, maxBytes, deadline, allowContainedSymlinks, signal))); } function dependencyOutput(stack: CsoStack, path: string): boolean { const first = path.split('/')[0]; @@ -567,13 +609,14 @@ function preparedEnvironment(plan: PreparationPlan): Record { return environment; } function provePreparedProjection(snapshot: string, preparedRoot: string, stack: CsoStack, - transformations: OfflinePreparationRequest['transformations'], deadline: number): { hash: string; transformations: PreparedApplication['transformations']; manifestHash: string; dependencyHash:string } { - const source = treeManifest(snapshot, MAX_SOURCE_BYTES, deadline), prepared = treeManifest(preparedRoot, MAX_PREPARED_BYTES, deadline, true), + transformations: OfflinePreparationRequest['transformations'], deadline: number, signal?: AbortSignal): { hash: string; transformations: PreparedApplication['transformations']; manifestHash: string; dependencyHash:string } { + const source = treeManifest(snapshot, MAX_SOURCE_BYTES, deadline, false, signal), prepared = treeManifest(preparedRoot, MAX_PREPARED_BYTES, deadline, true, signal), sourceByPath = new Map(source.map(entry => [entry.path, entry])), preparedByPath = new Map(prepared.map(entry => [entry.path, entry])), synthetic = new Map(transformations.map(item => [item.path, item])); if (synthetic.size !== transformations.length) fail('INVALID_SCHEMA', 'Offline preparation transformations contain duplicate paths'); const actualTransformations: PreparedApplication['transformations'] = []; for (const entry of source) { + checkDeadline(deadline, signal); const actual = preparedByPath.get(entry.path), transformed = synthetic.get(entry.path); if (!actual || actual.kind !== 'file') fail('ISOLATION_FAILED', `Offline lifecycle execution removed or replaced captured source: ${entry.path}`); if (transformed) { @@ -581,12 +624,14 @@ function provePreparedProjection(snapshot: string, preparedRoot: string, stack: } else if (canonical(actual) !== canonical(entry)) fail('ISOLATION_FAILED', `Offline lifecycle execution changed captured source bytes or mode: ${entry.path}`); } for (const transformed of transformations) { + checkDeadline(deadline, signal); const actual = preparedByPath.get(transformed.path); if (!actual || actual.kind !== 'file' || actual.sha256 !== transformed.sha256) fail('ISOLATION_FAILED', `Offline preparation did not preserve its declared transformation: ${transformed.path}`); actualTransformations.push({ path: transformed.path, sha256: transformed.sha256, mode: actual.mode, reason: transformed.reason }); } for (const entry of prepared) { + checkDeadline(deadline, signal); if (sourceByPath.has(entry.path) || synthetic.has(entry.path) || dependencyOutput(stack, entry.path)) continue; fail('ISOLATION_FAILED', `Offline lifecycle execution wrote outside its dependency roots: ${entry.path}`); } @@ -639,18 +684,23 @@ export class PreparationExecutor { admission: PreparationRuntimeAdmission; snapshot: string; deadline: number; + signal?: AbortSignal; offline?: boolean; existingClosure?: DependencyClosure; }): Promise { - checkDeadline(options.deadline); + checkDeadline(options.deadline, options.signal); const hash = validatePlan(options.plan, options.snapshot), runtime = runtimeFromAdmission(options.plan, options.admission); + checkDeadline(options.deadline, options.signal); validateRunner(this.options.runner, options.plan.stack); if (options.existingClosure) { - const verified = validateClosure(options.plan, options.admission, options.existingClosure, this.options.cache, this.options.materializationRoot); + const verified = validateClosure(options.plan, options.admission, options.existingClosure, this.options.cache, + options.deadline, options.signal, this.options.materializationRoot); if (verified.materializedRoot) fs.rmSync(verified.materializedRoot, { recursive: true, force: true }); return options.existingClosure; } const publicInputs = options.plan.inputs.map((input, index) => ({ input, index })).filter(item => item.input.kind === 'public'); + if (publicInputs.length > MAX_ARCHIVES) fail('INSUFFICIENT_CAPACITY', 'Preparation plan exceeds the archive-count limit'); + const publicInputsByIndex = new Map(publicInputs.map(item => [item.index, item])); if (options.offline && publicInputs.length) fail('MISSING_INPUT', 'Offline preparation requires a matching retained dependency closure and verified cache entries'); if (!publicInputs.length) { const body: Omit = { schemaVersion: 1, stack: options.plan.stack, planHash: hash, @@ -665,14 +715,14 @@ export class PreparationExecutor { schemaVersion: 1, planHash: hash, stack: options.plan.stack, runtime: { id: runtime.id, image: runtime.image, platform: runtime.platform }, metadata: structuredClone(options.plan.metadata), inputs: structuredClone(publicInputs), commands: structuredClone(options.plan.acquisition), stagingRoot: acquisitionStaging, - deadline: options.deadline, limits: { maxArchives: 25_000, + deadline: options.deadline, limits: { maxArchives: MAX_ARCHIVES, maxArchiveBytes: Math.min(this.options.cache.maxEntryBytes, MAX_ACQUISITION_ARCHIVE_BYTES), maxTotalArchiveBytes: Math.min(this.options.cache.maxBytes, MAX_ACQUISITION_ARCHIVE_BYTES), maxOutputBytes: MAX_OUTPUT }, network: { mode: 'registry-restricted', allowedHosts: [...options.plan.registryHosts] }, sourceMounted: false, }; try { const receipt = await this.options.runner.acquire(deepFreeze(request)); - checkDeadline(options.deadline); + checkDeadline(options.deadline, options.signal); validateAcquisitionReceipt(receipt, request, options.plan); if (!Array.isArray(receipt.artifacts) || receipt.artifacts.length > request.limits.maxArchives) fail('TOOL_FAILED', 'Acquisition receipt contains an invalid number of archives'); @@ -680,7 +730,8 @@ export class PreparationExecutor { uniqueBytes = new Map(), validated: Array<{ artifact: AcquisitionArtifactReceipt; selected: typeof publicInputs[number] }> = []; let totalUniqueBytes = 0; for (const artifact of receipt.artifacts) { - const selected = publicInputs.find(item => item.index === artifact.inputIndex); + checkDeadline(options.deadline, options.signal); + const selected = publicInputsByIndex.get(artifact.inputIndex); if (!selected || seenInputs.has(artifact.inputIndex) || !SHA256.test(artifact.sha256) || artifact.registryResponseSha256 !== artifact.sha256 || !Number.isSafeInteger(artifact.bytes) || artifact.bytes < 0 || artifact.bytes > this.options.cache.maxEntryBytes || !options.plan.registryHosts.includes(artifact.requestedHost) || !receipt.network.contactedHosts.includes(artifact.requestedHost)) @@ -696,7 +747,7 @@ export class PreparationExecutor { if (!receipt.network.contactedHosts.includes(resolvedHost)) fail('TOOL_FAILED', 'Direct archive response host was not contacted through the registry broker'); } if (selected.input.url && artifact.requestedUrl !== selected.input.url) fail('TOOL_FAILED', 'Acquired archive request URL does not match its lockfile URL'); - const hashes = stagedFileHashes(request.stagingRoot, artifact.stagingPath, artifact.bytes); + const hashes = stagedFileHashes(request.stagingRoot, artifact.stagingPath, artifact.bytes, options.deadline, options.signal); if (hashes.sha256 !== artifact.sha256 || (selected.input.integritySource === 'lock' && !integrityMatches(selected.input.integrity, hashes)) || (selected.input.integritySource !== 'lock' && selected.input.integritySource !== 'registry-on-acquisition')) fail('INCOMPATIBLE_INPUT', `Acquired archive failed lock/registry integrity verification: ${selected.input.name}@${selected.input.version}`); @@ -711,9 +762,10 @@ export class PreparationExecutor { // and the aggregate unique-byte ceiling have been validated. try { for (const { artifact, selected } of validated) { + checkDeadline(options.deadline, options.signal); const absoluteStaged = resolve(request.stagingRoot, ...artifact.stagingPath.split('/')); const cacheRelative = relative(this.options.cache.stagingRoot, absoluteStaged).split(sep).join('/'); - const cached = this.options.cache.promote(cacheRelative, artifact.sha256); + const cached = this.options.cache.promote(cacheRelative, artifact.sha256, { deadline: options.deadline, signal: options.signal }); archives.push({ inputIndex: artifact.inputIndex, name: selected.input.name, version: selected.input.version, installPath: artifact.installPath, sha256: cached.sha256, bytes: cached.bytes, requestedHost: artifact.requestedHost, requestedUrl: artifact.requestedUrl, resolvedUrl: artifact.resolvedUrl, @@ -726,8 +778,11 @@ export class PreparationExecutor { } } const covered = new Set(archives.map(archive => logicalInput(options.plan.inputs[archive.inputIndex]))); - for (const { input } of publicInputs) if (!covered.has(logicalInput(input))) - fail('MISSING_INPUT', `Acquisition did not produce a compatible archive for ${input.name}@${input.version}`); + for (const { input } of publicInputs) { + checkDeadline(options.deadline, options.signal); + if (!covered.has(logicalInput(input))) fail('MISSING_INPUT', `Acquisition did not produce a compatible archive for ${input.name}@${input.version}`); + } + checkDeadline(options.deadline, options.signal); archives.sort((a, b) => a.installPath.localeCompare(b.installPath)); const body: Omit = { schemaVersion: 1, stack: options.plan.stack, planHash: hash, catalogRevision: options.admission.catalogRevision, runtimeId: runtime.id, runtimeImage: runtime.image, platform: runtime.platform, @@ -748,12 +803,15 @@ export class PreparationExecutor { snapshot: string; closure: DependencyClosure; deadline: number; + signal?: AbortSignal; database?: RailsDatabaseSelection; }): Promise { - checkDeadline(options.deadline); + checkDeadline(options.deadline, options.signal); const hash = validatePlan(options.plan, options.snapshot), runtime = runtimeFromAdmission(options.plan, options.admission); + checkDeadline(options.deadline, options.signal); validateRunner(this.options.runner, options.plan.stack); - const materialized = validateClosure(options.plan, options.admission, options.closure, this.options.cache, this.options.materializationRoot), + const materialized = validateClosure(options.plan, options.admission, options.closure, this.options.cache, + options.deadline, options.signal, this.options.materializationRoot), archives = materialized.mounts; let database: OfflinePreparationRequest['database']; let synthetic: Array<{ path: string; content: string }> = []; @@ -772,7 +830,7 @@ export class PreparationExecutor { } else if (options.database) fail('INVALID_ARGUMENT', 'Database preparation is only valid for Rails'); const transformations = synthetic.map(file => ({ ...file, sha256: sha256(file.content), reason: 'Synthetic isolated Rails test configuration' })); const configurationHash = sha256(canonical(transformations)), databaseHash = sha256(canonical(database ?? null)); - const sourceHash = treeHash(options.snapshot, MAX_SOURCE_BYTES, options.deadline); + const sourceHash = treeHash(options.snapshot, MAX_SOURCE_BYTES, options.deadline, false, options.signal); const request: OfflinePreparationRequest = { schemaVersion: 1, planHash: hash, stack: options.plan.stack, runtime: { id: runtime.id, image: runtime.image, platform: runtime.platform }, sourceRoot: resolve(options.snapshot), sourceHash, @@ -786,16 +844,16 @@ export class PreparationExecutor { try { const result = await this.options.runner.prepareOffline(deepFreeze(request)); returnedPreparedRoot = typeof result?.preparedRoot === 'string' ? result.preparedRoot : undefined; - checkDeadline(options.deadline); + checkDeadline(options.deadline, options.signal); validateOfflineReceipt(result?.receipt, request); const preparedRoot = resolve(result.preparedRoot); const sourceRoot = resolve(options.snapshot); if (preparedRoot === sourceRoot || preparedRoot.startsWith(`${sourceRoot}${sep}`) || sourceRoot.startsWith(`${preparedRoot}${sep}`) || preparedRoot === this.options.cache.root || preparedRoot.startsWith(`${this.options.cache.root}${sep}`)) fail('UNSAFE_PATH', 'Prepared application must be a separate disposable copy outside cache and source roots'); - const projection = provePreparedProjection(options.snapshot, preparedRoot, options.plan.stack, request.transformations, options.deadline), + const projection = provePreparedProjection(options.snapshot, preparedRoot, options.plan.stack, request.transformations, options.deadline, options.signal), preparedManifestHash = projection.manifestHash,preparedDependencyHash=projection.dependencyHash; - if (treeHash(options.snapshot, MAX_SOURCE_BYTES, options.deadline) !== sourceHash) + if (treeHash(options.snapshot, MAX_SOURCE_BYTES, options.deadline, false, options.signal) !== sourceHash) fail('SNAPSHOT_RACE', 'Offline preparation changed its read-only input source'); completed = true; return Object.freeze({ schemaVersion: 1 as const, stack: options.plan.stack, preparedRoot, sourceHash, preparedManifestHash,preparedDependencyHash, diff --git a/lib/cso/snapshot.ts b/lib/cso/snapshot.ts index 055cb0d89..c174a9b48 100644 --- a/lib/cso/snapshot.ts +++ b/lib/cso/snapshot.ts @@ -272,7 +272,11 @@ export function assertSnapshot(runDir: string, manifest: SnapshotManifest): void const presentPaths=new Set(entries.map(e=>e.path)),presentIds=new Set(entries.map(e=>e.pathId)),deletedPaths=new Set(deleted.map(item=>item.path)),deletedIds=new Set(deleted.map(item=>item.pathId)); if(presentPaths.size!==entries.length||presentIds.size!==entries.length||deletedPaths.size!==deleted.length||deletedIds.size!==deleted.length||deleted.some(item=>presentPaths.has(item.path)||presentIds.has(item.pathId))||canonical(deleted.map(item=>item.path))!==canonical([...deletedPaths].sort())||!/^([a-f0-9]{64})$/.test(manifest.originalHash)||snapshotOriginalIdentity(entries,deleted)!==manifest.originalHash)throw new CsoError('INCOMPATIBLE_INPUT','Snapshot original identity is inconsistent'); if(manifest.changedPaths!==undefined){if(!Array.isArray(manifest.changedPaths))throw new CsoError('INCOMPATIBLE_INPUT','Snapshot changed paths are invalid');const changed=manifest.changedPaths.map(relativePath);if(new Set(changed).size!==changed.length||canonical(changed)!==canonical([...changed].sort()))throw new CsoError('INCOMPATIBLE_INPUT','Snapshot changed paths are invalid');} - const expected=entries.filter(e=>e.executionHash).sort((a,b)=>a.path.localeCompare(b.path)); + // Capture already records entries in Git's deterministic code-unit path + // order. Preserve that manifest order here: localeCompare can reorder an + // uppercase path such as README.md after lowercase source files, producing + // a different execution identity from the one written at capture time. + const expected=entries.filter(e=>e.executionHash); if(!/^([a-f0-9]{64})$/.test(manifest.executionHash)||sha256(canonical(expected.map(e=>[e.path,e.executionHash,e.mode])))!==manifest.executionHash)throw new CsoError('INCOMPATIBLE_INPUT','Snapshot execution identity is inconsistent'); const before=listed();if(canonical(before)!==canonical(expected.map(e=>e.path)))throw new CsoError('INCOMPATIBLE_INPUT','Retained snapshot membership changed'); for (const e of expected) { diff --git a/lib/cso/state.ts b/lib/cso/state.ts index a4acaa0a7..0a2d53c26 100644 --- a/lib/cso/state.ts +++ b/lib/cso/state.ts @@ -654,32 +654,40 @@ function acquireRunLease(dir:string):HeldRunLease{ function releaseRunLease(lease:HeldRunLease,path=lease.path):void{const directory=dirname(path);releaseKnownLease(path,join(directory,basename(lease.decision)),join(directory,basename(lease.active)),lease.token,lease.identity,lease.decisionIdentity,true);} export function withLock(dir: string, fn: () => T): T | Promise> { const lease=acquireRunLease(dir),unlock=()=>releaseRunLease(lease); + let value:T; try { - const value=fn(); - if(value&&typeof (value as any).then==='function')return Promise.resolve(value).finally(unlock) as Promise>; - unlock();return value as any; + value=fn(); } catch(error){ try{unlock();}catch(releaseError){throw releaseError;} throw error; } + if(value&&typeof (value as any).then==='function')return Promise.resolve(value).finally(unlock) as Promise>; + // Keep release errors outside the callback catch path. Retrying an exact + // release after it partially succeeds can only obscure which lease phase + // changed and attempts the same fail-closed cleanup twice. + unlock();return value as any; } -function boundedMarker(path:string):string{ +function boundedMarker(path:string,admit:()=>void=()=>{}):string{ + admit(); try{const stat=fs.lstatSync(path);if(stat.isSymbolicLink()||!stat.isFile()||stat.size>8192)return'';return fs.readFileSync(path,'utf8');}catch{return'';} } /** A detached watchdog owns these paths until it records exact cleanup or an acknowledgement. */ -export function hasPendingWatchdogCleanup(dir:string):boolean{ +export function hasPendingWatchdogCleanup(dir:string,admit:()=>void=()=>{}):boolean{ let visited=0,pending=false; const walk=(at:string,depth:number)=>{ if(pending||depth>6||visited++>4000)return; - let entries:fs.Dirent[];try{entries=fs.readdirSync(at,{withFileTypes:true});}catch{return;} + const entries:fs.Dirent[]=[];let directory:fs.Dir; + admit();try{directory=fs.opendirSync(at);}catch{return;} + try{for(;;){admit();const entry=directory.readSync();if(!entry)break;entries.push(entry);}}finally{directory.closeSync();} const names=new Set(entries.map(entry=>entry.name)); - if(names.has('attempt.ready')&&!names.has('attempt.stopped')&&!boundedMarker(join(at,'attempt.event')).includes('execution-copy cleanup complete')){pending=true;return;} - if(names.has('watchdog.ready')&&!names.has('watchdog.stopped')&&!boundedMarker(join(at,'watchdog.event')).includes('cleanup complete')){pending=true;return;} + if(names.has('attempt.ready')&&!names.has('attempt.stopped')&&!boundedMarker(join(at,'attempt.event'),admit).includes('execution-copy cleanup complete')){pending=true;return;} + if(names.has('watchdog.ready')&&!names.has('watchdog.stopped')&&!boundedMarker(join(at,'watchdog.event'),admit).includes('cleanup complete')){pending=true;return;} for(const entry of entries){if(!/^[A-Za-z0-9._-]{1,120}$/.test(entry.name)||!entry.isDirectory())continue;walk(join(at,entry.name),depth+1);if(pending)return;} }; for(const name of ['supervision','preparation-execution']){ - const root=join(dir,name);if(!fs.existsSync(root))continue; + admit();const root=join(dir,name);if(!fs.existsSync(root))continue; + admit(); const rootStat=fs.lstatSync(root);if(rootStat.isSymbolicLink()||!rootStat.isDirectory())throw new CsoError('UNSAFE_PATH','Watchdog supervision state is not a private directory'); walk(root,0);if(pending)return true; } @@ -693,64 +701,109 @@ export function finalizeReplayTemporary(dir:string):void{ fs.rmSync(dir,{recursive:true,force:true}); } -function repairBundleExpiry(dir:string,run:string,now:number,runExpired:boolean):boolean{ - const bundles=join(dir,'bundles');if(!fs.existsSync(bundles))return false; +/** Remove one private tree cooperatively without following links or holding directory handles across checks. */ +function boundedRemoveTree(root:string,admit:()=>void,preserveRootName?:string):void{ + type Frame={path:string;root:boolean;names?:string[];index:number}; + const stack:Frame[]=[{path:root,root:true,index:0}]; + while(stack.length){ + const frame=stack[stack.length-1]; + if(!frame.names){ + admit();let stat:fs.Stats;try{stat=fs.lstatSync(frame.path);}catch(error:any){if(error?.code==='ENOENT'){stack.pop();continue;}throw error;} + if(stat.isSymbolicLink()||!stat.isDirectory()){admit();fs.unlinkSync(frame.path);stack.pop();continue;} + const names:string[]=[];admit();const directory=fs.opendirSync(frame.path); + try{for(;;){admit();const entry=directory.readSync();if(!entry)break;if(!(frame.root&&entry.name===preserveRootName))names.push(entry.name);}}finally{directory.closeSync();} + frame.names=names;frame.index=0; + } + if(frame.indexvoid):void{ + boundedRemoveTree(root,admit,'.mutation-lock-leases'); + admit();const directory=fs.opendirSync(root);try{for(;;){admit();const entry=directory.readSync();if(!entry)break;if(entry.name!=='.mutation-lock-leases')throw new CsoError('SNAPSHOT_RACE','Private retention tree changed during bounded cleanup');}}finally{directory.closeSync();} + // Only the helper's fixed-size lease protocol remains. Consuming it with the + // directory preserves the exact-release invariant without an unbounded walk. + admit();fs.rmSync(root,{recursive:true,force:true}); +} + +function repairBundleExpiry(dir:string,run:string,now:number,runExpired:boolean,admit:()=>void):boolean{ + admit();const bundles=join(dir,'bundles');if(!fs.existsSync(bundles))return false; + admit(); const stat=fs.lstatSync(bundles);if(stat.isSymbolicLink()||!stat.isDirectory())throw new CsoError('UNSAFE_PATH','Repair bundle archive is not a private directory'); - let retained=false; - for(const name of fs.readdirSync(bundles)){ - const match=name.match(/^([a-f0-9]{32})\.json$/);if(!match){if(runExpired)fs.rmSync(join(bundles,name),{recursive:true,force:true});continue;} + let retained=false,remaining=0;admit();const directory=fs.opendirSync(bundles); + try{for(;;){ + admit();const entry=directory.readSync();if(!entry)break;const name=entry.name; + const match=name.match(/^([a-f0-9]{32})\.json$/);if(!match){if(runExpired)boundedRemoveTree(join(bundles,name),admit);else remaining++;continue;} + admit(); const value=readJson(join(bundles,name)) as Record,id=match[1],created=Date.parse(value?.createdAt),expires=Date.parse(value?.expiresAt); if(value?.schemaVersion!==3||value?.id!==id||value?.runId!==run||value?.verification?.id!==id||value?.verification?.runId!==run||value?.verification?.createdAt!==value.createdAt|| !Number.isFinite(created)||new Date(created).toISOString()!==value.createdAt||!Number.isFinite(expires)||value.expiresAt!==new Date(created+30*86400_000).toISOString()) throw new CsoError('INCOMPATIBLE_INPUT','Repair bundle retention identity is invalid'); - if(expires<=now)fs.unlinkSync(join(bundles,name));else retained=true; - } - if(!fs.readdirSync(bundles).length)fs.rmdirSync(bundles); + if(expires<=now){admit();fs.unlinkSync(join(bundles,name));}else{retained=true;remaining++;} + }}finally{directory.closeSync();} + if(!remaining){admit();fs.rmdirSync(bundles);} return retained; } -function cleanupRun(dir:string,run:string,now:number,pinned:boolean):void{ +function cleanupRun(dir:string,run:string,now:number,pinned:boolean,admit:()=>void):void{ + admit(); let lease:HeldRunLease; try{lease=acquireRunLease(dir);}catch(error){if(error instanceof CsoError&&error.code==='INSUFFICIENT_CAPACITY')return;throw error;} let releasePath=lease.path,consumed=false; try{ - if(hasPendingWatchdogCleanup(dir))return; - const created=Number(run.split('-')[0]),runExpired=now-created>30*86400_000,retainedBundle=repairBundleExpiry(dir,run,now,runExpired); - const ephemeral=fs.existsSync(join(dir,EPHEMERAL_REPLAY)); + if(hasPendingWatchdogCleanup(dir,admit))return; + const created=Number(run.split('-')[0]),runExpired=now-created>30*86400_000,retainedBundle=repairBundleExpiry(dir,run,now,runExpired,admit); + admit();const ephemeral=fs.existsSync(join(dir,EPHEMERAL_REPLAY)); if(ephemeral||(runExpired&&!pinned&&!retainedBundle)){ - const before=fs.lstatSync(dir),tomb=join(dirname(dir),`.retired-${run}-${randomBytes(16).toString('hex')}`);fs.renameSync(dir,tomb);releasePath=join(tomb,'.mutation-lock-leases',basename(lease.path));const after=fs.lstatSync(tomb); + admit();const before=fs.lstatSync(dir),tomb=join(dirname(dir),`.retired-${run}-${randomBytes(16).toString('hex')}`);fs.renameSync(dir,tomb);releasePath=join(tomb,'.mutation-lock-leases',basename(lease.path));admit();const after=fs.lstatSync(tomb); if(before.dev!==after.dev||before.ino!==after.ino)throw new CsoError('SNAPSHOT_RACE','Expired run changed while it was retired'); // The retired name is outside the public run namespace. Consume the // exclusive lease with the tree so no release/delete gap can admit a // second helper against the same directory. - fs.rmSync(tomb,{recursive:true,force:true});consumed=true;return; + consumeLeasedTree(tomb,admit);consumed=true;return; } - if(now-created>7*86400_000)for(const p of ['snapshot','readable'])fs.rmSync(join(dir,p),{recursive:true,force:true}); - if(runExpired)for(const p of ['reviews','replays','dependency-closures','scanner-outcomes','verification-attempts'])fs.rmSync(join(dir,p),{recursive:true,force:true}); + if(now-created>7*86400_000)for(const p of ['snapshot','readable'])boundedRemoveTree(join(dir,p),admit); + if(runExpired)for(const p of ['reviews','replays','dependency-closures','scanner-outcomes','verification-attempts'])boundedRemoveTree(join(dir,p),admit); }finally{if(!consumed)releaseRunLease(lease,releasePath);} } -function cleanupRetiredRun(dir:string):void{ +function cleanupRetiredRun(dir:string,admit:()=>void):void{ + admit(); let lease:HeldRunLease;try{lease=acquireRunLease(dir);}catch(error){if(error instanceof CsoError&&error.code==='INSUFFICIENT_CAPACITY')return;throw error;} - let consumed=false;try{if(hasPendingWatchdogCleanup(dir))return;fs.rmSync(dir,{recursive:true,force:true});consumed=true;}finally{if(!consumed)releaseRunLease(lease);} + let consumed=false;try{if(hasPendingWatchdogCleanup(dir,admit))return;consumeLeasedTree(dir,admit);consumed=true;}finally{if(!consumed)releaseRunLease(lease);} } -export function retention(now = Date.now()): void { +export interface RetentionOptions { deadlineMs?:number; maxEntries?:number } +export interface RetentionResult { complete:boolean; visited:number } +class RetentionBudgetExhausted extends Error {} +export function retention(now = Date.now(),options:RetentionOptions={}): RetentionResult { + const deadlineMs=options.deadlineMs??Number.MAX_SAFE_INTEGER,maxEntries=options.maxEntries??Number.MAX_SAFE_INTEGER; + if(!Number.isSafeInteger(deadlineMs)||deadlineMs<0||!Number.isSafeInteger(maxEntries)||maxEntries<1)throw new CsoError('INVALID_ARGUMENT','Invalid retention maintenance budget'); + let visited=0;const admit=()=>{if(Date.now()>=deadlineMs||visited>=maxEntries)throw new RetentionBudgetExhausted();visited++;}; + const names=(dir:string,pattern:RegExp):string[]=>{const found:string[]=[];admit();const directory=fs.opendirSync(dir);try{for(;;){admit();const entry=directory.readSync();if(!entry)break;if(pattern.test(entry.name))found.push(entry.name);}}finally{directory.closeSync();}return found;}; const root = privateRoot(); - const retainedParents=new Set(); + const retainedParents=new Set(),repositories:{repo:string;repoDir:string;runs:string[];retired:string[]}[]=[]; const retainedReport=(repoDir:string,repo:string,run:string):RunReportV3|undefined=>{ const file=join(repoDir,run,'report.json'); try{ + admit(); const stat=fs.lstatSync(file); if(!stat.isFile()||stat.isSymbolicLink()||stat.nlink!==1||stat.size<=0||stat.size>MAX_STATE_FILE|| (process.getuid&&stat.uid!==process.getuid())||(process.platform!=='win32'&&(stat.mode&0o077)!==0))return; + admit(); const report=JSON.parse(fs.readFileSync(file,'utf8')); if(report?.schemaVersion!==3||report.runId!==run||report.repoId!==repo||!Array.isArray(report.coverage)||!Array.isArray(report.findings)|| !['running','finished','interrupted'].includes(report.status))return; return report as RunReportV3; - }catch{return;} + }catch(error){if(error instanceof RetentionBudgetExhausted)throw error;return;} }; - for(const repo of fs.readdirSync(root).filter(x=>/^[a-f0-9]{24}$/.test(x))){ - const repoDir=secureDirectory(join(root,repo)); - for(const run of fs.readdirSync(repoDir).filter(x=>/^\d{13}-[a-f0-9]{16}$/.test(x))){ + try{ + for(const repo of names(root,/^[a-f0-9]{24}$/)){ + admit();const repoDir=secureDirectory(join(root,repo)),entries=names(repoDir,/^(?:\d{13}-[a-f0-9]{16}|\.retired-\d{13}-[a-f0-9]{16}-[a-f0-9]{32})$/),runs=entries.filter(x=>/^\d{13}-/.test(x)),retired=entries.filter(x=>x.startsWith('.retired-')); + repositories.push({repo,repoDir,runs,retired}); + } + // Discover every live recheck pin before destructive cleanup. An exhausted + // discovery pass returns without deleting a parent that may still be in use. + for(const {repo,repoDir,runs} of repositories)for(const run of runs){ if(now-Number(run.split('-')[0])>30*86400_000)continue; const report=retainedReport(repoDir,repo,run),parent=report?.parent as Record|undefined; if(!report||!['running','interrupted'].includes(report.status)||!Number.isFinite(Date.parse(report.deadline))||Date.parse(report.deadline)<=now|| @@ -759,23 +812,27 @@ export function retention(now = Date.now()): void { const original=retainedReport(repoDir,repo,parent.runId); if(original?.status==='finished'&&original.findings.some(f=>f.id===parent.findingId))retainedParents.add(`${repo}/${parent.runId}`); } - } - for (const repo of fs.readdirSync(root).filter(x => /^[a-f0-9]{24}$/.test(x))) { - const repoDir = secureDirectory(join(root,repo)); - for(const retired of fs.readdirSync(repoDir).filter(x=>/^\.retired-\d{13}-[a-f0-9]{16}-[a-f0-9]{32}$/.test(x)))cleanupRetiredRun(join(repoDir,retired)); - for (const run of fs.readdirSync(repoDir).filter(x => /^\d{13}-[a-f0-9]{16}$/.test(x))) { - const dir = secureDirectory(join(repoDir,run)); - // Every destructive retention decision owns the same exclusive lease as - // writers and replay. Whole runs are atomically retired before release. - cleanupRun(dir,run,now,retainedParents.has(`${repo}/${run}`)); + for (const {repo,repoDir,runs,retired} of repositories) { + for(const name of retired)cleanupRetiredRun(join(repoDir,name),admit); + for (const run of runs) { + admit();const dir = secureDirectory(join(repoDir,run)); + // Every destructive retention decision owns the same exclusive lease as + // writers and replay. Whole runs are atomically retired before release. + cleanupRun(dir,run,now,retainedParents.has(`${repo}/${run}`),admit); + } } - } - const legacy=join(root,'legacy-imports'); - if(fs.existsSync(legacy)){ - const directory=fs.lstatSync(legacy);if(directory.isSymbolicLink()||!directory.isDirectory())throw new CsoError('UNSAFE_PATH','Legacy report archive is not a private directory'); - for(const name of fs.readdirSync(legacy).filter(value=>/^[a-f0-9]{64}\.json$/.test(value))){ - const file=join(legacy,name),stat=fs.lstatSync(file);if(stat.isSymbolicLink()||!stat.isFile()||(process.getuid&&stat.uid!==process.getuid()))throw new CsoError('UNSAFE_PATH','Legacy report archive contains an unsafe artifact'); - if(now-stat.mtimeMs>30*86400_000)fs.unlinkSync(file); + admit();const legacy=join(root,'legacy-imports'); + if(fs.existsSync(legacy)){ + admit();const directory=fs.lstatSync(legacy);if(directory.isSymbolicLink()||!directory.isDirectory())throw new CsoError('UNSAFE_PATH','Legacy report archive is not a private directory'); + for(const name of names(legacy,/^[a-f0-9]{64}\.json$/)){ + admit();const file=join(legacy,name),stat=fs.lstatSync(file);if(stat.isSymbolicLink()||!stat.isFile()||(process.getuid&&stat.uid!==process.getuid()))throw new CsoError('UNSAFE_PATH','Legacy report archive contains an unsafe artifact'); + admit(); + if(now-stat.mtimeMs>30*86400_000)fs.unlinkSync(file); + } } + return{complete:true,visited}; + }catch(error){ + if(error instanceof RetentionBudgetExhausted)return{complete:false,visited}; + throw error; } } diff --git a/lib/cso/verification.ts b/lib/cso/verification.ts index b1e9f38db..3659a1f7e 100644 --- a/lib/cso/verification.ts +++ b/lib/cso/verification.ts @@ -317,7 +317,7 @@ export async function verifyRepair(params:{runId:string;runDir:string;manifest:S for(const boundary of [...new Set([...request.boundaryFiles,...request.testFiles,...beforeStartPlan.entrypointFiles])]){const e=params.manifest.entries.find(x=>x.path===boundary);if(!e||e.transformation)throw new CsoError('INCOMPATIBLE_INPUT',`Snapshot transformation changes or withholds a verification input: ${boundary}`);} for(const change of request.changes){const path=relativePath(change.path),entry=params.manifest.entries.find(x=>x.path===path);containedFile(snapshot,path);if(change.beforeSha256===null){if(entry)throw new CsoError('INCOMPATIBLE_INPUT',`Declared new repair path already exists in the snapshot: ${path}`);}else if(!entry||entry.transformation)throw new CsoError('INCOMPATIBLE_INPUT',`Snapshot transformation changes or withholds a repair input: ${path}`);} let guardedCleanup:(()=>Promise)|undefined;if(params.watchdogPath){const deadline=Math.min(params.attemptDeadline??Date.now()+300_000,Date.now()+300_000);guardedCleanup=await attemptGuard(params.runDir,work,params.watchdogPath,deadline);}secureDirectory(observations); - let certified:ReturnType|undefined,beforeObs:VerificationObservation|undefined,afterObs:VerificationObservation|undefined,beforeWitness:AssertionWitnessReceipt|undefined,afterWitness:AssertionWitnessReceipt|undefined,failure:unknown; + let certified:ReturnType|undefined,beforeObs:VerificationObservation|undefined,afterObs:VerificationObservation|undefined,beforeWitness:AssertionWitnessReceipt|undefined,afterWitness:AssertionWitnessReceipt|undefined,failure:unknown,missingExternalWitness=false; try{ preparePatchedSource(snapshot,after,request); const afterTestPlan=canonicalTestPlan(after,stack),afterStartPlan=canonicalStartPlan(after,stack,request.port);if(canonical(afterTestPlan)!==canonical(beforeTestPlan))throw new CsoError('ASSERTION_FAILED','Repair changed the canonical project test suite, runner configuration, or discovered test inputs'); @@ -335,13 +335,13 @@ export async function verifyRepair(params:{runId:string;runDir:string;manifest:S afterObs=observedAfter; certified=certify({...params,request,identityRequest,before:beforeObs,after:afterObs,beforeRoot:snapshot,afterRoot:after,startPlanHash:beforeStartPlan.signature,testPlanHash:beforeTestPlan.signature,testToolchain:beforeTestPlan.toolchain,minimumPassingTests:beforeTestPlan.minimumPassingTests,...(beforeWitness&&afterWitness?{witness:{before:beforeWitness,after:afterWitness}}:{})}); }catch(error){failure=error;} - try{if(guardedCleanup)await guardedCleanup();else fs.rmSync(work,{recursive:true,force:true});}catch(error){failure=error;certified=undefined;} - if(!failure&&certified&&!certified.bundle){const manifest=certified.manifest,review=manifest.review,missingWitness=!manifest.assertionAssurance&&manifest.testCompletionAssurance==='self_reported'&&manifest.result==='inconclusive'&&manifest.before.booted&&manifest.before.legitimate&&manifest.before.security==='intended_failure'&&manifest.before.existingTests&&manifest.after.booted&&manifest.after.legitimate&&manifest.after.security==='pass'&&manifest.after.existingTests&&review.independent&&review.rootCauseRepaired&&review.featurePreserved&&!review.boundaryMocks;failure=missingWitness?new CsoError('PREREQUISITE','Repair verification retained self-reported project-test diagnostics but requires a helper-authenticated out-of-process external assertion witness before a runtime-tested bundle can be issued'):new CsoError('ASSERTION_FAILED',manifest.result==='inconclusive'?'Repair verification was inconclusive; no repair bundle was issued':'Repair failed one or more required boot, control, security, existing-test, or review assertions; no repair bundle was issued');} + try{if(guardedCleanup)await guardedCleanup();else fs.rmSync(work,{recursive:true,force:true});}catch(error){failure=error;certified=undefined;missingExternalWitness=false;} + if(!failure&&certified&&!certified.bundle){const manifest=certified.manifest,review=manifest.review;missingExternalWitness=!manifest.assertionAssurance&&manifest.testCompletionAssurance==='self_reported'&&manifest.result==='inconclusive'&&manifest.before.booted&&manifest.before.legitimate&&manifest.before.security==='intended_failure'&&manifest.before.existingTests&&manifest.after.booted&&manifest.after.legitimate&&manifest.after.security==='pass'&&manifest.after.existingTests&&review.independent&&review.rootCauseRepaired&&review.featurePreserved&&!review.boundaryMocks;failure=missingExternalWitness?new CsoError('PREREQUISITE','Repair verification retained self-reported project-test diagnostics but requires a helper-authenticated out-of-process external assertion witness before a runtime-tested bundle can be issued'):new CsoError('ASSERTION_FAILED',manifest.result==='inconclusive'?'Repair verification was inconclusive; no repair bundle was issued':'Repair failed one or more required boot, control, security, existing-test, or review assertions; no repair bundle was issued');} if(failure){ if(beforeObs){ const cause=failure instanceof CsoError?failure:new CsoError('ASSERTION_FAILED','Repair validation failed after the before-phase observation'); const reproduction=!beforeObs.booted?'blocked':!beforeObs.legitimate?'inconclusive':beforeObs.security==='intended_failure'?'reproduced':beforeObs.security==='pass'?'disproved':'inconclusive',harnessHash=verificationHarnessHash(request,snapshot); - const missingWitness=cause.code==='PREREQUISITE'&&cause.message.includes('out-of-process external assertion witness')&&reproduction==='reproduced'&&afterObs?.booted===true&&afterObs.legitimate===true&&afterObs.security==='pass'&&beforeObs.existingTests&&afterObs.existingTests; + const missingWitness=missingExternalWitness&&cause.code==='PREREQUISITE'&&reproduction==='reproduced'&&afterObs?.booted===true&&afterObs.legitimate===true&&afterObs.security==='pass'&&beforeObs.existingTests&&afterObs.existingTests; const raw={schemaVersion:3 as const,artifactKind:'repair_candidate' as const,runId:params.runId,findingId:identityRequest.findingId,createdAt:new Date().toISOString(),bundleIssued:false as const,runtime:{image:params.runtime.image,platform:params.runtime.platform,profile:params.runtime.id},policyHash:params.policyHash,requestHash:sha256(canonical(identityRequest)),harnessHash,sourceHash:params.manifest.executionHash,request:identityRequest,patchHash:patchHash(identityRequest),testToolchain:beforeTestPlan.toolchain,testCompletionAssurance:'self_reported' as const,...(params.preparation?{preparationHash:sha256(canonical(params.preparation))}:{}),before:{...beforeObs,inputHash:harnessHash},...(afterObs?{after:{...afterObs,inputHash:harnessHash}}:{}),reproduction,repair:missingWitness?'proposed' as const:'failed' as const,failure:{code:cause.code,message:cause.message}},safe=sanitizeHelperForJson(raw) as Omit,id=sha256(canonical(safe)).slice(0,32),attempt:FailedVerificationAttempt={...safe,id}; validateVerificationObservation(attempt.before);if(attempt.after)validateVerificationObservation(attempt.after);if(params.persist!==false)writeJsonExclusive(join(params.runDir,'verification-attempts',`${id}.json`),attempt); throw new VerificationAttemptError(cause,attempt); diff --git a/scripts/cso-eval.ts b/scripts/cso-eval.ts index 400a799e2..268e95b84 100644 --- a/scripts/cso-eval.ts +++ b/scripts/cso-eval.ts @@ -55,6 +55,17 @@ export interface EvalFinding { id: string; evidence: 'supported' | 'hypothesis' | 'legacy_review'; claimedTested: boolean; /** Written only by the independent evaluator after reviewing the reported trace. */ judgment: 'correct' | 'incorrect' | 'unadjudicated'; matchedCaseId?: string; + /** + * Written only by the independent evaluator. A producer's tested claim is + * valid only when this finding, rather than merely another finding in the + * same corpus cell, owns the trusted repair and recheck observations. + */ + trustedVerification?: { + repair: Outcome; + repairEvidenceHash?: string; + recheck: Outcome; + recheckEvidenceHash?: string; + }; } export interface EvalResult { cellId: string; sourceHash: string; skillHash: string; model: string; host: string; budgetSeconds: number; @@ -445,11 +456,33 @@ function validateResult(result: EvalResult, cell: EvalCell, corpus: CorpusManife if (result.recheckEvidenceHash !== undefined && !HEX.test(result.recheckEvidenceHash)) throw new Error('INVALID_RECHECK_EVIDENCE'); if (result.usage && (result.usage.source !== 'host' || (result.usage.tokens !== undefined && !validNumber(result.usage.tokens)) || (result.usage.costUSD !== undefined && !validNumber(result.usage.costUSD)))) throw new Error('INVALID_EVAL_USAGE'); if (result.producerReceiptHash !== undefined && !HEX.test(result.producerReceiptHash)) throw new Error('INVALID_PRODUCER_RECEIPT_PROVENANCE'); - const ids = new Set(); + const ids = new Set(), repairEvidence = new Set(), recheckEvidence = new Set(); for (const finding of result.findings) { if (!finding.id || ids.has(finding.id) || !['supported', 'hypothesis', 'legacy_review'].includes(finding.evidence) || !['correct', 'incorrect', 'unadjudicated'].includes(finding.judgment) || typeof finding.claimedTested !== 'boolean') throw new Error('INVALID_FINDING_JUDGMENT'); ids.add(finding.id); if (finding.judgment === 'correct' && (cell.variant !== 'vulnerable' || finding.matchedCaseId !== cell.caseId)) throw new Error('INVALID_ORACLE_MATCH'); + if (finding.trustedVerification !== undefined) { + const verification = finding.trustedVerification as Record, allowed = new Set(['repair', 'repairEvidenceHash', 'recheck', 'recheckEvidenceHash']); + if (!verification || typeof verification !== 'object' || Array.isArray(verification) || Object.keys(verification).some(key => !allowed.has(key)) || + !['passed', 'failed', 'blocked', 'not_attempted'].includes(String(verification.repair)) || + !['passed', 'failed', 'blocked', 'not_attempted'].includes(String(verification.recheck)) || !finding.claimedTested || cell.version !== 'v3' || cell.mode !== 'comprehensive') + throw new Error('INVALID_TRUSTED_FINDING_VERIFICATION'); + const repairHash = verification.repairEvidenceHash, recheckHash = verification.recheckEvidenceHash; + if (!(repairHash === undefined || (typeof repairHash === 'string' && HEX.test(repairHash))) || + !(recheckHash === undefined || (typeof recheckHash === 'string' && HEX.test(recheckHash))) || + (verification.repair === 'passed') !== (repairHash !== undefined) || + (verification.recheck === 'passed') !== (recheckHash !== undefined) || + (verification.recheck === 'passed' && verification.repair !== 'passed')) + throw new Error('INVALID_TRUSTED_FINDING_VERIFICATION'); + if (typeof repairHash === 'string') { + if (repairEvidence.has(repairHash)) throw new Error('DUPLICATE_TRUSTED_REPAIR_BINDING'); + repairEvidence.add(repairHash); + } + if (typeof recheckHash === 'string') { + if (recheckEvidence.has(recheckHash)) throw new Error('DUPLICATE_TRUSTED_RECHECK_BINDING'); + recheckEvidence.add(recheckHash); + } + } } if (cell.mode === 'daily' && [result.setup, result.reproduction, result.repair, result.recheck].some(value => value !== 'not_attempted')) throw new Error('DAILY_EVAL_EXECUTED_APPLICATION'); } @@ -465,14 +498,29 @@ function measuredRate(cells: EvalCell[], results: Map, field if (!result.oracleEvidenceHash || result.oracleVersion !== corpus.version || result.setup !== 'passed') return false; if (field === 'reproduction') return true; if (result.reproduction !== 'passed' || result.repair !== 'passed' || !result.heldOutAssertionsPassed) return false; + if (cell.version === 'v3' && !result.findings.some(finding => trustedFindingVerification(cell, result, finding, 'repair', corpus))) return false; if (field === 'repair') return true; // Correct alternative patches need not match the reference fix byte-for-byte. // Closure needs a fresh source and a separate trusted current-source observation. - return result.freshRecheck && !!result.currentSourceHash && result.currentSourceHash !== cell.sourceHash - && !!result.recheckEvidenceHash && result.recheckEvidenceHash !== result.oracleEvidenceHash; + return result.freshRecheck && !!result.currentSourceHash && result.currentSourceHash !== cell.sourceHash && + !!result.recheckEvidenceHash && result.recheckEvidenceHash !== result.oracleEvidenceHash && + (cell.version !== 'v3' || result.findings.some(finding => trustedFindingVerification(cell, result, finding, 'recheck', corpus))); }).length, cells.length); } +function trustedFindingVerification(cell: EvalCell, result: EvalResult, finding: EvalFinding, stage: 'repair' | 'recheck', corpus: CorpusManifest): boolean { + const verification = finding.trustedVerification; + if (cell.version !== 'v3' || cell.mode !== 'comprehensive' || cell.variant !== 'vulnerable' || finding.evidence !== 'supported' || + finding.judgment !== 'correct' || finding.matchedCaseId !== cell.caseId || !finding.claimedTested || !verification || + result.setup !== 'passed' || result.reproduction !== 'passed' || result.repair !== 'passed' || !result.heldOutAssertionsPassed || + verification.repair !== 'passed' || verification.repairEvidenceHash !== result.oracleEvidenceHash || result.oracleVersion !== corpus.version) + return false; + if (stage === 'repair') return true; + return result.recheck === 'passed' && verification.recheck === 'passed' && verification.recheckEvidenceHash === result.recheckEvidenceHash && + result.freshRecheck && !!result.currentSourceHash && result.currentSourceHash !== cell.sourceHash && + !!result.recheckEvidenceHash && result.recheckEvidenceHash !== result.oracleEvidenceHash; +} + export function scoreEval(matrix: EvalMatrix, observations: EvalResult[], qualification: ReleaseQualification = { containment: {} }, corpus: CorpusManifest = loadCorpusManifest()) { validateMatrix(matrix, corpus); if (!Array.isArray(observations)) throw new Error('INVALID_EVAL_RESULTS'); @@ -503,8 +551,7 @@ export function scoreEval(matrix: EvalMatrix, observations: EvalResult[], qualif const tokens = completed.flatMap(result => result.usage?.tokens === undefined ? [] : [result.usage.tokens]); const falseTested = selected.reduce((count, cell) => { const result = results.get(cell.id); if (!result) return count; - const valid = cell.version === 'v3' && cell.variant === 'vulnerable' && measuredRate([cell], results, 'repair', corpus).numerator === 1; - return count + result.findings.filter(finding => finding.claimedTested && !(valid && finding.evidence === 'supported' && finding.judgment === 'correct')).length; + return count + result.findings.filter(finding => finding.claimedTested && !trustedFindingVerification(cell, result, finding, 'repair', corpus)).length; }, 0); groups.push({ version, mode, cells: selected.length, submitted: completed.length, missing: selected.length - completed.length, reports: rate(completed.filter(result => result.reportPresent).length, selected.length), precision: rate(correct, reported), recall: rate(expected.filter(found).length, expected.length), highCriticalRecall: rate(highCritical.filter(found).length, highCritical.length), @@ -526,7 +573,10 @@ export function scoreEval(matrix: EvalMatrix, observations: EvalResult[], qualif const eligible = matrix.cells.filter(cell => cell.version === 'v3' && cell.mode === 'comprehensive' && cell.variant === 'vulnerable' && cell.stack === stack); // A repaired oracle case demonstrates the full find-to-repair workflow only // when the producer also reported the matching supported finding. - const successful = new Set(eligible.filter(cell => found(cell) && measuredRate([cell], results, 'repair', corpus).numerator === 1).map(cell => cell.caseId)); + const successful = new Set(eligible.filter(cell => { + const result = results.get(cell.id); + return result?.findings.some(finding => trustedFindingVerification(cell, result, finding, 'repair', corpus)); + }).map(cell => cell.caseId)); return [stack, { correctHeldOutRepairs: successful.size, denominator: new Set(eligible.map(cell => cell.caseId)).size }]; })); const assessedAll = observations.length === matrix.cells.length && groups.every(group => !group.unadjudicated); diff --git a/test/cso-cache.test.ts b/test/cso-cache.test.ts index 4d05422b4..b3449d606 100644 --- a/test/cso-cache.test.ts +++ b/test/cso-cache.test.ts @@ -106,6 +106,56 @@ describe('CSO immutable public archive cache', () => { expect(copies.map(item => createHash('sha256').update(fs.readFileSync(item.path)).digest('hex')).sort()).toEqual([first.digest, second.digest].sort()); }); + test('bounds every public operation by an optional absolute deadline', () => { + const { root, staging, cache } = fixture(1024); + const archive = stage(staging, 'deadline', 'bounded'); + const expired = Date.now() - 1; + expect(code(() => cache.promote(archive.path, archive.digest, expired))).toBe('DEADLINE'); + const entry = cache.promote(archive.path, archive.digest); + expect(code(() => cache.get(entry.sha256, { deadline: expired }))).toBe('DEADLINE'); + expect(code(() => cache.stats({ deadline: expired }))).toBe('DEADLINE'); + const destination = path.join(root, 'deadline-materialization'); fs.mkdirSync(destination, { mode: 0o700 }); + expect(code(() => cache.materialize([entry.sha256], destination, { deadline: expired }))).toBe('DEADLINE'); + expect(fs.readdirSync(destination)).toEqual([]); + }); + + test('cancels between archive-copy chunks and removes the partial incoming object', () => { + const { staging, cacheRoot, cache } = fixture(512 * 1024); + const archive = stage(staging, 'cancel-copy', Buffer.alloc(192 * 1024, 0x61)); + const controller = new AbortController(), original = fs.writeSync.bind(fs); let archiveWrites = 0; + const writer = spyOn(fs, 'writeSync').mockImplementation(((fd: number, buffer: string | NodeJS.ArrayBufferView, + offsetOrPosition?: number | null, lengthOrEncoding?: number | BufferEncoding, position?: number | null) => { + const written = (original as any)(fd, buffer, offsetOrPosition, lengthOrEncoding, position); + if (Buffer.isBuffer(buffer) && lengthOrEncoding === 64 * 1024 && written > 0) { archiveWrites++; controller.abort(); } + return written; + }) as typeof fs.writeSync); + try { + expect(code(() => cache.promote(archive.path, archive.digest, { deadline: Date.now() + 60_000, signal: controller.signal }))).toBe('CANCELLED'); + } finally { writer.mockRestore(); } + expect(archiveWrites).toBe(1); + expect(fs.readdirSync(path.join(cacheRoot, 'incoming'))).toEqual([]); + expect(fs.readdirSync(path.join(cacheRoot, 'entries'))).toEqual([]); + expect(fs.readdirSync(path.join(cacheRoot, 'metadata'))).toEqual([]); + }); + + test('cancels between cache-hit read chunks without damaging the immutable entry', () => { + const { staging, cacheRoot, cache } = fixture(512 * 1024); + const archive = stage(staging, 'cancel-read', Buffer.alloc(192 * 1024, 0x62)), entry = cache.promote(archive.path, archive.digest); + const controller = new AbortController(), original = fs.readSync.bind(fs); let archiveReads = 0; + const reader = spyOn(fs, 'readSync').mockImplementation(((fd: number, buffer: NodeJS.ArrayBufferView, + offset: number, length: number, position: number | null) => { + const read = original(fd, buffer, offset, length, position); + if (length === 64 * 1024 && read > 0) { archiveReads++; controller.abort(); } + return read; + }) as typeof fs.readSync); + try { + expect(code(() => cache.get(entry.sha256, { signal: controller.signal }))).toBe('CANCELLED'); + } finally { reader.mockRestore(); } + expect(archiveReads).toBe(1); + expect(fs.existsSync(entry.path)).toBe(true); + expect(fs.readdirSync(path.join(cacheRoot, 'metadata'))).toEqual([`${entry.sha256}.json`]); + }); + test('never overwrites an immutable entry and recovers interrupted entry publication', () => { const { staging, cacheRoot, cache } = fixture(); const archive = stage(staging, 'one', 'same bytes'); diff --git a/test/cso-eval.test.ts b/test/cso-eval.test.ts index 43f714eb6..3102e0d14 100644 --- a/test/cso-eval.test.ts +++ b/test/cso-eval.test.ts @@ -43,11 +43,13 @@ function portableSkill(version: 'v2' | 'v3', sectionMarker: string): string { function syntheticResult(cell: EvalCell): EvalResult { const runtime = cell.mode === 'comprehensive'; const positive = cell.variant === 'vulnerable'; + const repairEvidenceHash = 'c'.repeat(64), recheckEvidenceHash = 'd'.repeat(64); return { cellId: cell.id, sourceHash: cell.sourceHash, skillHash: cell.skillHash, model: cell.model, host: cell.host, budgetSeconds: cell.budgetSeconds, reportPresent: true, reportComplete: true, - findings: positive ? [{ id: 'finding-1', evidence: cell.version === 'v2' ? 'legacy_review' : 'supported', claimedTested: runtime && cell.version === 'v3', judgment: 'correct', matchedCaseId: cell.caseId }] : [], + findings: positive ? [{ id: 'finding-1', evidence: cell.version === 'v2' ? 'legacy_review' : 'supported', claimedTested: runtime && cell.version === 'v3', judgment: 'correct', matchedCaseId: cell.caseId, + ...(runtime && cell.version === 'v3' ? { trustedVerification: { repair: 'passed' as const, repairEvidenceHash, recheck: 'passed' as const, recheckEvidenceHash } } : {}) }] : [], setup: runtime ? 'passed' : 'not_attempted', reproduction: runtime && positive ? 'passed' : 'not_attempted', repair: runtime && positive ? 'passed' : 'not_attempted', recheck: runtime && positive ? 'passed' : 'not_attempted', - ...(runtime && positive ? { oracleEvidenceHash: 'c'.repeat(64), oracleVersion: CORPUS_VERSION, currentSourceHash: corpus.cases.find(fixture => fixture.id === cell.caseId)!.filesHash.fixed, recheckEvidenceHash: 'd'.repeat(64) } : {}), + ...(runtime && positive ? { oracleEvidenceHash: repairEvidenceHash, oracleVersion: CORPUS_VERSION, currentSourceHash: corpus.cases.find(fixture => fixture.id === cell.caseId)!.filesHash.fixed, recheckEvidenceHash } : {}), heldOutAssertionsPassed: runtime && positive, freshRecheck: runtime && positive, latencyMs: 1500, firstUsefulResultMs: positive ? 500 : null, }; } @@ -247,6 +249,14 @@ describe('CSO matched evaluation accounting', () => { const result = syntheticResult(cell); delete result.oracleEvidenceHash; const score = scoreEval(matrix, [result]); expect(group(score, 'v3', 'comprehensive').falseTested).toBe(1); expect(group(score, 'v3', 'comprehensive').repair.numerator).toBe(0); }); + test('one trusted cell repair cannot certify another claimed-tested finding', () => { + const results = matrix.cells.map(syntheticResult), cell = matrix.cells.find(item => item.version === 'v3' && item.mode === 'comprehensive' && item.variant === 'vulnerable')!, result = results.find(item => item.cellId === cell.id)!; + result.findings.push({ id: 'unbound-tested-claim', evidence: 'supported', claimedTested: true, judgment: 'correct', matchedCaseId: cell.caseId }); + const score = scoreEval(matrix, results, qualification); + expect(group(score, 'v3', 'comprehensive').falseTested).toBe(1); expect(score.gates.zeroFalselyTestedRepairs).toBe('fail'); expect(score.status).toBe('partial'); + result.findings[1].trustedVerification = structuredClone(result.findings[0].trustedVerification); + expect(() => scoreEval(matrix, results, qualification)).toThrow('DUPLICATE_TRUSTED_REPAIR_BINDING'); + }); test('per-stack held-out repair requires the matching supported discovery', () => { const results = matrix.cells.map(syntheticResult); for (let index = 0; index < matrix.cells.length; index++) { @@ -280,6 +290,9 @@ describe('CSO matched evaluation accounting', () => { expect(() => scoreEval(matrix, [result, result])).toThrow('UNKNOWN_OR_DUPLICATE'); expect(() => scoreEval(matrix, [{ ...result, model: 'other-model' }])).toThrow('UNMATCHED_EVAL_RESULT'); expect(() => scoreEval(matrix, [{ ...result, setup: 'passed' }])).toThrow('DAILY_EVAL_EXECUTED'); + const comprehensive = syntheticResult(matrix.cells.find(cell => cell.version === 'v3' && cell.mode === 'comprehensive' && cell.variant === 'vulnerable')!); + comprehensive.findings[0].trustedVerification = { repair: 'failed', repairEvidenceHash: 'not-a-hash', recheck: 'not_attempted' }; + expect(() => scoreEval(matrix, [comprehensive])).toThrow('INVALID_TRUSTED_FINDING_VERIFICATION'); }); test('missing containment tests remain unmeasured and a failed canary fails qualification', () => { const results = matrix.cells.map(syntheticResult); diff --git a/test/cso-preparation-executor.test.ts b/test/cso-preparation-executor.test.ts index 0913d036a..5d6cdf211 100644 --- a/test/cso-preparation-executor.test.ts +++ b/test/cso-preparation-executor.test.ts @@ -1,4 +1,4 @@ -import { afterEach, describe, expect, test } from 'bun:test'; +import { afterEach, describe, expect, spyOn, test } from 'bun:test'; import * as fs from 'node:fs'; import * as os from 'node:os'; import * as path from 'node:path'; @@ -218,6 +218,43 @@ describe('CSO constrained dependency preparation executor', () => { throw new Error('expected poisoned cache reuse to fail'); }); + test('threads cancellation into retained-closure cache verification', async () => { + const source = snapshot('node'), plan = inspectPreparation(source, 'node'), admission = admitPreparationRuntime({ plan, platform: 'linux/amd64', catalog: catalog('node') }); + const runner = new FakeRunner(), cache = cacheFixture(), executor = new PreparationExecutor({ cache, runner }), deadline = Date.now() + 60_000; + const closure = await executor.acquire({ plan, admission, snapshot: source, deadline }); + const controller = new AbortController(), original = fs.readSync.bind(fs); let cacheReads = 0; + const reader = spyOn(fs, 'readSync').mockImplementation(((fd: number, buffer: NodeJS.ArrayBufferView, + offset: number, length: number, position: number | null) => { + const read = original(fd, buffer, offset, length, position); + if (length === 64 * 1024 && read > 0) { cacheReads++; controller.abort(); } + return read; + }) as typeof fs.readSync); + try { + await executor.acquire({ plan, admission, snapshot: source, deadline, signal: controller.signal, offline: true, existingClosure: closure }); + throw new Error('expected retained closure verification to be cancelled'); + } catch (error) { expect(errorCode(error)).toBe('CANCELLED'); } + finally { reader.mockRestore(); } + expect(cacheReads).toBe(1); expect(runner.acquireCalls).toBe(1); + }); + + test('cancels acquisition while independently hashing staged archive bytes', async () => { + const source = snapshot('node'), plan = inspectPreparation(source, 'node'), admission = admitPreparationRuntime({ plan, platform: 'linux/amd64', catalog: catalog('node') }); + const runner = new FakeRunner(), cache = cacheFixture(), executor = new PreparationExecutor({ cache, runner }); + const controller = new AbortController(), original = fs.readSync.bind(fs); let stagedReads = 0; + const reader = spyOn(fs, 'readSync').mockImplementation(((fd: number, buffer: NodeJS.ArrayBufferView, + offset: number, length: number, position: number | null) => { + const read = original(fd, buffer, offset, length, position); + if (length === 64 * 1024 && read > 0) { stagedReads++; controller.abort(); } + return read; + }) as typeof fs.readSync); + try { + await executor.acquire({ plan, admission, snapshot: source, deadline: Date.now() + 60_000, signal: controller.signal }); + throw new Error('expected staged archive verification to be cancelled'); + } catch (error) { expect(errorCode(error)).toBe('CANCELLED'); } + finally { reader.mockRestore(); } + expect(stagedReads).toBe(1); expect(runner.acquireCalls).toBe(1); expect(cache.stats().entries).toBe(0); + }); + test('run-owned archive copies survive a cache eviction between validation and runner start', async () => { const source = snapshot('node'), plan = inspectPreparation(source, 'node'), admission = admitPreparationRuntime({ plan, platform: 'linux/amd64', catalog: catalog('node') }); const runner = new FakeRunner(), base = root('cso-eviction-race-'), staging = path.join(base, 'staging'); fs.mkdirSync(staging, { mode: 0o700 }); diff --git a/test/cso-snapshot-identity.test.ts b/test/cso-snapshot-identity.test.ts index 868e9c980..685006aa6 100644 --- a/test/cso-snapshot-identity.test.ts +++ b/test/cso-snapshot-identity.test.ts @@ -24,6 +24,7 @@ describe('CSO retained snapshot identity',()=>{ for(const entry of [dependency,venv]){expect(entry).toMatchObject({originalHash:'not-read',transformation:'excluded: host dependencies, metadata, state, or agent configuration'});expect(entry?.executionHash).toBeUndefined();} expect(agent).toMatchObject({originalHash:sha256('untrusted agent instructions\n'),transformation:'excluded: host dependencies, metadata, state, or agent configuration'});expect(agent?.executionHash).toBeUndefined();expect(fs.readFileSync(path.join(runDir,'readable','packages/web/.agents/skills/x/SKILL.md'),'utf8')).toBe('untrusted agent instructions\n'); }); + test('validates the captured execution identity when uppercase and lowercase paths coexist',async()=>{const {repo,runDir,git}=fixture();fs.writeFileSync(path.join(repo,'README.md'),'fixture documentation\n');git('add','README.md');git('commit','-qm','add uppercase path');const manifest=await capture(repo,runDir);expect(manifest.entries.filter(item=>item.executionHash).map(item=>item.path)).toEqual(['README.md','app.js']);expect(()=>assertSnapshot(runDir,manifest)).not.toThrow();}); test('binds original, sanitized, and opaque path identities',async()=>{const {repo,runDir}=fixture(),manifest=await capture(repo,runDir);assertSnapshot(runDir,manifest);const changedEntry=structuredClone(manifest);changedEntry.entries[0].originalHash='a'.repeat(64);expect(()=>assertSnapshot(runDir,changedEntry)).toThrow('original identity');const changedRoot=structuredClone(manifest);changedRoot.originalHash='b'.repeat(64);expect(()=>assertSnapshot(runDir,changedRoot)).toThrow('original identity');const changedPathId=structuredClone(manifest);changedPathId.entries[0].pathId='c'.repeat(32);expect(()=>assertSnapshot(runDir,changedPathId)).toThrow('invalid source entry');}); test.skipIf(process.platform==='win32')('preserves retained source modes under the required restrictive umask',async()=>{const {repo,runDir,git}=fixture(),script=path.join(repo,'tool.sh');fs.writeFileSync(script,'#!/bin/sh\nexit 0\n');fs.chmodSync(path.join(repo,'app.js'),0o644);fs.chmodSync(script,0o755);git('add','app.js','tool.sh');git('commit','-qm','add executable input');const previous=process.umask(0o077);let manifest;try{manifest=await capture(repo,runDir);}finally{process.umask(previous);}expect(manifest.entries.find(item=>item.path==='app.js')?.mode).toBe(0o644);expect(manifest.entries.find(item=>item.path==='tool.sh')?.mode).toBe(0o755);expect(fs.statSync(path.join(runDir,'snapshot','app.js')).mode&0o777).toBe(0o644);expect(fs.statSync(path.join(runDir,'snapshot','tool.sh')).mode&0o777).toBe(0o755);assertSnapshot(runDir,manifest);}); test('rejects a replaced snapshot root before following it',async()=>{const {repo,runDir}=fixture(),manifest=await capture(repo,runDir),snapshot=path.join(runDir,'snapshot'),replacement=path.join(runDir,'snapshot-replacement');fs.renameSync(snapshot,replacement);fs.symlinkSync(replacement,snapshot);expect(()=>assertSnapshot(runDir,manifest)).toThrow('private owned directory');}); diff --git a/test/cso-snapshot-state.test.ts b/test/cso-snapshot-state.test.ts index ee88ffc3f..6e09ac097 100644 --- a/test/cso-snapshot-state.test.ts +++ b/test/cso-snapshot-state.test.ts @@ -29,6 +29,9 @@ describe('CSO dirty snapshot boundary',()=>{ test('rejects a source mode change between copying and manifest persistence',async()=>{const p=repo(),run=newRun(p),source=path.join(p,'tracked.txt'),target=path.join(run.dir,'snapshot','tracked.txt'),write=fs.writeFileSync,patched=spyOn(fs,'writeFileSync').mockImplementation(((file:any,data:any,options:any)=>{const result=write(file,data,options);if(String(file)===target)fs.chmodSync(source,0o755);return result;}) as typeof fs.writeFileSync);try{await expect(capture(p,run.dir)).rejects.toThrow('Source changed during capture');}finally{patched.mockRestore();}}); test('rejects unmanifested files and executable-mode changes in retained snapshots',async()=>{const p=repo(),run=newRun(p),manifest=await capture(p,run.dir),snapshot=path.join(run.dir,'snapshot');assertSnapshot(run.dir,manifest);fs.writeFileSync(path.join(snapshot,'injected.js'),'malicious\n');expect(()=>assertSnapshot(run.dir,manifest)).toThrow('membership');fs.unlinkSync(path.join(snapshot,'injected.js'));fs.chmodSync(path.join(snapshot,'tracked.txt'),0o755);expect(()=>assertSnapshot(run.dir,manifest)).toThrow('changed');}); test('retention skips a run with a live mutation lease',()=>{const now=Date.now(),repo='d'.repeat(24),run=`${now-31*86400_000}-${'1'.repeat(16)}`,dir=secureDirectory(path.join(privateRoot(),repo,run)),payload=path.join(dir,'report.json');fs.writeFileSync(payload,'live lease payload\n',{mode:0o600});withLock(dir,()=>{retention(now);expect(fs.existsSync(payload)).toBe(true);});retention(now);expect(fs.existsSync(dir)).toBe(false);}); + test('bounded retention stops before destructive cleanup when its maintenance budget is exhausted',()=>{const now=Date.now(),repo='e'.repeat(24),run=`${now-31*86400_000}-${'8'.repeat(16)}`,dir=secureDirectory(path.join(privateRoot(),repo,run));fs.writeFileSync(path.join(dir,'report.json'),'expired private state\n',{mode:0o600});const result=retention(now,{deadlineMs:0,maxEntries:1});expect(result).toEqual({complete:false,visited:0});expect(fs.existsSync(dir)).toBe(true);retention(now);expect(fs.existsSync(dir)).toBe(false);}); + test('bounded retention completes discovery before deleting any run',()=>{const now=Date.now(),repo='f'.repeat(24),run=`${now-31*86400_000}-${'9'.repeat(16)}`,dir=secureDirectory(path.join(privateRoot(),repo,run));fs.writeFileSync(path.join(dir,'report.json'),'expired private state\n',{mode:0o600});const result=retention(now,{deadlineMs:now+60_000,maxEntries:6});expect(result).toEqual({complete:false,visited:6});expect(fs.existsSync(dir)).toBe(true);retention(now);expect(fs.existsSync(dir)).toBe(false);}); + test('bounded retention resumes an interrupted tree deletion without one recursive walk',()=>{const previous=process.env.GSTACK_HOME,isolate=tmp();process.env.GSTACK_HOME=isolate;try{const now=Date.now(),repo='1'.repeat(24),run=`${now-31*86400_000}-${'a'.repeat(16)}`,repoDir=path.join(privateRoot(),repo),dir=secureDirectory(path.join(repoDir,run)),snapshot=secureDirectory(path.join(dir,'snapshot'));for(let index=0;index<20;index++)fs.writeFileSync(path.join(snapshot,`${index}.txt`),'private state\n',{mode:0o600});const result=retention(now,{deadlineMs:now+60_000,maxEntries:30});expect(result).toEqual({complete:false,visited:30});expect(fs.existsSync(dir)).toBe(false);expect(fs.readdirSync(repoDir).some(name=>name.startsWith(`.retired-${run}-`))).toBe(true);retention(now);expect(fs.readdirSync(repoDir).some(name=>name.startsWith(`.retired-${run}-`))).toBe(false);}finally{if(previous===undefined)delete process.env.GSTACK_HOME;else process.env.GSTACK_HOME=previous;}}); test('repair bundles expire from their authenticated bundle time, independent of run age',()=>{const now=Date.now(),repo='c'.repeat(24),run=`${now-31*86400_000}-${'2'.repeat(16)}`,dir=secureDirectory(path.join(privateRoot(),repo,run)),bundles=secureDirectory(path.join(dir,'bundles')),expiredId='3'.repeat(32),retainedId='4'.repeat(32),record=(id:string,created:number)=>{const createdAt=new Date(created).toISOString();return{schemaVersion:3,id,runId:run,createdAt,expiresAt:new Date(created+30*86400_000).toISOString(),verification:{id,runId:run,createdAt}};};writeHelperJson(path.join(bundles,`${expiredId}.json`),record(expiredId,now-31*86400_000));writeHelperJson(path.join(bundles,`${retainedId}.json`),record(retainedId,now-86400_000));retention(now);expect(fs.existsSync(path.join(bundles,`${expiredId}.json`))).toBe(false);expect(fs.existsSync(path.join(bundles,`${retainedId}.json`))).toBe(true);expect(fs.existsSync(dir)).toBe(true);retention(now+30*86400_000);expect(fs.existsSync(dir)).toBe(false);}); test('replay temporary state survives cleanup failure until every detached watchdog acknowledges cleanup',()=>{const now=Date.now(),repo='b'.repeat(24),run=`${now}-${'5'.repeat(16)}`,dir=secureDirectory(path.join(privateRoot(),repo,run)),control=secureDirectory(path.join(dir,'supervision','repair-attempt')),docker=secureDirectory(path.join(dir,'preparation-execution','offline-attempt'));fs.writeFileSync(path.join(control,'attempt.ready'),'ready\n',{mode:0o600});fs.writeFileSync(path.join(control,'attempt.terminal'),'normal cleanup complete\n',{mode:0o600});fs.writeFileSync(path.join(docker,'watchdog.ready'),'ready\n',{mode:0o600});fs.writeFileSync(path.join(docker,'watchdog.event'),'cleanup incomplete; retrying exact journaled resources\n',{mode:0o600});finalizeReplayTemporary(dir);expect(fs.existsSync(path.join(dir,'.ephemeral-replay.json'))).toBe(true);retention(now);expect(fs.existsSync(control)).toBe(true);expect(fs.existsSync(docker)).toBe(true);fs.writeFileSync(path.join(control,'attempt.stopped'),'normal cleanup acknowledged\n',{mode:0o600});retention(now);expect(fs.existsSync(dir)).toBe(true);fs.writeFileSync(path.join(docker,'watchdog.stopped'),'normal cleanup acknowledged\n',{mode:0o600});retention(now);expect(fs.existsSync(dir)).toBe(false);}); test('retention recovers a hidden run retirement left by a dead helper',()=>{const now=Date.now(),repo='a'.repeat(24),run=`${now-31*86400_000}-${'6'.repeat(16)}`,retired=secureDirectory(path.join(privateRoot(),repo,`.retired-${run}-${'7'.repeat(32)}`));fs.writeFileSync(path.join(retired,'payload'),'private state\n',{mode:0o600});retention(now);expect(fs.existsSync(retired)).toBe(false);}); @@ -161,6 +164,7 @@ describe('private state and process output',()=>{ test('an active lease binds a separate decision to its candidate and active inode',()=>{const dir=tmp();withLock(dir,()=>{const leases=path.join(dir,'.mutation-lock-leases'),names=fs.readdirSync(leases);expect(names.length).toBe(3);const candidate=names.find(name=>name.endsWith('.json'))!,decision=names.find(name=>name.endsWith('.decision'))!,active=names.find(name=>name.includes('.active.'))!,candidateStat=fs.statSync(path.join(leases,candidate)),decisionStat=fs.statSync(path.join(leases,decision)),activeStat=fs.statSync(path.join(leases,active)),record=JSON.parse(fs.readFileSync(path.join(leases,decision),'utf8'));expect(`${candidateStat.dev}:${candidateStat.ino}`).toBe(`${activeStat.dev}:${activeStat.ino}`);expect(candidateStat.nlink).toBe(2);expect(activeStat.nlink).toBe(2);expect(`${decisionStat.dev}:${decisionStat.ino}`).not.toBe(`${candidateStat.dev}:${candidateStat.ino}`);expect(decisionStat.nlink).toBe(1);expect(record.kind).toBe('ticket');expect(record.candidateDev).toBe(String(candidateStat.dev));expect(record.candidateIno).toBe(String(candidateStat.ino));});expect(fs.readdirSync(path.join(dir,'.mutation-lock-leases'))).toEqual([]);}); test('post-link verification failure cleans every recorded lease phase',()=>{const dir=tmp();expect(withLock(dir,()=>1)).toBe(1);const leases=path.join(dir,'.mutation-lock-leases'),fakeToken='f'.repeat(32),fakeTemp=path.join(leases,`${fakeToken}.json.tmp.${process.pid}.deadbeef`),link=fs.linkSync,injectedOwner=JSON.stringify({pid:process.pid,token:fakeToken,createdAt:Date.now()})+'\n';let injected=false,caught:unknown;const publisher=spyOn(fs,'linkSync').mockImplementation(((source:any,target:any)=>{const result=link(source,target);if(!injected&&String(target).includes('.active.')){injected=true;fs.writeFileSync(fakeTemp,injectedOwner,{mode:0o600,flag:'wx'});}return result;}) as typeof fs.linkSync);try{withLock(dir,()=>1);}catch(error){caught=error;}finally{publisher.mockRestore();}expect(injected).toBe(true);expect(caught).toMatchObject({code:'INSUFFICIENT_CAPACITY'});expect(fs.readdirSync(leases)).toEqual([path.basename(fakeTemp)]);fs.unlinkSync(fakeTemp);expect(withLock(dir,()=>2)).toBe(2);expect(fs.readdirSync(leases)).toEqual([]);}); test('an async callback retains its lease until settlement without blocking its own event loop',async()=>{const dir=tmp();let release!:()=>void;const pending=new Promise(resolve=>{release=resolve;}),outer=withLock(dir,async()=>{await pending;return 7;}),started=Date.now();expect(()=>withLock(dir,()=>99)).toThrow('Another operation in this helper');expect(Date.now()-started).toBeLessThan(200);release();expect(await outer).toBe(7);expect(withLock(dir,()=>9)).toBe(9);}); + test('a synchronous exact-release failure is attempted only once',()=>{const dir=tmp(),lstat=fs.lstatSync;let observed=0,reader:any;try{expect(()=>withLock(dir,()=>{const leases=path.join(dir,'.mutation-lock-leases'),names=fs.readdirSync(leases),candidate=path.join(leases,names.find(name=>name.endsWith('.json'))!),active=path.join(leases,names.find(name=>name.includes('.active.'))!),value=fs.readFileSync(active);fs.unlinkSync(active);fs.writeFileSync(active,value,{mode:0o600,flag:'wx'});reader=spyOn(fs,'lstatSync').mockImplementation(((file:any,options?:any)=>{if(String(file)===candidate)observed++;return options===undefined?lstat(file):lstat(file,options);}) as typeof fs.lstatSync);})).toThrow('active phase changed before cleanup');expect(observed).toBe(1);}finally{reader?.mockRestore();}}); test('exact release rejects active or decision inode substitution and overrides callback success or failure',()=>{for(const phase of ['active','decision'] as const){const dir=tmp();let replacement='';expect(()=>withLock(dir,()=>{const leases=path.join(dir,'.mutation-lock-leases'),name=fs.readdirSync(leases).find(value=>phase==='active'?value.includes('.active.'):value.endsWith('.decision'))!,lease=path.join(leases,name),value=fs.readFileSync(lease);fs.unlinkSync(lease);fs.writeFileSync(lease,value,{mode:0o600,flag:'wx'});replacement=lease;if(phase==='active')throw new Error('callback failed');return 1;})).toThrow(/changed before (cleanup|exact release)/);expect(fs.existsSync(replacement)).toBe(true);fs.rmSync(dir,{recursive:true,force:true});}}); test('concurrent stale-lock recovery never admits overlapping report writers',async()=>{const dir=tmp(),lock=path.join(dir,'.mutation-lock'),script=path.join(dir,'racer.ts');fs.mkdirSync(lock);fs.writeFileSync(path.join(lock,'owner.json'),JSON.stringify({pid:2147483647,token:'stale',createdAt:0,expiresAt:0}));fs.writeFileSync(script,`import fs from 'node:fs';import path from 'node:path';import {withLock} from ${JSON.stringify(path.resolve(import.meta.dir,'../lib/cso/state.ts'))};const dir=process.env.RACE_DIR!;try{await withLock(dir,async()=>{const active=path.join(dir,'active');try{fs.writeFileSync(active,String(process.pid),{flag:'wx'});}catch{fs.appendFileSync(path.join(dir,'overlap'),'yes\\n');}await Bun.sleep(50);try{if(fs.readFileSync(active,'utf8')===String(process.pid))fs.unlinkSync(active);}catch{}});}catch{}`);const children=Array.from({length:8},()=>spawn(process.execPath,[script],{env:{...process.env,RACE_DIR:dir},stdio:'ignore'}));await Promise.all(children.map(child=>new Promise(resolve=>child.on('close',()=>resolve()))));expect(fs.existsSync(path.join(dir,'overlap'))).toBe(false);}); test('concurrent recovery of one dead hard-link publication has a winner and no raw race errors',async()=>{