From d86e4585e4e1069fa95aa49de951827fbdccf8f3 Mon Sep 17 00:00:00 2001 From: SnapOtter Date: Sun, 14 Jun 2026 12:21:46 +0800 Subject: [PATCH] feat(enterprise): add unified webhook system with admin alerts --- apps/api/src/jobs/alert-evaluator.ts | 124 +++++++++ apps/api/src/jobs/system-jobs.ts | 7 + apps/api/src/routes/enterprise/index.ts | 2 + apps/api/src/routes/enterprise/webhooks.ts | 277 +++++++++++++++++++++ 4 files changed, 410 insertions(+) create mode 100644 apps/api/src/jobs/alert-evaluator.ts create mode 100644 apps/api/src/routes/enterprise/webhooks.ts diff --git a/apps/api/src/jobs/alert-evaluator.ts b/apps/api/src/jobs/alert-evaluator.ts new file mode 100644 index 00000000..acbc9278 --- /dev/null +++ b/apps/api/src/jobs/alert-evaluator.ts @@ -0,0 +1,124 @@ +/** + * Alert condition evaluator. + * + * A periodic system job (every 60s) that checks several health/security + * conditions and delivers alerts to webhook destinations of type "alerts". + * + * Conditions checked: + * - Disk space below threshold (< 1 GB) + * - Auth anomaly (> 20 login failures in 5 minutes) + * - Backup staleness (> 48 hours since last completed backup) + * - License expiring (< 30 days remaining) + */ +import { statfs } from "node:fs/promises"; +import { and, eq, gte, sql } from "drizzle-orm"; +import { env } from "../config.js"; +import { db, schema } from "../db/index.js"; +import { getSettingString } from "../lib/settings-helpers.js"; + +export async function evaluateAlerts(): Promise { + // 1. Read webhook destinations from settings + const destJson = await getSettingString("webhook_destinations", "[]"); + let destinations: { url: string; authHeader: string; enabled: boolean; type: string }[]; + try { + destinations = JSON.parse(destJson); + } catch { + return; + } + + // 2. Filter to type = "alerts" and enabled = true + const alertDests = destinations.filter((d) => d.enabled && d.type === "alerts"); + if (alertDests.length === 0) return; + + // 3. Check conditions + const alerts: Record[] = []; + + // a. Disk space below threshold (< 1 GB) + try { + const stats = await statfs(env.WORKSPACE_PATH); + const freeGb = (stats.bfree * stats.bsize) / 1024 ** 3; + if (freeGb < 1) { + alerts.push({ condition: "disk_space_low", freeGb, threshold: 1 }); + } + } catch { + // statfs may fail on some filesystems; skip check + } + + // b. Auth anomaly (> 20 failures in last 5 minutes) + try { + const recentFailures = await db + .select({ count: sql`count(*)::int` }) + .from(schema.auditLog) + .where( + and( + eq(schema.auditLog.action, "LOGIN_FAILED"), + gte(schema.auditLog.createdAt, new Date(Date.now() - 5 * 60 * 1000)), + ), + ); + if (recentFailures[0].count > 20) { + alerts.push({ + condition: "auth_anomaly", + failedLogins: recentFailures[0].count, + windowMinutes: 5, + }); + } + } catch { + // Query may fail if audit_log is unavailable + } + + // c. Backup staleness (> 48 hours) + try { + const backupResult = await getSettingString("backup_last_completed", ""); + if (backupResult) { + const lastBackup = JSON.parse(backupResult); + const ageHours = (Date.now() - new Date(lastBackup.timestamp).getTime()) / 3_600_000; + if (ageHours > 48) { + alerts.push({ condition: "backup_stale", ageHours, threshold: 48 }); + } + } else { + alerts.push({ condition: "backup_never_run" }); + } + } catch { + // Backup check is best-effort + } + + // d. License expiration (< 30 days) + try { + const { getActiveLicense } = await import("@snapotter/enterprise"); + const license = getActiveLicense(); + if (license?.expiresAt) { + const daysLeft = (new Date(license.expiresAt).getTime() - Date.now()) / 86_400_000; + if (daysLeft < 30) { + alerts.push({ condition: "license_expiring", daysLeft: Math.floor(daysLeft) }); + } + } + } catch { + // Enterprise package not available; skip + } + + // 4. If no alerts triggered, nothing to do + if (alerts.length === 0) return; + + // 5. Deliver to each enabled "alerts" webhook + for (const dest of alertDests) { + let authHeader = dest.authHeader; + + // Decrypt auth header if encrypted + if (authHeader) { + try { + const { isEncrypted, decrypt } = await import("../lib/encryption.js"); + if (isEncrypted(authHeader) && env.DATA_ENCRYPTION_KEY) { + const decrypted = await decrypt(authHeader, env.DATA_ENCRYPTION_KEY); + authHeader = decrypted ?? ""; + } + } catch { + // Use raw value if decryption fails + } + } + + const { deliverWebhook } = await import("../lib/webhook-delivery.js"); + await deliverWebhook(dest.url, authHeader, alerts, { maxRetries: 1 }); + } + + console.log(`Alert evaluation complete: ${alerts.length} alert(s) delivered`); +} diff --git a/apps/api/src/jobs/system-jobs.ts b/apps/api/src/jobs/system-jobs.ts index cc5158e3..9a21f110 100644 --- a/apps/api/src/jobs/system-jobs.ts +++ b/apps/api/src/jobs/system-jobs.ts @@ -27,6 +27,7 @@ export const SYSTEM_JOBS = { auditArchive: "system:audit-archive", storageReconciliation: "system:storage-reconciliation", gdprExport: "system:gdpr-export", + alertEvaluator: "system:alert-evaluator", } as const; // -- Scheduling --------------------------------------------------------------- @@ -54,6 +55,8 @@ export async function scheduleSystemJobs(): Promise { await q.upsertJobScheduler(SYSTEM_JOBS.storageReconciliation, { pattern: "0 3 * * 0", }); + // Alert evaluator: every 60 seconds + await q.upsertJobScheduler(SYSTEM_JOBS.alertEvaluator, { every: 60_000 }); } /** Enqueue a one-shot system job (e.g. startup cleanup trigger). */ @@ -96,6 +99,10 @@ export async function runSystemJob(job: Job): Promise { .where(eq(schema.jobs.id, exportData.jobId)); return { outputRef }; } + case SYSTEM_JOBS.alertEvaluator: { + const { evaluateAlerts } = await import("./alert-evaluator.js"); + return evaluateAlerts(); + } default: // batch-finalize runs on the system pool too but is routed by the // worker before calling runSystemJob. Anything else is a bug. diff --git a/apps/api/src/routes/enterprise/index.ts b/apps/api/src/routes/enterprise/index.ts index 9e5f3c18..5d9cd0e9 100644 --- a/apps/api/src/routes/enterprise/index.ts +++ b/apps/api/src/routes/enterprise/index.ts @@ -7,6 +7,7 @@ import { registerLegalHoldRoutes } from "./legal-hold.js"; import { registerScimRoutes } from "./scim.js"; import { registerSiemRoutes } from "./siem.js"; import { registerUpgradeRoutes } from "./upgrade.js"; +import { registerWebhookRoutes } from "./webhooks.js"; export async function registerEnterpriseRoutes(app: FastifyInstance) { await registerAuditExport(app); @@ -17,4 +18,5 @@ export async function registerEnterpriseRoutes(app: FastifyInstance) { await registerScimRoutes(app); await registerSiemRoutes(app); await registerUpgradeRoutes(app); + await registerWebhookRoutes(app); } diff --git a/apps/api/src/routes/enterprise/webhooks.ts b/apps/api/src/routes/enterprise/webhooks.ts new file mode 100644 index 00000000..63c16151 --- /dev/null +++ b/apps/api/src/routes/enterprise/webhooks.ts @@ -0,0 +1,277 @@ +/** + * Unified webhook destination management (enterprise). + * + * CRUD for webhook destinations stored as a JSON array in the settings table + * under the key "webhook_destinations". Each destination can be type "siem" + * (forward audit events) or "alerts" (receive admin alert conditions). + * + * Gated behind the `webhooks:manage` permission + `admin_alerts` enterprise feature. + */ +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 { auditFromRequest } from "../../lib/audit.js"; +import { encrypt } from "../../lib/encryption.js"; +import { deliverWebhook } from "../../lib/webhook-delivery.js"; +import { requirePermission } from "../../permissions.js"; + +const SETTINGS_KEY = "webhook_destinations"; + +const webhookSchema = z.object({ + name: z.string().min(1).max(100), + url: z.string().url(), + authHeader: z.string().default(""), + eventFilter: z.array(z.string()).default([]), + batchIntervalSeconds: z.number().min(10).max(3600).default(30), + enabled: z.boolean().default(true), + type: z.enum(["siem", "alerts"]).default("alerts"), +}); + +export type WebhookDestination = z.infer; + +async function readDestinations(): Promise { + const [row] = await db + .select({ value: schema.settings.value }) + .from(schema.settings) + .where(eq(schema.settings.key, SETTINGS_KEY)); + + if (!row) return []; + + try { + return JSON.parse(row.value) as WebhookDestination[]; + } catch { + return []; + } +} + +async function writeDestinations(destinations: WebhookDestination[]): Promise { + const value = JSON.stringify(destinations); + 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 }); + } +} + +async function checkFeatureGate(reply: FastifyReply): Promise { + let featureEnabled = false; + try { + const { isFeatureEnabled } = await import("@snapotter/enterprise"); + featureEnabled = isFeatureEnabled("admin_alerts"); + } catch { + // Enterprise package not available + } + if (!featureEnabled) { + reply.status(403).send({ + error: "Webhook management requires a license with the admin_alerts feature", + }); + return false; + } + return true; +} + +export async function registerWebhookRoutes(app: FastifyInstance): Promise { + // GET /api/v1/enterprise/webhooks -- list all destinations + app.get("/api/v1/enterprise/webhooks", async (request: FastifyRequest, reply: FastifyReply) => { + const user = await requirePermission("webhooks:manage")(request, reply); + if (!user) return; + if (!(await checkFeatureGate(reply))) return; + + const destinations = await readDestinations(); + + // Mask auth headers in response + const masked = destinations.map((d) => ({ + ...d, + authHeader: d.authHeader ? "***" : "", + })); + + return reply.send({ destinations: masked }); + }); + + // POST /api/v1/enterprise/webhooks -- create a destination + app.post( + "/api/v1/enterprise/webhooks", + async (request: FastifyRequest<{ Body: unknown }>, reply: FastifyReply) => { + const user = await requirePermission("webhooks:manage")(request, reply); + if (!user) return; + if (!(await checkFeatureGate(reply))) return; + + const parsed = webhookSchema.safeParse(request.body); + if (!parsed.success) { + return reply + .status(400) + .send({ error: "Invalid webhook destination", details: parsed.error.issues }); + } + + const dest = { ...parsed.data }; + + // Encrypt the auth header before storage + if (dest.authHeader && env.DATA_ENCRYPTION_KEY) { + dest.authHeader = await encrypt(dest.authHeader, env.DATA_ENCRYPTION_KEY); + } + + const destinations = await readDestinations(); + destinations.push(dest); + await writeDestinations(destinations); + + await auditFromRequest(request)("SETTINGS_UPDATED", { + adminId: user.id, + username: user.username, + keys: [SETTINGS_KEY], + action: "webhook_created", + name: dest.name, + }); + + return reply.status(201).send({ ok: true, index: destinations.length - 1 }); + }, + ); + + // PUT /api/v1/enterprise/webhooks/:index -- update a destination by index + app.put( + "/api/v1/enterprise/webhooks/:index", + async ( + request: FastifyRequest<{ Params: { index: string }; Body: unknown }>, + reply: FastifyReply, + ) => { + const user = await requirePermission("webhooks:manage")(request, reply); + if (!user) return; + if (!(await checkFeatureGate(reply))) return; + + const index = parseInt(request.params.index, 10); + if (Number.isNaN(index) || index < 0) { + return reply.status(400).send({ error: "Invalid index" }); + } + + const parsed = webhookSchema.safeParse(request.body); + if (!parsed.success) { + return reply + .status(400) + .send({ error: "Invalid webhook destination", details: parsed.error.issues }); + } + + const destinations = await readDestinations(); + if (index >= destinations.length) { + return reply.status(404).send({ error: "Webhook destination not found" }); + } + + const dest = { ...parsed.data }; + + // Encrypt the auth header before storage + if (dest.authHeader && env.DATA_ENCRYPTION_KEY) { + dest.authHeader = await encrypt(dest.authHeader, env.DATA_ENCRYPTION_KEY); + } + + destinations[index] = dest; + await writeDestinations(destinations); + + await auditFromRequest(request)("SETTINGS_UPDATED", { + adminId: user.id, + username: user.username, + keys: [SETTINGS_KEY], + action: "webhook_updated", + name: dest.name, + }); + + return reply.send({ ok: true }); + }, + ); + + // DELETE /api/v1/enterprise/webhooks/:index -- remove a destination + app.delete( + "/api/v1/enterprise/webhooks/:index", + async (request: FastifyRequest<{ Params: { index: string } }>, reply: FastifyReply) => { + const user = await requirePermission("webhooks:manage")(request, reply); + if (!user) return; + if (!(await checkFeatureGate(reply))) return; + + const index = parseInt(request.params.index, 10); + if (Number.isNaN(index) || index < 0) { + return reply.status(400).send({ error: "Invalid index" }); + } + + const destinations = await readDestinations(); + if (index >= destinations.length) { + return reply.status(404).send({ error: "Webhook destination not found" }); + } + + const removed = destinations.splice(index, 1)[0]; + await writeDestinations(destinations); + + await auditFromRequest(request)("SETTINGS_UPDATED", { + adminId: user.id, + username: user.username, + keys: [SETTINGS_KEY], + action: "webhook_deleted", + name: removed.name, + }); + + return reply.send({ ok: true }); + }, + ); + + // POST /api/v1/enterprise/webhooks/:index/test -- send a test ping + app.post( + "/api/v1/enterprise/webhooks/:index/test", + async (request: FastifyRequest<{ Params: { index: string } }>, reply: FastifyReply) => { + const user = await requirePermission("webhooks:manage")(request, reply); + if (!user) return; + if (!(await checkFeatureGate(reply))) return; + + const index = parseInt(request.params.index, 10); + if (Number.isNaN(index) || index < 0) { + return reply.status(400).send({ error: "Invalid index" }); + } + + const destinations = await readDestinations(); + if (index >= destinations.length) { + return reply.status(404).send({ error: "Webhook destination not found" }); + } + + const dest = destinations[index]; + + // Decrypt auth header if needed + let authHeader = dest.authHeader; + if (authHeader) { + try { + const { isEncrypted, decrypt } = await import("../../lib/encryption.js"); + if (isEncrypted(authHeader) && env.DATA_ENCRYPTION_KEY) { + const decrypted = await decrypt(authHeader, env.DATA_ENCRYPTION_KEY); + authHeader = decrypted ?? ""; + } + } catch { + // Use raw value if decryption fails + } + } + + const testEvent = [ + { + condition: "test_ping", + message: "This is a test webhook from SnapOtter", + timestamp: new Date().toISOString(), + }, + ]; + + const result = await deliverWebhook(dest.url, authHeader, testEvent, { maxRetries: 0 }); + + return reply.send({ + ok: result.success, + statusCode: result.statusCode, + error: result.error, + }); + }, + ); + + app.log.info("Enterprise webhook routes registered"); +}