feat(enterprise): add SIEM webhook forwarding with circuit breaker

apps/api/src/jobs/siem-forward.ts

@@ -0,0 +1,120 @@
+/**
+ * SIEM forwarding job.
+ *
+ * Reads unforwarded audit log entries and delivers them to the configured
+ * SIEM endpoint via the webhook delivery module. Implements a circuit
+ * breaker (5 consecutive failures = disabled until manual reset) and a
+ * cursor-based approach to avoid re-sending events.
+ *
+ * State keys in the settings table:
+ *   - siem_last_forwarded_id: cursor (last successfully forwarded audit log ID)
+ *   - siem_consecutive_failures: circuit breaker counter
+ */
+import { asc, eq, gt } from "drizzle-orm";
+import { env } from "../config.js";
+import { db, schema } from "../db/index.js";
+import { decrypt, isEncrypted } from "../lib/encryption.js";
+import { deliverWebhook } from "../lib/webhook-delivery.js";
+import { readSiemConfig } from "../routes/enterprise/siem.js";
+
+const BATCH_LIMIT = 500;
+const CIRCUIT_BREAKER_THRESHOLD = 5;
+const CURSOR_KEY = "siem_last_forwarded_id";
+const FAILURES_KEY = "siem_consecutive_failures";
+
+async function readSettingValue(key: string): Promise<string | null> {
+  const [row] = await db
+    .select({ value: schema.settings.value })
+    .from(schema.settings)
+    .where(eq(schema.settings.key, key));
+  return row?.value ?? null;
+}
+
+async function upsertSetting(key: string, value: string): Promise<void> {
+  const [existing] = await db
+    .select()
+    .from(schema.settings)
+    .where(eq(schema.settings.key, key));
+
+  if (existing) {
+    await db
+      .update(schema.settings)
+      .set({ value, updatedAt: new Date() })
+      .where(eq(schema.settings.key, key));
+  } else {
+    await db.insert(schema.settings).values({ key, value });
+  }
+}
+
+export async function runSiemForward(): Promise<{ forwarded: number } | void> {
+  // 1. Read SIEM config
+  const config = await readSiemConfig();
+  if (!config || !config.enabled || !config.webhookUrl) {
+    return;
+  }
+
+  // 2. Circuit breaker check
+  const failureCountStr = await readSettingValue(FAILURES_KEY);
+  const failureCount = failureCountStr ? parseInt(failureCountStr, 10) : 0;
+  if (failureCount >= CIRCUIT_BREAKER_THRESHOLD) {
+    console.warn(
+      `SIEM forwarding circuit breaker open: ${failureCount} consecutive failures. ` +
+      "Reset siem_consecutive_failures to 0 in settings to re-enable.",
+    );
+    return;
+  }
+
+  // 3. Read cursor
+  const cursor = await readSettingValue(CURSOR_KEY);
+
+  // 4. Query audit_log for new rows
+  const conditions = cursor ? gt(schema.auditLog.id, cursor) : undefined;
+  const rows = await db
+    .select()
+    .from(schema.auditLog)
+    .where(conditions)
+    .orderBy(asc(schema.auditLog.createdAt))
+    .limit(BATCH_LIMIT);
+
+  if (rows.length === 0) {
+    return;
+  }
+
+  // 5. Decrypt auth header if encrypted
+  let authHeader = config.authHeader;
+  if (authHeader && isEncrypted(authHeader) && env.DATA_ENCRYPTION_KEY) {
+    const decrypted = await decrypt(authHeader, env.DATA_ENCRYPTION_KEY);
+    authHeader = decrypted ?? "";
+  }
+
+  // 6. Map rows to SIEM event payload
+  const events = rows.map((row) => ({
+    timestamp: row.createdAt.toISOString(),
+    event: row.action,
+    actorId: row.actorId,
+    actorUsername: row.actorUsername,
+    targetType: row.targetType,
+    targetId: row.targetId,
+    ip: row.ipAddress,
+    details: row.details,
+  }));
+
+  // 7. Deliver via webhook
+  const result = await deliverWebhook(config.webhookUrl, authHeader, events);
+
+  // 8. Update state based on result
+  if (result.success) {
+    const lastId = rows[rows.length - 1].id;
+    await upsertSetting(CURSOR_KEY, lastId);
+    if (failureCount > 0) {
+      await upsertSetting(FAILURES_KEY, "0");
+    }
+    return { forwarded: rows.length };
+  }
+
+  // Failure: increment circuit breaker
+  await upsertSetting(FAILURES_KEY, String(failureCount + 1));
+  console.error(
+    `SIEM forwarding failed (attempt ${failureCount + 1}/${CIRCUIT_BREAKER_THRESHOLD}): ${result.error}`,
+  );
+}

