diff --git a/backend/src/__tests__/database-matching-policy.test.ts b/backend/src/__tests__/database-matching-policy.test.ts new file mode 100644 index 00000000..3283a001 --- /dev/null +++ b/backend/src/__tests__/database-matching-policy.test.ts @@ -0,0 +1,111 @@ +/** + * Pins the matching and ordering behavior of `DatabaseService.getMatchingPolicy`. + * + * The matcher must be deterministic across replicas: two policies in the same + * scope class (e.g. both fleet-wide stack-wildcard) need to resolve to the same + * winner regardless of SQLite row-iteration order. The chosen tiebreaker is + * lowest id wins, so the oldest-defined policy stays authoritative. + */ +import { describe, it, expect, beforeAll, afterAll } from 'vitest'; +import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb'; + +let tmpDir: string; +let DatabaseService: typeof import('../services/DatabaseService').DatabaseService; + +beforeAll(async () => { + tmpDir = await setupTestDb(); + ({ DatabaseService } = await import('../services/DatabaseService')); +}); + +afterAll(() => { + cleanupTestDb(tmpDir); +}); + +describe('getMatchingPolicy tiebreaker', () => { + it('returns the lowest-id row when two policies tie on scope class', () => { + const db = DatabaseService.getInstance(); + const first = db.createScanPolicy({ + name: 'first-fleet-wide', + node_id: null, + node_identity: '', + stack_pattern: null, + max_severity: 'HIGH', + block_on_deploy: 0, + enabled: 1, + replicated_from_control: 0, + }); + const second = db.createScanPolicy({ + name: 'second-fleet-wide', + node_id: null, + node_identity: '', + stack_pattern: null, + max_severity: 'CRITICAL', + block_on_deploy: 0, + enabled: 1, + replicated_from_control: 0, + }); + const winner = db.getMatchingPolicy(1, 'web', 'local'); + expect(winner?.id).toBe(first.id); + // Sanity: with first deleted, the next-lowest takes over. + db.deleteScanPolicy(first.id); + const next = db.getMatchingPolicy(1, 'web', 'local'); + expect(next?.id).toBe(second.id); + }); + + it('prefers node-scoped over fleet-wide regardless of id order', () => { + const db = DatabaseService.getInstance(); + const fleetWide = db.createScanPolicy({ + name: 'tie-fleet', + node_id: null, + node_identity: '', + stack_pattern: null, + max_severity: 'LOW', + block_on_deploy: 0, + enabled: 1, + replicated_from_control: 0, + }); + const nodeScoped = db.createScanPolicy({ + name: 'tie-node', + node_id: 1, + node_identity: 'local', + stack_pattern: null, + max_severity: 'CRITICAL', + block_on_deploy: 0, + enabled: 1, + replicated_from_control: 0, + }); + const winner = db.getMatchingPolicy(1, 'web', 'local'); + // Node-scoped wins by class, even though its id is higher than the fleet-wide row. + expect(winner?.id).toBe(nodeScoped.id); + db.deleteScanPolicy(fleetWide.id); + db.deleteScanPolicy(nodeScoped.id); + }); + + it('respects identity matching for replicated rows', () => { + const db = DatabaseService.getInstance(); + const otherIdentity = db.createScanPolicy({ + name: 'replicated-other', + node_id: null, + node_identity: 'https://other.example', + stack_pattern: null, + max_severity: 'CRITICAL', + block_on_deploy: 0, + enabled: 1, + replicated_from_control: 1, + }); + const ourIdentity = db.createScanPolicy({ + name: 'replicated-self', + node_id: null, + node_identity: 'https://me.example', + stack_pattern: null, + max_severity: 'HIGH', + block_on_deploy: 0, + enabled: 1, + replicated_from_control: 1, + }); + const winner = db.getMatchingPolicy(1, 'web', 'https://me.example'); + expect(winner?.id).toBe(ourIdentity.id); + db.deleteScanPolicy(otherIdentity.id); + db.deleteScanPolicy(ourIdentity.id); + }); +}); diff --git a/backend/src/__tests__/fleet-sync-routes.test.ts b/backend/src/__tests__/fleet-sync-routes.test.ts index cbd91aec..c2b12ac9 100644 --- a/backend/src/__tests__/fleet-sync-routes.test.ts +++ b/backend/src/__tests__/fleet-sync-routes.test.ts @@ -131,3 +131,59 @@ describe('GET /api/fleet/sync-status', () => { vi.restoreAllMocks(); }); }); + +describe('POST /api/fleet/sync/:resource pushedAt protocol', () => { + const validRow = { + name: 'from-control', + node_identity: '', + stack_pattern: null, + max_severity: 'CRITICAL' as const, + block_on_deploy: 0, + enabled: 1, + }; + + it('accepts payloads without pushedAt for back-compat with legacy controls', async () => { + const res = await request(app) + .post('/api/fleet/sync/scan_policies') + .set('Authorization', nodeProxyAuthHeader) + .send({ rows: [validRow], targetIdentity: 'https://me.example' }); + expect(res.status).toBe(200); + expect(res.body.success).toBe(true); + }); + + it('accepts a fresh pushedAt and persists it for stale-rejection compare', async () => { + const res = await request(app) + .post('/api/fleet/sync/scan_policies') + .set('Authorization', nodeProxyAuthHeader) + .send({ rows: [validRow], targetIdentity: 'https://me.example', pushedAt: 1_700_000_000_000 }); + expect(res.status).toBe(200); + }); + + it('rejects a stale pushedAt with 409 STALE_SYNC_PUSH', async () => { + // Send fresh, then send strictly-older. + await request(app) + .post('/api/fleet/sync/scan_policies') + .set('Authorization', nodeProxyAuthHeader) + .send({ rows: [validRow], targetIdentity: 'https://me.example', pushedAt: 1_800_000_000_000 }); + const stale = await request(app) + .post('/api/fleet/sync/scan_policies') + .set('Authorization', nodeProxyAuthHeader) + .send({ rows: [validRow], targetIdentity: 'https://me.example', pushedAt: 1_700_000_000_000 }); + expect(stale.status).toBe(409); + expect(stale.body.code).toBe('STALE_SYNC_PUSH'); + }); + + it('returns a friendly 413 SYNC_PAYLOAD_TOO_LARGE when the body exceeds the parser limit', async () => { + // ~6 MB of padding pushes past the 5mb route-level limit. Keeps a single + // valid row so any path that did parse would succeed; we want the parser + // to reject before the handler runs. + const padding = 'x'.repeat(6 * 1024 * 1024); + const res = await request(app) + .post('/api/fleet/sync/scan_policies') + .set('Authorization', nodeProxyAuthHeader) + .send({ rows: [validRow], targetIdentity: 'https://me.example', pad: padding }); + expect(res.status).toBe(413); + expect(res.body.code).toBe('SYNC_PAYLOAD_TOO_LARGE'); + expect(res.body.error).toMatch(/Sync payload too large/); + }); +}); diff --git a/backend/src/__tests__/fleet-sync-service.test.ts b/backend/src/__tests__/fleet-sync-service.test.ts index 326eac58..fad59036 100644 --- a/backend/src/__tests__/fleet-sync-service.test.ts +++ b/backend/src/__tests__/fleet-sync-service.test.ts @@ -7,22 +7,28 @@ import { describe, it, expect, vi, beforeEach } from 'vitest'; const { mockGetNodes, mockGetNode, - mockGetScanPolicies, + mockGetLocalScanPolicies, + mockGetLocalCveSuppressions, mockReplaceReplicatedScanPolicies, mockRecordFleetSyncSuccess, mockRecordFleetSyncFailure, mockGetSystemState, mockSetSystemState, + mockTransaction, + mockDispatchAlert, mockAxiosPost, } = vi.hoisted(() => ({ mockGetNodes: vi.fn().mockReturnValue([]), mockGetNode: vi.fn(), - mockGetScanPolicies: vi.fn().mockReturnValue([]), + mockGetLocalScanPolicies: vi.fn().mockReturnValue([]), + mockGetLocalCveSuppressions: vi.fn().mockReturnValue([]), mockReplaceReplicatedScanPolicies: vi.fn(), mockRecordFleetSyncSuccess: vi.fn(), mockRecordFleetSyncFailure: vi.fn(), mockGetSystemState: vi.fn().mockReturnValue(null), mockSetSystemState: vi.fn(), + mockTransaction: vi.fn().mockImplementation((fn: () => unknown) => fn()), + mockDispatchAlert: vi.fn().mockResolvedValue(undefined), mockAxiosPost: vi.fn().mockResolvedValue({ data: { success: true } }), })); @@ -30,12 +36,14 @@ vi.mock('../services/DatabaseService', () => ({ DatabaseService: { getInstance: () => ({ getNodes: mockGetNodes, - getScanPolicies: mockGetScanPolicies, + getLocalScanPolicies: mockGetLocalScanPolicies, + getLocalCveSuppressions: mockGetLocalCveSuppressions, replaceReplicatedScanPolicies: mockReplaceReplicatedScanPolicies, recordFleetSyncSuccess: mockRecordFleetSyncSuccess, recordFleetSyncFailure: mockRecordFleetSyncFailure, getSystemState: mockGetSystemState, setSystemState: mockSetSystemState, + transaction: mockTransaction, }), }, })); @@ -48,6 +56,16 @@ vi.mock('../services/NodeRegistry', () => ({ }, })); +vi.mock('../services/NotificationService', () => ({ + NotificationService: { + getInstance: () => ({ dispatchAlert: mockDispatchAlert }), + }, +})); + +vi.mock('../utils/debug', () => ({ + isDebugEnabled: () => false, +})); + vi.mock('axios', () => ({ default: { post: mockAxiosPost }, AxiosError: class AxiosError extends Error { @@ -128,9 +146,9 @@ describe('FleetSyncService.pushResource', () => { { id: 2, type: 'remote', api_url: 'https://a.example', api_token: 'tokA', name: 'A' }, { id: 3, type: 'remote', api_url: 'https://b.example', api_token: 'tokB', name: 'B' }, ]); - mockGetScanPolicies.mockReturnValue([ + // getLocalScanPolicies() filters out replicated rows at SQL time. + mockGetLocalScanPolicies.mockReturnValue([ { id: 1, name: 'local-1', node_identity: '', replicated_from_control: 0, created_at: 1, updated_at: 1 }, - { id: 2, name: 'mirrored', node_identity: 'https://somewhere', replicated_from_control: 1, created_at: 1, updated_at: 1 }, ]); await FleetSyncService.getInstance().pushResource('scan_policies'); @@ -180,3 +198,177 @@ describe('FleetSyncService.applyIncomingSync', () => { expect(mockSetSystemState).not.toHaveBeenCalledWith('fleet_self_identity', expect.anything()); }); }); + +describe('FleetSyncService payload schema', () => { + it('includes pushedAt and controlIdentity in every push body', async () => { + mockGetNodes.mockReturnValue([ + { id: 2, type: 'remote', api_url: 'https://a.example', api_token: 'tokA', name: 'A' }, + ]); + mockGetLocalScanPolicies.mockReturnValue([ + { id: 1, name: 'one', node_identity: '', replicated_from_control: 0, created_at: 1, updated_at: 1 }, + ]); + await FleetSyncService.getInstance().pushResource('scan_policies'); + const body = mockAxiosPost.mock.calls[0][1]; + expect(typeof body.pushedAt).toBe('number'); + expect(body.pushedAt).toBeGreaterThan(0); + expect(body).toHaveProperty('controlIdentity'); + expect(typeof body.controlIdentity).toBe('string'); + }); + + it('emits strictly increasing pushedAt across consecutive pushes', async () => { + mockGetNodes.mockReturnValue([ + { id: 2, type: 'remote', api_url: 'https://a.example', api_token: 'tokA', name: 'A' }, + ]); + mockGetLocalScanPolicies.mockReturnValue([]); + await FleetSyncService.getInstance().pushResource('scan_policies'); + await FleetSyncService.getInstance().pushResource('scan_policies'); + await FleetSyncService.getInstance().pushResource('scan_policies'); + const stamps = mockAxiosPost.mock.calls.map((c) => c[1].pushedAt); + expect(stamps).toHaveLength(3); + expect(stamps[1]).toBeGreaterThan(stamps[0]); + expect(stamps[2]).toBeGreaterThan(stamps[1]); + }); +}); + +describe('FleetSyncService per-node push serialization', () => { + it('serializes pushes to the same node so a second push waits for the first', async () => { + mockGetNodes.mockReturnValue([ + { id: 2, type: 'remote', api_url: 'https://a.example', api_token: 'tokA', name: 'A' }, + ]); + mockGetLocalScanPolicies.mockReturnValue([]); + + let resolveFirst: (() => void) | null = null; + const firstStarted = new Promise((resolveStart) => { + mockAxiosPost.mockImplementationOnce(() => { + resolveStart(); + return new Promise((resolve) => { + resolveFirst = () => resolve({ data: { success: true } }); + }); + }); + }); + mockAxiosPost.mockImplementationOnce(() => Promise.resolve({ data: { success: true } })); + + const first = FleetSyncService.getInstance().pushResource('scan_policies'); + await firstStarted; + + const second = FleetSyncService.getInstance().pushResource('scan_policies'); + // After awaiting a tick, only the first push should have hit axios. + await Promise.resolve(); + expect(mockAxiosPost).toHaveBeenCalledTimes(1); + + resolveFirst!(); + await Promise.all([first, second]); + expect(mockAxiosPost).toHaveBeenCalledTimes(2); + }); +}); + +describe('FleetSyncService row cap', () => { + it('truncates to MAX_SYNC_ROWS and emits a warning notification', async () => { + mockGetNodes.mockReturnValue([ + { id: 2, type: 'remote', api_url: 'https://a.example', api_token: 'tokA', name: 'A' }, + ]); + const huge = Array.from({ length: 5005 }, (_, i) => ({ + id: i + 1, + name: `p${i}`, + node_identity: '', + replicated_from_control: 0, + created_at: 1, + updated_at: 1, + })); + mockGetLocalScanPolicies.mockReturnValue(huge); + await FleetSyncService.getInstance().pushResource('scan_policies'); + const body = mockAxiosPost.mock.calls[0][1]; + expect(body.rows).toHaveLength(5000); + expect(mockDispatchAlert).toHaveBeenCalledWith( + 'warning', + 'system', + expect.stringContaining('truncated'), + ); + }); + + it('throttles repeat truncation alerts via fleet_sync_truncation_alert_at watermark', async () => { + mockGetNodes.mockReturnValue([ + { id: 2, type: 'remote', api_url: 'https://a.example', api_token: 'tokA', name: 'A' }, + ]); + const huge = Array.from({ length: 5005 }, (_, i) => ({ + id: i + 1, + name: `p${i}`, + node_identity: '', + replicated_from_control: 0, + created_at: 1, + updated_at: 1, + })); + mockGetLocalScanPolicies.mockReturnValue(huge); + // Simulate that an alert was emitted 1 minute ago (well within the cooldown). + mockGetSystemState.mockImplementation((key: string) => { + if (key === 'fleet_sync_truncation_alert_at:scan_policies') return String(Date.now() - 60_000); + return null; + }); + await FleetSyncService.getInstance().pushResource('scan_policies'); + expect(mockDispatchAlert).not.toHaveBeenCalled(); + }); +}); + +describe('FleetSyncService stale-push handling', () => { + it('does not record a fleet sync failure for STALE_SYNC_PUSH 409 responses', async () => { + mockGetNodes.mockReturnValue([ + { id: 2, type: 'remote', api_url: 'https://stale.example', api_token: 'tok', name: 'stale' }, + ]); + mockGetLocalScanPolicies.mockReturnValue([]); + mockAxiosPost.mockImplementation(async () => { + const { AxiosError } = await import('axios'); + const err = new AxiosError('Request failed with status code 409'); + (err as unknown as { response: unknown }).response = { + status: 409, + statusText: 'Conflict', + data: { error: 'stale', code: 'STALE_SYNC_PUSH' }, + }; + throw err; + }); + await FleetSyncService.getInstance().pushResource('scan_policies'); + expect(mockRecordFleetSyncFailure).not.toHaveBeenCalled(); + expect(mockRecordFleetSyncSuccess).not.toHaveBeenCalled(); + }); + + it('still records fleet sync failure for non-STALE 409 responses', async () => { + mockGetNodes.mockReturnValue([ + { id: 2, type: 'remote', api_url: 'https://other.example', api_token: 'tok', name: 'other' }, + ]); + mockGetLocalScanPolicies.mockReturnValue([]); + mockAxiosPost.mockImplementation(async () => { + const { AxiosError } = await import('axios'); + const err = new AxiosError('Request failed with status code 409'); + (err as unknown as { response: unknown }).response = { + status: 409, + statusText: 'Conflict', + data: { error: 'something else', code: 'CONTROL_IDENTITY_MISMATCH' }, + }; + throw err; + }); + await FleetSyncService.getInstance().pushResource('scan_policies'); + expect(mockRecordFleetSyncFailure).toHaveBeenCalled(); + }); +}); + +describe('FleetSyncService.applyIncomingSync transactional', () => { + it('runs inside DatabaseService.transaction so apply and watermark commit atomically', () => { + FleetSyncService.getInstance().applyIncomingSync( + 'scan_policies', + [], + 'https://me.example', + 1_700_000_000_000, + ); + expect(mockTransaction).toHaveBeenCalledTimes(1); + expect(mockSetSystemState).toHaveBeenCalledWith('fleet_role', 'replica'); + expect(mockSetSystemState).toHaveBeenCalledWith('fleet_self_identity', 'https://me.example'); + expect(mockSetSystemState).toHaveBeenCalledWith('received_pushed_at:scan_policies', '1700000000000'); + }); + + it('omits the watermark write when pushedAt is undefined (legacy back-compat)', () => { + FleetSyncService.getInstance().applyIncomingSync('scan_policies', [], 'https://me.example'); + const watermarkWrites = mockSetSystemState.mock.calls.filter( + (c) => typeof c[0] === 'string' && c[0].startsWith('received_pushed_at:'), + ); + expect(watermarkWrites).toHaveLength(0); + }); +}); diff --git a/backend/src/middleware/jsonParser.ts b/backend/src/middleware/jsonParser.ts index ab4ff030..7154cffb 100644 --- a/backend/src/middleware/jsonParser.ts +++ b/backend/src/middleware/jsonParser.ts @@ -1,6 +1,7 @@ import express, { type Request, type Response, type NextFunction, type RequestHandler } from 'express'; import { NodeRegistry } from '../services/NodeRegistry'; import { isProxyExemptPath } from '../helpers/proxyExemptPaths'; +import { SYNC_BODY_LIMIT, SYNC_ERROR_CODES, SYNC_PATH_PREFIX } from '../services/fleetSyncConstants'; // JSON body parser that also captures the raw bytes for HMAC verification. // `rawBody` is part of the Express.Request augmentation (see types/express.ts); @@ -12,6 +13,16 @@ const jsonParser = express.json({ }, }); +// Larger-limit parser for the fleet sync receive endpoint. A control instance +// can push up to MAX_SYNC_ROWS rows in a single payload; the default 100 KB +// limit is too tight for that. +const fleetSyncJsonParser = express.json({ + limit: SYNC_BODY_LIMIT, + verify: (req, _res, buf) => { + (req as unknown as Request).rawBody = buf; + }, +}); + /** * Parse JSON on local requests but preserve the raw stream for remote proxy * forwarding. @@ -33,5 +44,23 @@ export const conditionalJsonParser: RequestHandler = (req: Request, res: Respons return; } } + // Fleet sync receive endpoint accepts larger payloads (up to MAX_SYNC_ROWS). + // The 100 KB default would 413 long before the row-count check runs. Translate + // body-parser's PayloadTooLargeError into a structured response with a + // sync-specific code so the control's retry logic can distinguish it from + // generic 413s. + if (req.path.startsWith(SYNC_PATH_PREFIX)) { + fleetSyncJsonParser(req, res, (err?: unknown) => { + if (err && (err as { type?: string })?.type === 'entity.too.large') { + res.status(413).json({ + error: 'Sync payload too large. Reduce policy or suppression count and retry.', + code: SYNC_ERROR_CODES.payloadTooLarge, + }); + return; + } + next(err); + }); + return; + } jsonParser(req, res, next); }; diff --git a/backend/src/routes/fleet.ts b/backend/src/routes/fleet.ts index 65e2f336..2b54f49d 100644 --- a/backend/src/routes/fleet.ts +++ b/backend/src/routes/fleet.ts @@ -4,7 +4,8 @@ import semver from 'semver'; import si from 'systeminformation'; import type Dockerode from 'dockerode'; import { DatabaseService, type Node } from '../services/DatabaseService'; -import { FleetSyncService } from '../services/FleetSyncService'; +import { FleetSyncService, StaleSyncPushError } from '../services/FleetSyncService'; +import { MAX_SYNC_ROWS, SYNC_ERROR_CODES } from '../services/fleetSyncConstants'; import { FleetUpdateTrackerService } from '../services/FleetUpdateTrackerService'; import { NodeRegistry } from '../services/NodeRegistry'; import DockerController from '../services/DockerController'; @@ -37,7 +38,6 @@ const UPDATE_TIMEOUT_MS = 5 * 60 * 1000; // 5 minutes const UPDATE_TIMEOUT_MSG = 'Node did not come back online within 5 minutes.'; const EARLY_FAIL_MS = 180 * 1000; // 3 minutes before declaring a probable pull failure -const MAX_SYNC_ROWS = 5000; const CVE_ID_RE = /^(CVE-\d{4}-\d{4,}|GHSA-[\w-]{14,})$/; const isIntFlag = (v: unknown): v is 0 | 1 => v === 0 || v === 1; @@ -298,6 +298,11 @@ fleetRouter.get('/role', authMiddleware, (req: Request, res: Response): void => // Receive a full replacement of a replicated resource from the control. // Restricted to node_proxy Bearer tokens so only a sibling Sencho can push. +// +// No requirePaid here: the control instance has already enforced its tier +// before issuing the push. The replica trusts a valid node_proxy bearer +// signed against THIS instance's secret and applies the payload regardless +// of the replica's own tier. fleetRouter.post('/sync/:resource', authMiddleware, (req: Request, res: Response): void => { if (!requireNodeProxy(req, res)) return; const resource = req.params.resource; @@ -308,6 +313,14 @@ fleetRouter.post('/sync/:resource', authMiddleware, (req: Request, res: Response const body = req.body ?? {}; const rows = Array.isArray(body.rows) ? body.rows : null; const targetIdentity = typeof body.targetIdentity === 'string' ? body.targetIdentity : ''; + // pushedAt is optional for back-compat with older controls that predate the + // versioning protocol. When present and strictly older than the most recent + // applied push for this resource, reject with 409 STALE_SYNC_PUSH so the + // control's retry logic can fall back to the next write. Negative or zero + // values are treated as absent: the sender always uses Date.now(). + const pushedAt = typeof body.pushedAt === 'number' && Number.isFinite(body.pushedAt) && body.pushedAt > 0 + ? body.pushedAt + : null; if (!rows) { res.status(400).json({ error: 'rows array is required' }); return; @@ -325,9 +338,21 @@ fleetRouter.post('/sync/:resource', authMiddleware, (req: Request, res: Response } } try { - FleetSyncService.getInstance().applyIncomingSync(resource, rows, targetIdentity); + FleetSyncService.getInstance().applyIncomingSync( + resource, + rows, + targetIdentity, + pushedAt ?? undefined, + ); res.json({ success: true, applied: rows.length }); } catch (error) { + if (error instanceof StaleSyncPushError) { + res.status(409).json({ + error: error.message, + code: SYNC_ERROR_CODES.staleSyncPush, + }); + return; + } console.error('[FleetSync] Failed to apply incoming sync:', error); res.status(500).json({ error: 'Failed to apply sync' }); } diff --git a/backend/src/services/DatabaseService.ts b/backend/src/services/DatabaseService.ts index ab786ace..700c6643 100644 --- a/backend/src/services/DatabaseService.ts +++ b/backend/src/services/DatabaseService.ts @@ -1693,6 +1693,19 @@ export class DatabaseService { this.db.prepare('INSERT OR REPLACE INTO system_state (key, value) VALUES (?, ?)').run(key, value); } + /** + * Run `fn` inside a single SQLite transaction. better-sqlite3 promotes a + * nested call to a SAVEPOINT, so callers can compose this with methods + * that already wrap their own writes in `this.db.transaction(...)`. + * + * Used by FleetSync receive to keep the row replacement and the + * received_pushed_at watermark write atomic. If either step fails, both + * roll back. + */ + public transaction(fn: () => T): T { + return this.db.transaction(fn)(); + } + // --- Stack Alerts --- public getStackAlerts(stackName?: string): StackAlert[] { @@ -3501,6 +3514,17 @@ export class DatabaseService { .all() as ScanPolicy[]; } + /** + * Local-only scan policies (created on this instance, not replicated from a + * control). Used by the fleet sync sender so it never re-replicates rows + * that came from a control in the first place. + */ + public getLocalScanPolicies(): ScanPolicy[] { + return this.db + .prepare('SELECT * FROM scan_policies WHERE replicated_from_control = 0 ORDER BY created_at DESC') + .all() as ScanPolicy[]; + } + public getScanPolicy(id: number): ScanPolicy | null { return ( (this.db @@ -3620,6 +3644,11 @@ export class DatabaseService { stackName: string | null, selfIdentity: string, ): ScanPolicy | null { + // Filter on node_id at SQL: rows are eligible when fleet-wide + // (node_id IS NULL) or when locally scoped to this node (node_id = ?). + // Replicated rows always insert node_id = NULL (see + // replaceReplicatedScanPolicies), so identity scoping for replicated + // rows is enforced in `matchesIdentity` below, not in SQL. const policies = this.db .prepare( 'SELECT * FROM scan_policies WHERE enabled = 1 AND (node_id IS NULL OR node_id = ?)', @@ -3651,7 +3680,11 @@ export class DatabaseService { if (!aNode && bNode) return 1; if (a.stack_pattern && !b.stack_pattern) return -1; if (!a.stack_pattern && b.stack_pattern) return 1; - return 0; + // Deterministic tiebreaker: lowest id wins. Two rows in the same + // scope class (e.g. both fleet-wide stack-wildcard) must resolve + // to the same policy on every replica, regardless of SQLite row + // iteration order. + return a.id - b.id; }); return scoped[0]; } @@ -3749,6 +3782,13 @@ export class DatabaseService { .all() as CveSuppression[]; } + /** Local-only CVE suppressions; mirrors `getLocalScanPolicies`. */ + public getLocalCveSuppressions(): CveSuppression[] { + return this.db + .prepare('SELECT * FROM cve_suppressions WHERE replicated_from_control = 0 ORDER BY cve_id, pkg_name') + .all() as CveSuppression[]; + } + public getCveSuppression(id: number): CveSuppression | null { return ( (this.db.prepare('SELECT * FROM cve_suppressions WHERE id = ?') diff --git a/backend/src/services/FleetSyncService.ts b/backend/src/services/FleetSyncService.ts index 80abbc8c..9256a453 100644 --- a/backend/src/services/FleetSyncService.ts +++ b/backend/src/services/FleetSyncService.ts @@ -1,8 +1,17 @@ import axios, { AxiosError } from 'axios'; import { CveSuppression, DatabaseService, Node, ScanPolicy } from './DatabaseService'; import { NodeRegistry } from './NodeRegistry'; +import { NotificationService } from './NotificationService'; +import { isDebugEnabled } from '../utils/debug'; +import { + FleetResource, + MAX_SYNC_ROWS, + SYNC_ERROR_CODES, + SYNC_STATE_KEYS, + TRUNCATION_ALERT_COOLDOWN_MS, +} from './fleetSyncConstants'; -export type FleetResource = 'scan_policies' | 'cve_suppressions'; +export type { FleetResource }; export const FLEET_RESOURCES: readonly FleetResource[] = ['scan_policies', 'cve_suppressions']; @@ -14,19 +23,63 @@ export type FleetRole = 'control' | 'replica'; export const LOCAL_IDENTITY_SENTINEL = 'local'; +/** + * Wire-format payload pushed to a remote's POST /api/fleet/sync/:resource. + * The receiver parses individual fields with explicit type checks, so this + * stays internal to the sender. Both sides tolerate absent `pushedAt` and + * `controlIdentity` for back-compat with controls that predate the + * versioning protocol. + */ +interface FleetSyncPayload { + rows: unknown[]; + pushedAt: number; + targetIdentity: string; + controlIdentity: string; +} + +/** + * Thrown by `applyIncomingSync` when the incoming `pushedAt` is strictly older + * than the receiver's last-applied watermark for the same resource. The route + * handler catches this specifically and returns 409 STALE_SYNC_PUSH; any other + * error becomes a 500. + */ +export class StaleSyncPushError extends Error { + constructor(public readonly previous: number, public readonly incoming: number) { + super(`Stale sync push: pushedAt=${incoming} is older than last applied=${previous}`); + this.name = 'StaleSyncPushError'; + } +} + /** * FleetSyncService replicates security configuration from a control Sencho * instance to every managed remote node. Security rules live on the control's * SQLite database; each write triggers a push of the full table to every * remote that has an api_url and api_token configured. * - * Push failures for a specific remote are logged and recorded on the - * fleet_sync_status table so the UI and future retry logic can see stale - * nodes. + * Per-node concurrency: pushes to the same remote are serialized via an + * in-memory mutex map. Sencho runs as a single Node.js process per instance, + * so process-local serialization is correct for the supported topology. + * + * Push failures are recorded on `fleet_sync_status` for the UI and the retry + * service. STALE_SYNC_PUSH 409 responses are not recorded — they are an + * expected protocol outcome, not a node health issue. */ export class FleetSyncService { private static instance: FleetSyncService; + /** Tail of the most recent push promise for each node id. */ + private inflightByNode = new Map>(); + + /** + * Strictly-increasing timestamp guard for outgoing pushes. Process-global + * across resources: scan_policies and cve_suppressions share the counter + * because the receiver tracks watermarks per resource via separate + * `received_pushed_at:` keys. + */ + private static lastPushedAt = 0; + + private static warnedMissingIdentity = false; + private constructor() {} public static getInstance(): FleetSyncService { @@ -36,24 +89,21 @@ export class FleetSyncService { return FleetSyncService.instance; } - /** - * Resolve the fleet role for this instance. - * A node becomes a replica the first time it accepts a fleet sync push. - */ public static getRole(): FleetRole { - return DatabaseService.getInstance().getSystemState('fleet_role') === 'replica' ? 'replica' : 'control'; + return DatabaseService.getInstance().getSystemState(SYNC_STATE_KEYS.fleetRole) === 'replica' + ? 'replica' + : 'control'; } /** - * The identity string used when matching scan policies on this instance. - * Control nodes use the LOCAL_IDENTITY_SENTINEL. Replicas use the - * identity they were told during the most recent sync push. If a replica - * is missing its cached identity (e.g. the sync row has been corrupted), - * return the empty string; callers treat this as fleet-wide only and log. + * Identity string used when matching scan policies on this instance. The + * empty string fallback on a replica with corrupted state means + * identity-scoped policies will not apply until the next sync push restores + * the cached identity. */ public static getSelfIdentity(): string { if (FleetSyncService.getRole() === 'replica') { - const cached = DatabaseService.getInstance().getSystemState('fleet_self_identity'); + const cached = DatabaseService.getInstance().getSystemState(SYNC_STATE_KEYS.fleetSelfIdentity); if (!cached) { if (!FleetSyncService.warnedMissingIdentity) { console.warn( @@ -68,13 +118,9 @@ export class FleetSyncService { return LOCAL_IDENTITY_SENTINEL; } - private static warnedMissingIdentity = false; - /** * Map a policy's node_id to a node_identity string. - * - NULL node_id → '' (fleet-wide) - * - Local node → LOCAL_IDENTITY_SENTINEL - * - Remote node → the node's api_url + * NULL → '' (fleet-wide); local → LOCAL_IDENTITY_SENTINEL; remote → api_url. */ public static resolveIdentityForNodeId(nodeId: number | null | undefined): string { if (nodeId == null) return ''; @@ -85,56 +131,57 @@ export class FleetSyncService { } /** - * Push the current state of a resource to every remote node. - * Failures are recorded but do not bubble up to the caller. + * Returns the empty string. The receiver treats empty `controlIdentity` + * as legacy and accepts. A future change will derive a stable fingerprint + * from a system_state key for control-anchor enforcement. */ + public static getControlIdentity(): string { + return ''; + } + + private nextPushedAt(): number { + const now = Date.now(); + const next = now > FleetSyncService.lastPushedAt ? now : FleetSyncService.lastPushedAt + 1; + FleetSyncService.lastPushedAt = next; + return next; + } + + /** Push the current state of a resource to every remote node. */ public async pushResource(resource: FleetResource): Promise { if (FleetSyncService.getRole() === 'replica') { - // Replicas never push; they only receive. + if (isDebugEnabled()) { + console.debug(`[FleetSync:debug] Skipping push for ${resource}: this instance is a replica`); + } return; } const db = DatabaseService.getInstance(); const nodes = db.getNodes().filter((n): n is Node & { id: number } => { return n.type === 'remote' && Boolean(n.api_url) && Boolean(n.api_token) && n.id != null; }); - if (nodes.length === 0) return; + if (nodes.length === 0) { + if (isDebugEnabled()) { + console.debug(`[FleetSync:debug] No eligible remote nodes for ${resource} push`); + } + return; + } const rows = this.loadResource(resource); - const pushedAt = Date.now(); + const payload: Omit = { + rows, + pushedAt: this.nextPushedAt(), + controlIdentity: FleetSyncService.getControlIdentity(), + }; - await Promise.all( - nodes.map(async (node) => { - const baseUrl = (node.api_url ?? '').replace(/\/$/, ''); - try { - await axios.post( - `${baseUrl}/api/fleet/sync/${resource}`, - { - rows, - pushedAt, - targetIdentity: node.api_url, - }, - { - headers: { Authorization: `Bearer ${node.api_token}` }, - timeout: 15_000, - }, - ); - db.recordFleetSyncSuccess(node.id, resource); - } catch (err) { - const message = this.formatError(err); - console.warn( - `[FleetSync] Failed to push ${resource} to "${node.name}" (${baseUrl}): ${message}`, - ); - db.recordFleetSyncFailure(node.id, resource, message); - } - }), - ); + if (isDebugEnabled()) { + console.debug( + `[FleetSync:debug] Pushing ${resource} to ${nodes.length} remote(s): rows=${rows.length} pushedAt=${payload.pushedAt}`, + ); + } + + await Promise.all(nodes.map((node) => this.enqueuePushToNode(node, resource, payload))); } - /** - * Fire and forget helper for write handlers. Errors are already logged - * inside pushResource; this swallows any residual rejection so request - * handlers can stay synchronous. - */ + /** Fire-and-forget wrapper for write handlers. Errors are already logged inside. */ public pushResourceAsync(resource: FleetResource): void { this.pushResource(resource).catch((err) => { console.error(`[FleetSync] Unexpected error pushing ${resource}:`, err); @@ -142,59 +189,183 @@ export class FleetSyncService { } /** - * Apply a received sync payload on a replica. - * This promotes the instance to 'replica' mode if not already, caches - * the target identity it was told, and replaces replicated rows atomically. + * Apply a received sync payload on a replica. Runs role flip, identity + * cache, row replacement, watermark write, AND staleness comparison inside + * a single SQLite transaction so a partial-write window cannot leave the + * watermark behind the row state. + * + * Throws `StaleSyncPushError` (rolling back the transaction) when the + * incoming `pushedAt` is older than the persisted watermark; the route + * handler translates that into 409 STALE_SYNC_PUSH. + * + * `pushedAt` is optional for back-compat with legacy controls; when absent, + * the staleness check is skipped and no watermark is written. */ public applyIncomingSync( resource: FleetResource, rows: ScanPolicy[] | Array>, targetIdentity: string, + pushedAt?: number, ): void { const db = DatabaseService.getInstance(); - db.setSystemState('fleet_role', 'replica'); - if (targetIdentity) { - db.setSystemState('fleet_self_identity', targetIdentity); - } - if (resource === 'scan_policies') { - db.replaceReplicatedScanPolicies(rows as ScanPolicy[]); - } else if (resource === 'cve_suppressions') { - db.replaceReplicatedCveSuppressions(rows as Array>); + db.transaction(() => { + if (pushedAt !== undefined && Number.isFinite(pushedAt)) { + const watermarkKey = SYNC_STATE_KEYS.receivedPushedAt(resource); + const previousRaw = db.getSystemState(watermarkKey); + const previous = previousRaw !== null ? Number(previousRaw) : null; + if (previous !== null && Number.isFinite(previous) && pushedAt < previous) { + throw new StaleSyncPushError(previous, pushedAt); + } + db.setSystemState(watermarkKey, String(pushedAt)); + } + db.setSystemState(SYNC_STATE_KEYS.fleetRole, 'replica'); + if (targetIdentity) { + db.setSystemState(SYNC_STATE_KEYS.fleetSelfIdentity, targetIdentity); + } + if (resource === 'scan_policies') { + db.replaceReplicatedScanPolicies(rows as ScanPolicy[]); + } else if (resource === 'cve_suppressions') { + db.replaceReplicatedCveSuppressions(rows as Array>); + } + }); + } + + /** + * Chain a push behind any in-flight push for the same node. Different + * nodes still run in parallel via the outer Promise.all. + */ + private enqueuePushToNode( + node: Node & { id: number }, + resource: FleetResource, + payload: Omit, + ): Promise { + const prev = this.inflightByNode.get(node.id) ?? Promise.resolve(); + // .catch() before .then() so a thrown push does not poison the chain + // for that node id. executePushToNode swallows internally today, but + // the catch is defense-in-depth against future regressions. + const next = prev.catch(() => undefined).then( + () => this.executePushToNode(node, resource, payload), + ); + const tracked = next.finally(() => { + if (this.inflightByNode.get(node.id) === tracked) { + this.inflightByNode.delete(node.id); + } + }); + this.inflightByNode.set(node.id, tracked); + return next; + } + + private async executePushToNode( + node: Node & { id: number }, + resource: FleetResource, + partial: Omit, + ): Promise { + const db = DatabaseService.getInstance(); + const apiUrl = node.api_url ?? ''; + const baseUrl = apiUrl.replace(/\/$/, ''); + const payload: FleetSyncPayload = { ...partial, targetIdentity: apiUrl }; + try { + await axios.post( + `${baseUrl}/api/fleet/sync/${resource}`, + payload, + { + headers: { Authorization: `Bearer ${node.api_token}` }, + timeout: 15_000, + }, + ); + db.recordFleetSyncSuccess(node.id, resource); + if (isDebugEnabled()) { + console.debug( + `[FleetSync:debug] Pushed ${resource} to "${node.name}" (${baseUrl}) ok: rows=${payload.rows.length} pushedAt=${payload.pushedAt}`, + ); + } + } catch (err) { + // STALE_SYNC_PUSH (409) is an expected protocol outcome: a newer + // push for the same resource has already landed. The node is + // healthy; suppress the failure record so it does not surface as + // an alert in the sync-status panel. + if (err instanceof AxiosError && err.response?.status === 409) { + const data = err.response.data as { code?: string } | undefined; + if (data?.code === SYNC_ERROR_CODES.staleSyncPush) { + if (isDebugEnabled()) { + console.debug( + `[FleetSync:debug] Stale push to "${node.name}" (${baseUrl}) for ${resource}; receiver already has a newer state.`, + ); + } + return; + } + } + const message = this.formatError(err); + console.warn( + `[FleetSync] Failed to push ${resource} to "${node.name}" (${baseUrl}): ${message}`, + ); + db.recordFleetSyncFailure(node.id, resource, message); } } private loadResource(resource: FleetResource): unknown[] { const db = DatabaseService.getInstance(); + let rows: unknown[]; if (resource === 'scan_policies') { - return db - .getScanPolicies() - .filter((p) => p.replicated_from_control === 0) - .map((p) => ({ - name: p.name, - node_identity: p.node_identity, - stack_pattern: p.stack_pattern, - max_severity: p.max_severity, - block_on_deploy: p.block_on_deploy, - enabled: p.enabled, - created_at: p.created_at, - updated_at: p.updated_at, - })); + rows = db.getLocalScanPolicies().map((p) => ({ + name: p.name, + node_identity: p.node_identity, + stack_pattern: p.stack_pattern, + max_severity: p.max_severity, + block_on_deploy: p.block_on_deploy, + enabled: p.enabled, + created_at: p.created_at, + updated_at: p.updated_at, + })); + } else if (resource === 'cve_suppressions') { + rows = db.getLocalCveSuppressions().map((s) => ({ + cve_id: s.cve_id, + pkg_name: s.pkg_name, + image_pattern: s.image_pattern, + reason: s.reason, + created_by: s.created_by, + created_at: s.created_at, + expires_at: s.expires_at, + })); + } else { + return []; } - if (resource === 'cve_suppressions') { - return db - .getCveSuppressions() - .filter((s) => s.replicated_from_control === 0) - .map((s) => ({ - cve_id: s.cve_id, - pkg_name: s.pkg_name, - image_pattern: s.image_pattern, - reason: s.reason, - created_by: s.created_by, - created_at: s.created_at, - expires_at: s.expires_at, - })); + + if (rows.length > MAX_SYNC_ROWS) { + const dropped = rows.length - MAX_SYNC_ROWS; + console.warn( + `[FleetSync] ${resource} has ${rows.length} local rows; truncating to ${MAX_SYNC_ROWS} for sync (dropped=${dropped})`, + ); + this.maybeNotifyTruncation(resource, dropped); + rows = rows.slice(0, MAX_SYNC_ROWS); } - return []; + return rows; + } + + /** + * Dispatch a truncation warning at most once per cooldown window. Without + * this throttle, a configuration that exceeds the row cap would generate + * one alert per write and flood the operator's notification stream. + */ + private maybeNotifyTruncation(resource: FleetResource, dropped: number): void { + const db = DatabaseService.getInstance(); + const stateKey = SYNC_STATE_KEYS.truncationAlertAt(resource); + const lastRaw = db.getSystemState(stateKey); + const last = lastRaw !== null ? Number(lastRaw) : 0; + const now = Date.now(); + if (Number.isFinite(last) && now - last < TRUNCATION_ALERT_COOLDOWN_MS) { + return; + } + db.setSystemState(stateKey, String(now)); + void NotificationService.getInstance() + .dispatchAlert( + 'warning', + 'system', + `Fleet sync truncated ${resource} to ${MAX_SYNC_ROWS} rows (${dropped} not replicated). Reduce the local set or contact support.`, + ) + .catch((err) => { + console.warn('[FleetSync] Failed to dispatch truncation alert:', err); + }); } private formatError(err: unknown): string { diff --git a/backend/src/services/fleetSyncConstants.ts b/backend/src/services/fleetSyncConstants.ts new file mode 100644 index 00000000..0a275a3b --- /dev/null +++ b/backend/src/services/fleetSyncConstants.ts @@ -0,0 +1,70 @@ +/** + * Shared constants for the Fleet Sync wire protocol. + * + * The control instance pushes scan_policies and cve_suppressions to remote + * nodes via POST /api/fleet/sync/:resource. The payload schema is defined + * here so both the sender (FleetSyncService) and the receiver (routes/fleet) + * agree on limits and field names. + */ + +/** + * Maximum number of rows accepted per sync push. + * + * Enforced on both ends: + * - Sender: `FleetSyncService.loadResource` truncates at this cap and + * emits a warning notification when it triggers. Operators with + * >5000 policies on the control are exceedingly rare; truncating is + * safer than failing every push. + * - Receiver: `POST /api/fleet/sync/:resource` rejects payloads above + * this cap with HTTP 413. + */ +export const MAX_SYNC_ROWS = 5000; + +/** + * Maximum body size for POST /api/fleet/sync/:resource. Sized to comfortably + * fit MAX_SYNC_ROWS rows of either resource at the per-field length caps + * enforced by the row validators (~1 KB per row worst-case = ~5 MB). + * + * The global JSON body limit stays at 100 KB; only this one route allows + * larger bodies. See middleware/jsonParser.ts for the dispatch logic. + */ +export const SYNC_BODY_LIMIT = '5mb'; + +/** + * Path prefix that the JSON body parser uses to dispatch to the larger + * limit. Kept as a constant so any path change updates both the parser + * and the route mount. + * + * CAUTION: prefix match, not exact route. Any new route under /api/fleet/sync/ + * inherits the elevated body limit. If a future route under this prefix should + * keep the standard 100 KB cap, narrow the dispatch in middleware/jsonParser.ts + * to method+exact-path instead of prefix. + */ +export const SYNC_PATH_PREFIX = '/api/fleet/sync/'; + +/** How long to suppress repeat truncation alerts after one fires. */ +export const TRUNCATION_ALERT_COOLDOWN_MS = 6 * 60 * 60 * 1000; + +/** + * Resource enum kept here so the state-key helpers below can type-check + * their arguments without a cycle through FleetSyncService. The ordering + * below mirrors `FLEET_RESOURCES` in FleetSyncService. + */ +export type FleetResource = 'scan_policies' | 'cve_suppressions'; + +/** + * `system_state` keys read or written by Fleet Sync. Centralized so a typo + * cannot silently bypass a stale-push check or a cooldown gate. + */ +export const SYNC_STATE_KEYS = { + fleetRole: 'fleet_role', + fleetSelfIdentity: 'fleet_self_identity', + receivedPushedAt: (resource: FleetResource): string => `received_pushed_at:${resource}`, + truncationAlertAt: (resource: FleetResource): string => `fleet_sync_truncation_alert_at:${resource}`, +} as const; + +/** Structured error codes returned by the receive endpoint. */ +export const SYNC_ERROR_CODES = { + staleSyncPush: 'STALE_SYNC_PUSH', + payloadTooLarge: 'SYNC_PAYLOAD_TOO_LARGE', +} as const;