feat(fleet): replicate scan policies across managed nodes (#649)

Scan policies now propagate from the control Sencho instance to every registered remote. The control is the source of truth; replicas render rules read-only with a managed-by-control banner. Pushes fire on every policy write, record per-node success and failure on a new fleet_sync_status table, and use node_proxy Bearer tokens exclusively so only sibling Senchos can apply incoming sync payloads. Policy scope now travels as a string identity (api_url or a local sentinel) so node-scoped rules evaluate correctly on each target.
2026-08-27 18:57:09 +00:00 · 2026-04-16 23:57:08 -04:00
parent 8ee0c0c476
commit 708d15b2b3
10 changed files with 888 additions and 35 deletions
@@ -308,14 +308,24 @@ export interface ScanPolicy {
    id: number;
    name: string;
    node_id: number | null;
+    node_identity: string;
    stack_pattern: string | null;
    max_severity: VulnSeverity;
    block_on_deploy: number;
    enabled: number;
+    replicated_from_control: number;
    created_at: number;
    updated_at: number;
 }

+export interface FleetSyncStatus {
+    node_id: number;
+    resource: string;
+    last_success_at: number | null;
+    last_failure_at: number | null;
+    last_error: string | null;
+}
+
 export interface ScanSummary {
    image_ref: string;
    highest_severity: VulnSeverity | null;
@@ -352,6 +362,7 @@ export class DatabaseService {
        this.migrateRegistries();
        this.migrateRoleAssignments();
        this.migrateNotificationRoutes();
+        this.migrateScanPolicyFleetColumns();
    }

    public static getInstance(): DatabaseService {
@@ -613,6 +624,15 @@ export class DatabaseService {
        updated_at INTEGER NOT NULL
      );

+      CREATE TABLE IF NOT EXISTS fleet_sync_status (
+        node_id INTEGER NOT NULL,
+        resource TEXT NOT NULL,
+        last_success_at INTEGER,
+        last_failure_at INTEGER,
+        last_error TEXT,
+        PRIMARY KEY (node_id, resource)
+      );
+
      CREATE TABLE IF NOT EXISTS stack_labels (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        node_id INTEGER NOT NULL DEFAULT 0,
@@ -878,6 +898,18 @@ export class DatabaseService {
        try { this.db.prepare('ALTER TABLE notification_history ADD COLUMN dispatch_error TEXT').run(); } catch { /* already exists */ }
    }

+    private migrateScanPolicyFleetColumns(): void {
+        const tryAddColumn = (table: string, col: string, def: string) => {
+            try {
+                this.db.prepare(`ALTER TABLE ${table} ADD COLUMN ${col} ${def}`).run();
+            } catch {
+                /* column already present */
+            }
+        };
+        tryAddColumn('scan_policies', 'node_identity', "TEXT NOT NULL DEFAULT ''");
+        tryAddColumn('scan_policies', 'replicated_from_control', 'INTEGER NOT NULL DEFAULT 0');
+    }
+
    // --- Agents ---

    public getAgents(): Agent[] {
@@ -2309,20 +2341,29 @@ export class DatabaseService {
        const now = Date.now();
        const result = this.db
            .prepare(
-                `INSERT INTO scan_policies (name, node_id, stack_pattern, max_severity, block_on_deploy, enabled, created_at, updated_at)
-                 VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
+                `INSERT INTO scan_policies (name, node_id, node_identity, stack_pattern, max_severity, block_on_deploy, enabled, replicated_from_control, created_at, updated_at)
+                 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
            )
            .run(
                policy.name,
                policy.node_id,
+                policy.node_identity ?? '',
                policy.stack_pattern,
                policy.max_severity,
                policy.block_on_deploy,
                policy.enabled,
+                policy.replicated_from_control ?? 0,
                now,
                now,
            );
-        return { ...policy, id: result.lastInsertRowid as number, created_at: now, updated_at: now };
+        return {
+            ...policy,
+            node_identity: policy.node_identity ?? '',
+            replicated_from_control: policy.replicated_from_control ?? 0,
+            id: result.lastInsertRowid as number,
+            created_at: now,
+            updated_at: now,
+        };
    }

    public updateScanPolicy(
@@ -2332,8 +2373,8 @@ export class DatabaseService {
        const existing = this.getScanPolicy(id);
        if (!existing) return null;
        const ALLOWED_COLUMNS = new Set([
-            'name', 'node_id', 'stack_pattern', 'max_severity',
-            'block_on_deploy', 'enabled',
+            'name', 'node_id', 'node_identity', 'stack_pattern', 'max_severity',
+            'block_on_deploy', 'enabled', 'replicated_from_control',
        ]);
        const fields: string[] = [];
        const values: unknown[] = [];
@@ -2356,9 +2397,41 @@ export class DatabaseService {
        this.db.prepare('DELETE FROM scan_policies WHERE id = ?').run(id);
    }

+    /**
+     * Replace all policies that were replicated from a control node with the
+     * provided rows in a single transaction. Local-only policies (created on
+     * this instance directly) are left untouched.
+     */
+    public replaceReplicatedScanPolicies(rows: ScanPolicy[]): void {
+        const now = Date.now();
+        const deleteStmt = this.db.prepare('DELETE FROM scan_policies WHERE replicated_from_control = 1');
+        const insertStmt = this.db.prepare(
+            `INSERT INTO scan_policies (name, node_id, node_identity, stack_pattern, max_severity, block_on_deploy, enabled, replicated_from_control, created_at, updated_at)
+             VALUES (?, ?, ?, ?, ?, ?, ?, 1, ?, ?)`,
+        );
+        const txn = this.db.transaction((policies: ScanPolicy[]) => {
+            deleteStmt.run();
+            for (const p of policies) {
+                insertStmt.run(
+                    p.name,
+                    null,
+                    p.node_identity ?? '',
+                    p.stack_pattern,
+                    p.max_severity,
+                    p.block_on_deploy,
+                    p.enabled,
+                    p.created_at ?? now,
+                    p.updated_at ?? now,
+                );
+            }
+        });
+        txn(rows);
+    }
+
    public getMatchingPolicy(
        nodeId: number,
        stackName: string | null,
+        selfIdentity: string,
    ): ScanPolicy | null {
        const policies = this.db
            .prepare(
@@ -2373,11 +2446,22 @@ export class DatabaseService {
            );
            return regex.test(stackName);
        };
-        const scoped = policies.filter((p) => matchesStack(p.stack_pattern));
+        const matchesIdentity = (p: ScanPolicy): boolean => {
+            // Locally created policies (never replicated) apply based on node_id logic already filtered.
+            if (p.replicated_from_control === 0) return true;
+            // Replicated policies without a specific identity are fleet-wide.
+            if (!p.node_identity) return true;
+            // Identity-scoped replicated policies only apply to their target instance.
+            return p.node_identity === selfIdentity;
+        };
+        const scoped = policies.filter((p) => matchesStack(p.stack_pattern) && matchesIdentity(p));
        if (scoped.length === 0) return null;
+        const isNodeScoped = (p: ScanPolicy): boolean => Boolean(p.node_id) || Boolean(p.node_identity);
        scoped.sort((a, b) => {
-            if (a.node_id && !b.node_id) return -1;
-            if (!a.node_id && b.node_id) return 1;
+            const aNode = isNodeScoped(a);
+            const bNode = isNodeScoped(b);
+            if (aNode && !bNode) return -1;
+            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;
@@ -2385,6 +2469,52 @@ export class DatabaseService {
        return scoped[0];
    }

+    // --- Fleet Sync Status ---
+
+    public getFleetSyncStatuses(): FleetSyncStatus[] {
+        return this.db
+            .prepare('SELECT * FROM fleet_sync_status ORDER BY node_id, resource')
+            .all() as FleetSyncStatus[];
+    }
+
+    public recordFleetSyncSuccess(nodeId: number, resource: string): void {
+        const now = Date.now();
+        this.db
+            .prepare(
+                `INSERT INTO fleet_sync_status (node_id, resource, last_success_at, last_failure_at, last_error)
+                 VALUES (?, ?, ?, NULL, NULL)
+                 ON CONFLICT(node_id, resource) DO UPDATE SET
+                   last_success_at = excluded.last_success_at,
+                   last_error = NULL`,
+            )
+            .run(nodeId, resource, now);
+    }
+
+    public recordFleetSyncFailure(nodeId: number, resource: string, error: string): void {
+        const now = Date.now();
+        this.db
+            .prepare(
+                `INSERT INTO fleet_sync_status (node_id, resource, last_failure_at, last_error)
+                 VALUES (?, ?, ?, ?)
+                 ON CONFLICT(node_id, resource) DO UPDATE SET
+                   last_failure_at = excluded.last_failure_at,
+                   last_error = excluded.last_error`,
+            )
+            .run(nodeId, resource, now, error);
+    }
+
+    public getFailedSyncTargets(resource: string, maxAgeMs: number): FleetSyncStatus[] {
+        const cutoff = Date.now() - maxAgeMs;
+        return this.db
+            .prepare(
+                `SELECT * FROM fleet_sync_status
+                 WHERE resource = ?
+                   AND (last_failure_at IS NOT NULL AND last_failure_at > ?)
+                   AND (last_success_at IS NULL OR last_success_at < last_failure_at)`,
+            )
+            .all(resource, cutoff) as FleetSyncStatus[];
+    }
+
    // --- Stack Labels ---

    public getLabels(nodeId: number): Label[] {
@@ -0,0 +1,187 @@
+import axios, { AxiosError } from 'axios';
+import { DatabaseService, Node, ScanPolicy } from './DatabaseService';
+import { NodeRegistry } from './NodeRegistry';
+
+export type FleetResource = 'scan_policies';
+
+export type FleetRole = 'control' | 'replica';
+
+export const LOCAL_IDENTITY_SENTINEL = 'local';
+
+/**
+ * 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.
+ */
+export class FleetSyncService {
+    private static instance: FleetSyncService;
+
+    private constructor() {}
+
+    public static getInstance(): FleetSyncService {
+        if (!FleetSyncService.instance) {
+            FleetSyncService.instance = new 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';
+    }
+
+    /**
+     * 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.
+     */
+    public static getSelfIdentity(): string {
+        if (FleetSyncService.getRole() === 'replica') {
+            const cached = DatabaseService.getInstance().getSystemState('fleet_self_identity');
+            if (!cached) {
+                if (!FleetSyncService.warnedMissingIdentity) {
+                    console.warn(
+                        '[FleetSync] Replica has no cached self-identity. Identity-scoped policies will not apply until the next sync push.',
+                    );
+                    FleetSyncService.warnedMissingIdentity = true;
+                }
+                return '';
+            }
+            return cached;
+        }
+        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
+     */
+    public static resolveIdentityForNodeId(nodeId: number | null | undefined): string {
+        if (nodeId == null) return '';
+        const node = NodeRegistry.getInstance().getNode(nodeId);
+        if (!node) return '';
+        if (node.type === 'remote' && node.api_url) return node.api_url;
+        return LOCAL_IDENTITY_SENTINEL;
+    }
+
+    /**
+     * Push the current state of a resource to every remote node.
+     * Failures are recorded but do not bubble up to the caller.
+     */
+    public async pushResource(resource: FleetResource): Promise<void> {
+        if (FleetSyncService.getRole() === 'replica') {
+            // Replicas never push; they only receive.
+            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;
+
+        const rows = this.loadResource(resource);
+        const pushedAt = Date.now();
+
+        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);
+                }
+            }),
+        );
+    }
+
+    /**
+     * Fire and forget helper for write handlers. Errors are already logged
+     * inside pushResource; this swallows any residual rejection so request
+     * handlers can stay synchronous.
+     */
+    public pushResourceAsync(resource: FleetResource): void {
+        this.pushResource(resource).catch((err) => {
+            console.error(`[FleetSync] Unexpected error pushing ${resource}:`, err);
+        });
+    }
+
+    /**
+     * 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.
+     */
+    public applyIncomingSync(resource: FleetResource, rows: ScanPolicy[], targetIdentity: string): 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);
+        }
+    }
+
+    private loadResource(resource: FleetResource): unknown[] {
+        const db = DatabaseService.getInstance();
+        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,
+                }));
+        }
+        return [];
+    }
+
+    private formatError(err: unknown): string {
+        if (err instanceof AxiosError) {
+            if (err.response) {
+                const data = err.response.data;
+                const detail = typeof data === 'object' && data && 'error' in data
+                    ? String((data as { error: unknown }).error)
+                    : err.response.statusText;
+                return `HTTP ${err.response.status}: ${detail}`;
+            }
+            return err.message;
+        }
+        return err instanceof Error ? err.message : String(err);
+    }
+}