apps/api/src/jobs/system-jobs.ts

@@ -16,11 +16,13 @@ import { db, schema } from "../db/index.js";
 import { getMaxAgeMs } from "../lib/cleanup.js";
 import { deletePrefix, listJobDirs, type ObjectInfo } from "../lib/object-storage.js";
 import { getQueue } from "./queues.js";
+import { runSiemForward } from "./siem-forward.js";
 
 export const SYSTEM_JOBS = {
   storageTtl: "system:storage-ttl",
   sessionPurge: "system:session-purge",
   retention: "system:retention",
+  siemForward: "system:siem-forward",
 } as const;
 
 // -- Scheduling ---------------------------------------------------------------
@@ -39,6 +41,7 @@ export async function scheduleSystemJobs(): Promise<void> {
   }
   await q.upsertJobScheduler(SYSTEM_JOBS.sessionPurge, { every: 60 * 60_000 });
   await q.upsertJobScheduler(SYSTEM_JOBS.retention, { every: 6 * 60 * 60_000 });
+  await q.upsertJobScheduler(SYSTEM_JOBS.siemForward, { every: 30_000 });
 }
 
 /** Enqueue a one-shot system job (e.g. startup cleanup trigger). */
@@ -58,6 +61,8 @@ export async function runSystemJob(job: Job): Promise<unknown> {
       return db.execute(sql`DELETE FROM sessions WHERE expires_at < now()`);
     case SYSTEM_JOBS.retention:
       return retentionSweep();
+    case SYSTEM_JOBS.siemForward:
+      return runSiemForward();
     default:
       // batch-finalize runs on the system pool too but is routed by the
       // worker before calling runSystemJob. Anything else is a bug.

apps/api/src/routes/enterprise/index.ts

@@ -1,6 +1,8 @@
 import type { FastifyInstance } from "fastify";
 import { registerAuditExport } from "./audit-export.js";
+import { registerSiemRoutes } from "./siem.js";
 
 export async function registerEnterpriseRoutes(app: FastifyInstance) {
   await registerAuditExport(app);
+  await registerSiemRoutes(app);
 }

apps/api/src/routes/enterprise/siem.ts

@@ -0,0 +1,148 @@
+import { eq } from "drizzle-orm";
+import type { FastifyInstance, FastifyReply, FastifyRequest } from "fastify";
+import { z } from "zod";
+import { env } from "../../config.js";
+import { db, schema } from "../../db/index.js";
+import { auditLog } from "../../lib/audit.js";
+import { encrypt, isEncrypted } from "../../lib/encryption.js";
+import { requirePermission } from "../../permissions.js";
+
+const configSchema = z.object({
+  webhookUrl: z.string().url(),
+  authHeader: z.string().default(""),
+  flushIntervalSeconds: z.number().min(10).max(3600).default(30),
+  enabled: z.boolean(),
+});
+
+export type SiemConfig = z.infer<typeof configSchema>;
+
+const SETTINGS_KEY = "siem_config";
+
+export async function registerSiemRoutes(app: FastifyInstance): Promise<void> {
+  // GET /api/v1/enterprise/siem/config
+  app.get(
+    "/api/v1/enterprise/siem/config",
+    async (request: FastifyRequest, reply: FastifyReply) => {
+      const user = await requirePermission("webhooks:manage")(request, reply);
+      if (!user) return;
+
+      // Enterprise feature gate
+      let featureEnabled = false;
+      try {
+        const { isFeatureEnabled } = await import("@snapotter/enterprise");
+        featureEnabled = isFeatureEnabled("siem_forwarding");
+      } catch {
+        // Enterprise package not available
+      }
+      if (!featureEnabled) {
+        return reply
+          .status(403)
+          .send({ error: "SIEM forwarding requires an enterprise license with the siem_forwarding feature" });
+      }
+
+      const [row] = await db
+        .select({ value: schema.settings.value })
+        .from(schema.settings)
+        .where(eq(schema.settings.key, SETTINGS_KEY));
+
+      if (!row) {
+        return reply.send({
+          webhookUrl: "",
+          authHeader: "",
+          flushIntervalSeconds: 30,
+          enabled: false,
+        });
+      }
+
+      const config = JSON.parse(row.value) as SiemConfig;
+      return reply.send({
+        ...config,
+        authHeader: config.authHeader ? "***" : "",
+      });
+    },
+  );
+
+  // PUT /api/v1/enterprise/siem/config
+  app.put(
+    "/api/v1/enterprise/siem/config",
+    async (
+      request: FastifyRequest<{ Body: unknown }>,
+      reply: FastifyReply,
+    ) => {
+      const user = await requirePermission("webhooks:manage")(request, reply);
+      if (!user) return;
+
+      // Enterprise feature gate
+      let featureEnabled = false;
+      try {
+        const { isFeatureEnabled } = await import("@snapotter/enterprise");
+        featureEnabled = isFeatureEnabled("siem_forwarding");
+      } catch {
+        // Enterprise package not available
+      }
+      if (!featureEnabled) {
+        return reply
+          .status(403)
+          .send({ error: "SIEM forwarding requires an enterprise license with the siem_forwarding feature" });
+      }
+
+      const parsed = configSchema.safeParse(request.body);
+      if (!parsed.success) {
+        return reply.status(400).send({ error: "Invalid SIEM config", details: parsed.error.issues });
+      }
+
+      const config = { ...parsed.data };
+
+      // Encrypt the auth header before storage if encryption key is set
+      if (config.authHeader && env.DATA_ENCRYPTION_KEY) {
+        config.authHeader = await encrypt(config.authHeader, env.DATA_ENCRYPTION_KEY);
+      }
+
+      const value = JSON.stringify(config);
+      const now = new Date();
+
+      const [existing] = await db
+        .select()
+        .from(schema.settings)
+        .where(eq(schema.settings.key, SETTINGS_KEY));
+
+      if (existing) {
+        await db
+          .update(schema.settings)
+          .set({ value, updatedAt: now })
+          .where(eq(schema.settings.key, SETTINGS_KEY));
+      } else {
+        await db.insert(schema.settings).values({ key: SETTINGS_KEY, value });
+      }
+
+      await auditLog(request.log, "SETTINGS_UPDATED", {
+        adminId: user.id,
+        username: user.username,
+        keys: [SETTINGS_KEY],
+      }, request.ip);
+
+      return reply.send({ ok: true });
+    },
+  );
+
+  app.log.info("Enterprise SIEM routes registered");
+}
+
+/**
+ * Read the raw SIEM config from the settings table.
+ * Returns null if not configured.
+ */
+export async function readSiemConfig(): Promise<SiemConfig | null> {
+  const [row] = await db
+    .select({ value: schema.settings.value })
+    .from(schema.settings)
+    .where(eq(schema.settings.key, SETTINGS_KEY));
+
+  if (!row) return null;
+
+  try {
+    return JSON.parse(row.value) as SiemConfig;
+  } catch {
+    return null;
+  }
+}

tests/integration/siem-forwarding.test.ts

@@ -0,0 +1,48 @@
+import { describe, it, expect, beforeAll, afterAll } from "vitest";
+import { buildTestApp, loginAsAdmin, type TestApp } from "./test-server.js";
+
+let testApp: TestApp;
+let adminToken: string;
+
+beforeAll(async () => {
+  testApp = await buildTestApp();
+  adminToken = await loginAsAdmin(testApp.app);
+}, 30_000);
+
+afterAll(async () => {
+  await testApp.cleanup();
+}, 10_000);
+
+describe("SIEM config", () => {
+  it("returns 403 for GET without enterprise license", async () => {
+    const res = await testApp.app.inject({
+      method: "GET",
+      url: "/api/v1/enterprise/siem/config",
+      headers: { authorization: `Bearer ${adminToken}` },
+    });
+    expect(res.statusCode).toBe(403);
+  });
+
+  it("returns 403 for PUT without enterprise license", async () => {
+    const res = await testApp.app.inject({
+      method: "PUT",
+      url: "/api/v1/enterprise/siem/config",
+      headers: { authorization: `Bearer ${adminToken}` },
+      payload: {
+        webhookUrl: "https://siem.example.com/input",
+        authHeader: "Splunk test",
+        flushIntervalSeconds: 30,
+        enabled: true,
+      },
+    });
+    expect(res.statusCode).toBe(403);
+  });
+
+  it("returns 401 without auth", async () => {
+    const res = await testApp.app.inject({
+      method: "GET",
+      url: "/api/v1/enterprise/siem/config",
+    });
+    expect(res.statusCode).toBe(401);
+  });
+});