diff --git a/packages/core/src/detection-engine/monitor.ts b/packages/core/src/detection-engine/monitor.ts index 33726d2..e98d2e3 100644 --- a/packages/core/src/detection-engine/monitor.ts +++ b/packages/core/src/detection-engine/monitor.ts @@ -40,7 +40,6 @@ export type PoolEventType = 'quarantined' | 'released'; const MAX_RECENT_TIMES = 100; const MAX_PATTERN_LENGTH = 100; -const MIN_SAMPLES_FOR_STATS = 10; function mean(values: number[]): number { /* v8 ignore start -- measured-unreachable path, see the coverage PR notes */ @@ -49,15 +48,6 @@ function mean(values: number[]): number { return values.reduce((a, b) => a + b, 0) / values.length; } -function stdev(values: number[]): number { - /* v8 ignore start -- measured-unreachable path, see the coverage PR notes */ - if (values.length <= 1) return 0; - /* v8 ignore stop -- measured-unreachable path, see the coverage PR notes */ - const avg = mean(values); - const squareDiffs = values.map((v) => (v - avg) ** 2); - return Math.sqrt(squareDiffs.reduce((a, b) => a + b, 0) / (values.length - 1)); -} - function truncatePattern(pattern: string): string { return pattern.length > 50 ? pattern.slice(0, 50) + '...' : pattern; } @@ -75,6 +65,8 @@ export class PerformanceMonitor { private readonly anomalyThreshold: number; private readonly slowPatternThreshold: number; private readonly historySize: number; + private readonly minSamplesForAnomaly: number; + private readonly recentTimesWindow: number; private readonly maxTrackedPatterns: number; /* The reference detection_anomaly_emission_cooldown: minimum seconds between anomaly events for the same pattern. */ @@ -91,12 +83,15 @@ export class PerformanceMonitor { historySize = 1000, maxTrackedPatterns = 1000, emissionCooldownSeconds = 60, + minSamplesForAnomaly = 30, ) { this.anomalyThreshold = Math.max(1.0, Math.min(10.0, anomalyThreshold)); this.slowPatternThreshold = Math.max(0.01, Math.min(10.0, slowPatternThreshold)); this.historySize = Math.max(100, Math.min(10000, historySize)); this.maxTrackedPatterns = Math.max(100, Math.min(5000, maxTrackedPatterns)); this.emissionCooldownSeconds = Math.max(1.0, Math.min(3600.0, emissionCooldownSeconds)); + this.minSamplesForAnomaly = Math.max(10, Math.min(1000, Math.trunc(minSamplesForAnomaly))); + this.recentTimesWindow = Math.max(this.minSamplesForAnomaly, MAX_RECENT_TIMES); } async recordMetric( @@ -158,7 +153,7 @@ export class PerformanceMonitor { if (!timeout) { stats.recentTimes.push(executionTime); - if (stats.recentTimes.length > MAX_RECENT_TIMES) { + if (stats.recentTimes.length > this.recentTimesWindow) { stats.recentTimes.shift(); } stats.maxExecutionTime = Math.max(stats.maxExecutionTime, executionTime); @@ -171,6 +166,36 @@ export class PerformanceMonitor { await this.checkAnomalies(metric, agentHandler, correlationId); } + private detectStatisticalAnomaly( + metric: PerformanceMetric, + stats: PatternStats | undefined, + ): Record | null { + if (!stats || stats.recentTimes.length < this.minSamplesForAnomaly) return null; + const times = stats.recentTimes; + const count = times.length; + let sum = 0; + for (let i = 0; i < count; i++) sum += times[i]!; + const avgTime = sum / count; + if (this.anomalyThreshold >= 0 && metric.executionTime <= avgTime) return null; + let squares = 0; + for (let i = 0; i < count; i++) { + const diff = times[i]! - avgTime; + squares += diff * diff; + } + const stdTime = Math.sqrt(squares / (count - 1)); + if (stdTime <= 0) return null; + const zScore = (metric.executionTime - avgTime) / stdTime; + if (zScore <= this.anomalyThreshold) return null; + return { + type: 'statistical_anomaly', + pattern: metric.pattern, + executionTime: metric.executionTime, + zScore, + avgTime, + stdTime, + }; + } + private async checkAnomalies( metric: PerformanceMetric, agentHandler: AgentHandlerProtocol | null, @@ -194,23 +219,8 @@ export class PerformanceMonitor { } const stats = this.patternStats.get(metric.pattern); - if (stats && stats.recentTimes.length >= MIN_SAMPLES_FOR_STATS) { - const avgTime = mean(stats.recentTimes); - const stdTime = stdev(stats.recentTimes); - if (stdTime > 0) { - const zScore = (metric.executionTime - avgTime) / stdTime; - if (Math.abs(zScore) > this.anomalyThreshold) { - anomalies.push({ - type: 'statistical_anomaly', - pattern: metric.pattern, - executionTime: metric.executionTime, - zScore, - avgTime, - stdTime, - }); - } - } - } + const statistical = this.detectStatisticalAnomaly(metric, stats); + if (statistical) anomalies.push(statistical); if (agentHandler && anomalies.length > 0) { /* The reference emission cooldown: at most one anomaly event batch per @@ -237,6 +247,8 @@ export class PerformanceMonitor { } } + if (this.anomalyCallbacks.length === 0) return; + for (const anomaly of anomalies) { const safe: Record = { ...anomaly }; if (typeof safe['pattern'] === 'string') { diff --git a/packages/core/src/detection-engine/scan-pool.ts b/packages/core/src/detection-engine/scan-pool.ts index f415ed1..4edc3b6 100644 --- a/packages/core/src/detection-engine/scan-pool.ts +++ b/packages/core/src/detection-engine/scan-pool.ts @@ -192,9 +192,15 @@ export class ScanPoolSupervisor { * contextvars; here one instance rides the request scan explicitly. * * The budget counts CPU time actually spent scanning (process.cpuUsage() - * deltas around each synchronous scan, user plus system), not wall time, so a - * busy or CPU-starved host does not push ordinary requests over the cap. The - * clock falls back to wall time on runtimes without process.cpuUsage (edge). + * deltas, user plus system), not wall time, so a busy or CPU-starved host + * does not push ordinary requests over the cap. The CPU is charged per pass + * (one loop over the pattern table), not per pattern: `openPass()` reads the + * clock once, `yieldIfDue()` charges at each EVENT_LOOP_SLICE_MS boundary + * (before the yield, restarting after, so other tasks' CPU is never + * charged), `suspended()` does the same around any await inside the pass, + * and `closePass()` makes the final charge. A budget without a cap never + * reads the CPU clock. The clock falls back to wall time on runtimes without + * process.cpuUsage (edge). * A synchronous RegExp cannot be preempted, so the budget is enforced * between patterns: once the summed per-pattern scan time reaches the cap * the remaining patterns are skipped and the caller records the exhaustion @@ -224,6 +230,7 @@ export class ScanTimeBudget { private spentMs = 0; private sliceStartMs = performance.now(); private warned = false; + private markSeconds: number | null = null; /** capSeconds Infinity means no cap (the reference's None). */ constructor( @@ -243,13 +250,36 @@ export class ScanTimeBudget { this.spentMs += elapsedSeconds * 1000; } - /** Runs one synchronous scan and charges its CPU time to the budget. */ - measure(scan: () => T): T { - const start = this.clock(); + /** Opens a pass: one CPU clock read, none without a cap. */ + openPass(): void { + this.markSeconds = Number.isFinite(this.capSeconds) ? this.clock() : null; + } + + /** Charges the CPU spent since the last mark; a no-op outside a pass. */ + charge(): void { + if (this.markSeconds === null) return; + const now = this.clock(); + this.record(now - this.markSeconds); + this.markSeconds = now; + } + + /** Restarts the mark after a suspension so the gap is not charged. */ + restart(): void { + if (this.markSeconds !== null) this.markSeconds = this.clock(); + } + + closePass(): void { + this.charge(); + this.markSeconds = null; + } + + /** Awaits work that suspends the pass, charging before and restarting after. */ + async suspended(work: Promise): Promise { + this.charge(); try { - return scan(); + return await work; } finally { - this.record(this.clock() - start); + this.restart(); } } @@ -262,7 +292,9 @@ export class ScanTimeBudget { async yieldIfDue(): Promise { if (performance.now() - this.sliceStartMs < EVENT_LOOP_SLICE_MS) return; + this.charge(); await yieldToEventLoop(); + this.restart(); this.sliceStartMs = performance.now(); } } diff --git a/packages/core/src/handlers/sus-patterns.ts b/packages/core/src/handlers/sus-patterns.ts index 41d15be..86aab7b 100644 --- a/packages/core/src/handlers/sus-patterns.ts +++ b/packages/core/src/handlers/sus-patterns.ts @@ -245,6 +245,7 @@ export class SusPatternsManager { config.detectionMonitorHistorySize, config.detectionMaxTrackedPatterns, config.detectionAnomalyEmissionCooldown, + config.detectionMinSamplesForAnomaly, ); this.scanPool = scanPool ?? new ScanPoolSupervisor(config.detectionCompilerTimeout * 1000); this.scanPool.onPoolEvent((event) => { @@ -352,10 +353,25 @@ export class SusPatternsManager { ipAddress?: string; } = {}, ): Promise<{ threats: InternalRegexThreat[]; matchedPatterns: string[]; timeouts: string[] }> { - const { rawViewOnly = null, urlDecodedViewOnly = null } = options; // Without a request budget (standalone detectPatternMatch) the scan still - // yields between slices, it just has no cap. + // yields between slices, it just has no cap. The CPU is charged per pass. const scanTime = options.scanTime ?? new ScanTimeBudget(); + scanTime.openPass(); + try { + return await this.scanPatternTable(content, context, correlationId, options, scanTime); + } finally { + scanTime.closePass(); + } + } + + private async scanPatternTable( + content: string, + context: string, + correlationId: string | null, + options: { rawViewOnly?: boolean | null; urlDecodedViewOnly?: boolean | null; ipAddress?: string }, + scanTime: ScanTimeBudget, + ): Promise<{ threats: InternalRegexThreat[]; matchedPatterns: string[]; timeouts: string[] }> { + const { rawViewOnly = null, urlDecodedViewOnly = null } = options; const threats: InternalRegexThreat[] = []; const matchedPatterns: string[] = []; const timeouts: string[] = []; @@ -505,19 +521,16 @@ export class SusPatternsManager { if (!this.scanWorkerPoolEnabled || SusPatternsManager.hasStructuralMatcher(pattern.source)) { return this.scanPool.run( pattern.source, - () => scanTime.measure( - () => this.checkRegexPattern(pattern, content, pattern.category, validatorContext, binaryPrefix), - ), + () => this.checkRegexPattern(pattern, content, pattern.category, validatorContext, binaryPrefix), ); } - const pool = await this.ensureWorkerPool(); + // Only the first scan builds the pool (a real suspension); later scans reuse it. + const pool = this.workerPool ?? await scanTime.suspended(this.ensureWorkerPool()); if (pool === null) { return this.scanPool.run( pattern.source, - () => scanTime.measure( - () => this.checkRegexPattern(pattern, content, pattern.category, validatorContext, binaryPrefix), - ), + () => this.checkRegexPattern(pattern, content, pattern.category, validatorContext, binaryPrefix), ); } @@ -529,7 +542,7 @@ export class SusPatternsManager { ? pattern.compiled.re.flags.replace('y', 'gm') : pattern.compiled.re.flags.replace('y', 'g'); // The worker's CPU is not observable from here, so its wall time is charged. - const outcome = await pool.run( + const outcome = await scanTime.suspended(pool.run( pattern.source, { patternSource: pattern.compiled.re.source, flags, content }, (candidates) => @@ -540,7 +553,7 @@ export class SusPatternsManager { validatorContext, binaryPrefix, ), - ); + )); scanTime.record(outcome.elapsedMs / 1000); return outcome; } @@ -553,11 +566,11 @@ export class SusPatternsManager { ); } - /* Lazily built on the first worker-path scan; a runtime without - worker_threads falls back to the inline path with a single warning - (the knob is opt-in and edge runtimes have no pool to run). */ + /* Lazily built on the first worker-path scan (the caller reuses a built + pool without a suspension); a runtime without worker_threads falls back + to the inline path with a single warning (the knob is opt-in and edge + runtimes have no pool to run). */ private async ensureWorkerPool(): Promise { - if (this.workerPool !== null) return this.workerPool; if (this.workerPoolUnavailable) return null; const pool = await createWorkerScanPool(this.scanPool.deadline, { logger: this.logger }); if (pool === null) { diff --git a/packages/core/src/models/config.ts b/packages/core/src/models/config.ts index a59753a..2fc5dae 100644 --- a/packages/core/src/models/config.ts +++ b/packages/core/src/models/config.ts @@ -349,7 +349,7 @@ export const SecurityConfigSchema = z.object({ detectionMaxScanValues: z.number().int().positive().default(512), detectionMaxScanChars: z.number().int().positive().default(65536), detectionMaxJsonDepth: z.number().int().positive().default(32), - detectionMinSamplesForAnomaly: z.number().int().positive().default(10), + detectionMinSamplesForAnomaly: z.number().int().min(10).max(1000).default(30), /* The GeoIP manager surface (reference ipinfo_token / ipinfo_db_path / geo_ip_db_max_age): the token authenticates the ipinfo.io download, the path is the local MMDB file and the max age is the mtime freshness gate diff --git a/packages/core/tests/conformance/events-harness.ts b/packages/core/tests/conformance/events-harness.ts index 568eb07..0efa9d7 100644 --- a/packages/core/tests/conformance/events-harness.ts +++ b/packages/core/tests/conformance/events-harness.ts @@ -395,6 +395,8 @@ export class EventsCaseRun { (drive['slow_pattern_threshold'] as number) ?? 0.1, 1000, 1000, + 60, + (drive['min_samples_for_anomaly'] as number) ?? 30, ); for (const sample of (drive['samples'] as Array>) ?? []) { await monitor.recordMetric( diff --git a/packages/core/tests/test-core/small-parity-gaps.test.ts b/packages/core/tests/test-core/small-parity-gaps.test.ts index db89207..d092c65 100644 --- a/packages/core/tests/test-core/small-parity-gaps.test.ts +++ b/packages/core/tests/test-core/small-parity-gaps.test.ts @@ -289,7 +289,7 @@ describe('lazyInit + scan tunable config keys (gaps 19, 97)', () => { expect(config.detectionMaxScanValues).toBe(512); expect(config.detectionMaxScanChars).toBe(65536); expect(config.detectionMaxJsonDepth).toBe(32); - expect(config.detectionMinSamplesForAnomaly).toBe(10); + expect(config.detectionMinSamplesForAnomaly).toBe(30); }); it('carries the lazyInit default', () => { diff --git a/packages/core/tests/test-detection-engine/monitor.test.ts b/packages/core/tests/test-detection-engine/monitor.test.ts index 9ab2266..d3b55aa 100644 --- a/packages/core/tests/test-detection-engine/monitor.test.ts +++ b/packages/core/tests/test-detection-engine/monitor.test.ts @@ -137,7 +137,7 @@ describe('PerformanceMonitor', () => { const callback = vi.fn(); monitor.registerAnomalyCallback(callback); - for (let i = 0; i < 20; i++) { + for (let i = 0; i < 35; i++) { await monitor.recordMetric('pattern-x', 0.01, 100, false); } @@ -149,6 +149,52 @@ describe('PerformanceMonitor', () => { expect(statisticalCalls.length).toBeGreaterThan(0); }); + it('does not flag a faster-than-average sample as a statistical anomaly', async () => { + const monitor = new PerformanceMonitor(2.0, 10); + const callback = vi.fn(); + monitor.registerAnomalyCallback(callback); + + for (let i = 0; i < 40; i++) { + await monitor.recordMetric('pattern-y', 1.0 + (i % 2) * 0.01, 100, false); + } + await monitor.recordMetric('pattern-y', 0.0, 100, false); + expect(callback).not.toHaveBeenCalled(); + + await monitor.recordMetric('pattern-y', 5.0, 100, false); + expect(callback).toHaveBeenCalledTimes(1); + expect((callback.mock.calls[0]![0] as Record)['type']).toBe( + 'statistical_anomaly', + ); + }); + + it('skips the statistical check when the recorded samples have no spread', async () => { + const monitor = new PerformanceMonitor(2.0, 10); + const callback = vi.fn(); + monitor.registerAnomalyCallback(callback); + + for (let i = 0; i < 35; i++) { + await monitor.recordMetric('pattern-flat', 0.5, 100, false); + } + await monitor.recordMetric('pattern-flat', 5.0, 100, false, true); + + const statisticalCalls = callback.mock.calls.filter( + (c) => (c[0] as Record)['type'] === 'statistical_anomaly', + ); + expect(statisticalCalls).toHaveLength(0); + }); + + it('waits for the configured minimum sample count', async () => { + const monitor = new PerformanceMonitor(2.0, 10, 1000, 1000, 60, 50); + const callback = vi.fn(); + monitor.registerAnomalyCallback(callback); + + for (let i = 0; i < 40; i++) { + await monitor.recordMetric('pattern-z', 0.01 + (i % 2) * 0.001, 100, false); + } + await monitor.recordMetric('pattern-z', 5.0, 100, false); + expect(callback).not.toHaveBeenCalled(); + }); + it('sends anomaly event to agent handler', async () => { const monitor = new PerformanceMonitor(3.0, 0.001); const agent = { sendEvent: vi.fn() }; diff --git a/packages/core/tests/test-detection-engine/scan-time-budget.test.ts b/packages/core/tests/test-detection-engine/scan-time-budget.test.ts index 7e647f0..53cb8a5 100644 --- a/packages/core/tests/test-detection-engine/scan-time-budget.test.ts +++ b/packages/core/tests/test-detection-engine/scan-time-budget.test.ts @@ -168,21 +168,97 @@ describe('the budget counts CPU time, not wall time', () => { it('CPU-heavy scanning still exhausts the budget', async () => { const m = manager(); + // Charged at slice boundaries now, so make each scan wall-slow (1 ms) to + // cross them; the injected CPU clock jumps 1 s on every read. + vi.spyOn(m as unknown as { checkRegexPattern: () => null }, 'checkRegexPattern').mockImplementation(() => { + const end = performance.now() + 1; + while (performance.now() < end) { /* busy wait */ } + return null; + }); let cpu = 0; const scanTime = new ScanTimeBudget(0.1, () => (cpu += 1)); const result = await m.detect('hello world', '127.0.0.1', 'query_param:q', null, { scanTime }); expect(result.threats.some((t) => t.pattern === BUDGET_PATTERN)).toBe(true); }); - it('measure charges the clock delta even when the scan throws', () => { + it('a pass charges the clock delta at close even when the scan throws', async () => { let cpu = 0; const scanTime = new ScanTimeBudget(1.0, () => (cpu += 0.6)); - expect(() => scanTime.measure(() => { throw new Error('boom'); })).toThrow('boom'); + scanTime.openPass(); + try { + throw new Error('boom'); + } catch { + // the engine closes the pass from a finally block + } finally { + scanTime.closePass(); + } expect(scanTime.exhausted).toBe(false); - scanTime.measure(() => 1); + scanTime.openPass(); + scanTime.closePass(); expect(scanTime.exhausted).toBe(true); }); + it('never reads the CPU clock without a cap', async () => { + const clock = vi.fn(() => 0); + const scanTime = new ScanTimeBudget(Infinity, clock); + scanTime.openPass(); + await scanTime.suspended(Promise.resolve()); + scanTime.closePass(); + expect(clock).not.toHaveBeenCalled(); + }); + + it('reads the CPU clock a few times per pass, not per pattern', async () => { + const m = manager(); + const scanned = vi.spyOn(m as unknown as { checkRegexPattern: () => null }, 'checkRegexPattern'); + const clock = vi.fn(() => 0); + const scanTime = new ScanTimeBudget(5, clock); + await m.detect('hello world', '127.0.0.1', 'query_param:q', null, { scanTime }); + expect(scanned.mock.calls.length).toBeGreaterThan(100); + expect(clock.mock.calls.length).toBeLessThanOrEqual(2 * 4 + 2 * 4); + expect(clock.mock.calls.length * 10).toBeLessThan(scanned.mock.calls.length); + }); + + it('a long pass stops within about one slice of the budget', async () => { + const m = manager(); + let cpu = 0; + // Each pattern costs 0.5 ms of wall and of fake CPU. + vi.spyOn(m as unknown as { checkRegexPattern: () => null }, 'checkRegexPattern').mockImplementation(() => { + const end = performance.now() + 0.5; + while (performance.now() < end) { /* busy wait */ } + cpu += 0.0005; + return null; + }); + const cap = 0.02; + const scanTime = new ScanTimeBudget(cap, () => cpu); + const result = await m.detect('hello world', '127.0.0.1', 'query_param:q', null, { scanTime }); + expect(result.threats.some((t) => t.pattern === BUDGET_PATTERN)).toBe(true); + // One slice (2 ms) plus one pattern (0.5 ms) of lag, with slack for timer jitter. + expect(cpu - cap).toBeGreaterThanOrEqual(0); + expect(cpu - cap).toBeLessThan(0.008); + }); + + it('does not charge another task\'s CPU burned while the pass is suspended', async () => { + const burn = (ms: number): Promise => new Promise((resolve) => { + setImmediate(() => { + const end = performance.now() + ms; + while (performance.now() < end) { /* busy wait */ } + resolve(); + }); + }); + const suspendedBudget = new ScanTimeBudget(0.03); + suspendedBudget.openPass(); + await suspendedBudget.suspended(burn(120)); + suspendedBudget.closePass(); + expect(suspendedBudget.exhausted).toBe(false); + + // Control: the same burn awaited without suspended() is charged. + const plainBudget = new ScanTimeBudget(0.03); + plainBudget.openPass(); + await burn(120); + plainBudget.closePass(); + expect(plainBudget.exhausted).toBe(true); + }); + it('cpuSeconds reads process CPU time and falls back to wall time without process', () => { expect(cpuSeconds()).toBeGreaterThan(0); vi.stubGlobal('process', undefined);