Merge pull request #3148 from bishnubista/fix-audit-log-replica-routing

fix(audit-logs): route request audit log reads through logsDb

messages/en-US.json

@@ -1957,7 +1957,7 @@
     "sshSudoModeCommandsDescription": "User can run only the specified commands with sudo.",
     "sshSudo": "Allow sudo",
     "sshSudoCommands": "Sudo Commands",
-    "sshSudoCommandsDescription": "Comma separated list of commands the user is allowed to run with sudo.",
+    "sshSudoCommandsDescription": "Comma separated list of commands the user is allowed to run with sudo. Absolute paths must be used.",
     "sshCreateHomeDir": "Create Home Directory",
     "sshUnixGroups": "Unix Groups",
     "sshUnixGroupsDescription": "Comma separated Unix groups to add the user to on the target host.",

server/lib/alerts/events/healthCheckEvents.ts

@@ -221,10 +221,18 @@ async function handleResource(
         )
         .where(eq(targets.resourceId, resource.resourceId));
 
+    const monitoredTargets = otherTargets.filter(
+        (t) => t.hcHealth !== "unknown"
+    );
+
     let health = "healthy";
-    const allUnknown = otherTargets.every((t) => t.hcHealth === "unknown");
+    const allUnknown = monitoredTargets.length === 0;
-    const allHealthy = otherTargets.every((t) => t.hcHealth === "healthy");
+    const allHealthy = monitoredTargets.every(
-    const allUnhealthy = otherTargets.every((t) => t.hcHealth === "unhealthy");
+        (t) => t.hcHealth === "healthy"
+    );
+    const allUnhealthy = monitoredTargets.every(
+        (t) => t.hcHealth === "unhealthy"
+    );
 
     if (allUnknown) {
         logger.debug(

server/lib/blueprints/types.ts

@@ -82,7 +82,7 @@ export const RuleSchema = z
     .object({
         action: z.enum(["allow", "deny", "pass"]),
         match: z.enum(["cidr", "path", "ip", "country", "asn", "region"]),
-        value: z.string(),
+        value: z.coerce.string(),
         priority: z.int().optional()
     })
     .refine(
@@ -340,7 +340,8 @@ export const ResourceSchema = z
             if (parts.includes("*", 1)) return false; // no further wildcards
             if (parts.length < 3) return false; // need at least *.label.tld
 
-            const labelRegex = /^[a-zA-Z0-9]([a-zA-Z0-9-]*[a-zA-Z0-9])?$|^[a-zA-Z0-9]$/;
+            const labelRegex =
+                /^[a-zA-Z0-9]([a-zA-Z0-9-]*[a-zA-Z0-9])?$|^[a-zA-Z0-9]$/;
             return parts.slice(1).every((label) => labelRegex.test(label));
         },
         {

server/lib/rebuildClientAssociations.ts

@@ -1826,3 +1826,77 @@ export async function verifyClientAssociationsCache(
         extraSiteIds: extraSiteIds.sort((a, b) => a - b)
     };
 }
+
+// cleanupSiteAssociations efficiently removes all client associations for a
+// site that is being deleted. Instead of calling
+// rebuildClientAssociationsFromSiteResource once per site resource (which is
+// O(resources) in DB round-trips and message fan-out), this function performs
+// a single bulk lookup of affected clients and site resources, deletes all
+// cache rows at once, and fires all peer/proxy removal messages in parallel.
+//
+// The caller is responsible for deleting the site row itself (and for sending
+// the newt/wg/terminate signal to the newt process).
+export async function cleanupSiteAssociations(
+    site: Site,
+    trx: Transaction | typeof db = db
+): Promise<void> {
+    const siteId = site.siteId;
+
+    logger.debug(`cleanupSiteAssociations: START siteId=${siteId}`);
+
+    // 1. Find every client currently cached against this site.
+    const cachedSiteClientRows = await trx
+        .select({ clientId: clientSitesAssociationsCache.clientId })
+        .from(clientSitesAssociationsCache)
+        .where(eq(clientSitesAssociationsCache.siteId, siteId));
+
+    const cachedClientIds = cachedSiteClientRows.map((r) => r.clientId);
+
+    // 2. Load full client details (needed for WireGuard public-key references).
+    const allClients =
+        cachedClientIds.length > 0
+            ? await trx
+                  .select({
+                      clientId: clients.clientId,
+                      pubKey: clients.pubKey,
+                      subnet: clients.subnet
+                  })
+                  .from(clients)
+                  .where(inArray(clients.clientId, cachedClientIds))
+            : [];
+
+    // 6. Bulk-delete all cache entries for this site.  Do this before sending
+    //    destination-update messages so updateClientSiteDestinations computes
+    //    the correct (post-deletion) set of destinations.
+    await trx
+        .delete(clientSitesAssociationsCache)
+        .where(eq(clientSitesAssociationsCache.siteId, siteId));
+
+    logger.debug(
+        `cleanupSiteAssociations: siteId=${siteId} cache cleared. clients=${allClients.length}`
+    );
+
+    // 7. Fire all removal messages in parallel.
+    const jobs: Promise<any>[] = [];
+
+    for (const client of allClients) {
+        // Tell each olm to drop the site's WireGuard peer.
+        if (site.publicKey) {
+            jobs.push(olmDeletePeer(client.clientId, siteId, site.publicKey));
+        }
+
+        // Recompute and push updated relay destinations (now excluding this site).
+        if (client.pubKey && client.subnet) {
+            jobs.push(updateClientSiteDestinations(client, trx));
+        }
+    }
+
+    await Promise.all(jobs).catch((error) => {
+        logger.error(
+            `cleanupSiteAssociations: error sending cleanup messages for siteId=${siteId}:`,
+            error
+        );
+    });
+
+    logger.debug(`cleanupSiteAssociations: DONE siteId=${siteId}`);
+}

server/routers/auditLogs/queryRequestAnalytics.ts

@@ -1,4 +1,4 @@
-import { logsDb, requestAuditLog, driver, primaryLogsDb } from "@server/db";
+import { logsDb, requestAuditLog, driver } from "@server/db";
 import { registry } from "@server/openApi";
 import { NextFunction } from "express";
 import { Request, Response } from "express";
@@ -74,12 +74,12 @@ async function query(query: Q) {
         );
     }
 
-    const [all] = await primaryLogsDb
+    const [all] = await logsDb
         .select({ total: count() })
         .from(requestAuditLog)
         .where(baseConditions);
 
-    const [blocked] = await primaryLogsDb
+    const [blocked] = await logsDb
         .select({ total: count() })
         .from(requestAuditLog)
         .where(and(baseConditions, eq(requestAuditLog.action, false)));
@@ -90,7 +90,7 @@ async function query(query: Q) {
 
     const DISTINCT_LIMIT = 500;
 
-    const requestsPerCountry = await primaryLogsDb
+    const requestsPerCountry = await logsDb
         .selectDistinct({
             code: requestAuditLog.location,
             count: totalQ
@@ -118,7 +118,7 @@ async function query(query: Q) {
     const booleanTrue = driver === "pg" ? sql`true` : sql`1`;
     const booleanFalse = driver === "pg" ? sql`false` : sql`0`;
 
-    const requestsPerDay = await primaryLogsDb
+    const requestsPerDay = await logsDb
         .select({
             day: groupByDayFunction.as("day"),
             allowedCount:

server/routers/auditLogs/queryRequestAuditLog.ts

@@ -1,4 +1,4 @@
-import { logsDb, primaryLogsDb, requestAuditLog, resources, siteResources, db, primaryDb } from "@server/db";
+import { logsDb, requestAuditLog, resources, siteResources, db, primaryDb } from "@server/db";
 import { registry } from "@server/openApi";
 import { NextFunction } from "express";
 import { Request, Response } from "express";
@@ -110,7 +110,7 @@ function getWhere(data: Q) {
 }
 
 export function queryRequest(data: Q) {
-    return primaryLogsDb
+    return logsDb
         .select({
             id: requestAuditLog.id,
                 timestamp: requestAuditLog.timestamp,
@@ -211,7 +211,7 @@ async function enrichWithResourceDetails(logs: Awaited<ReturnType<typeof queryRe
 }
 
 export function countRequestQuery(data: Q) {
-    const countQuery = primaryLogsDb
+    const countQuery = logsDb
         .select({ count: count() })
         .from(requestAuditLog)
         .where(getWhere(data));
@@ -254,34 +254,34 @@ async function queryUniqueFilterAttributes(
         uniqueResources,
         uniqueSiteResources
     ] = await Promise.all([
-        primaryLogsDb
+        logsDb
             .selectDistinct({ actor: requestAuditLog.actor })
             .from(requestAuditLog)
             .where(baseConditions)
             .limit(DISTINCT_LIMIT + 1),
-        primaryLogsDb
+        logsDb
             .selectDistinct({ locations: requestAuditLog.location })
             .from(requestAuditLog)
             .where(baseConditions)
             .limit(DISTINCT_LIMIT + 1),
-        primaryLogsDb
+        logsDb
             .selectDistinct({ hosts: requestAuditLog.host })
             .from(requestAuditLog)
             .where(baseConditions)
             .limit(DISTINCT_LIMIT + 1),
-        primaryLogsDb
+        logsDb
             .selectDistinct({ paths: requestAuditLog.path })
             .from(requestAuditLog)
             .where(baseConditions)
             .limit(DISTINCT_LIMIT + 1),
-        primaryLogsDb
+        logsDb
             .selectDistinct({
                 id: requestAuditLog.resourceId
             })
             .from(requestAuditLog)
             .where(baseConditions)
             .limit(DISTINCT_LIMIT + 1),
-        primaryLogsDb
+        logsDb
             .selectDistinct({
                 id: requestAuditLog.siteResourceId
             })

server/routers/site/deleteSite.ts

@@ -1,8 +1,8 @@
 import { Request, Response, NextFunction } from "express";
 import { z } from "zod";
-import { db, Site, siteNetworks, siteResources } from "@server/db";
+import { db } from "@server/db";
-import { newts, newtSessions, sites } from "@server/db";
+import { newts, sites } from "@server/db";
-import { eq, inArray } from "drizzle-orm";
+import { eq } from "drizzle-orm";
 import response from "@server/lib/response";
 import HttpCode from "@server/types/HttpCode";
 import createHttpError from "http-errors";
@@ -11,7 +11,7 @@ import { deletePeer } from "../gerbil/peers";
 import { fromError } from "zod-validation-error";
 import { sendToClient } from "#dynamic/routers/ws";
 import { OpenAPITags, registry } from "@server/openApi";
-import { rebuildClientAssociationsFromSiteResource } from "@server/lib/rebuildClientAssociations";
+import { cleanupSiteAssociations } from "@server/lib/rebuildClientAssociations";
 import { usageService } from "@server/lib/billing/usageService";
 import { FeatureId } from "@server/lib/billing";
 
@@ -63,7 +63,11 @@ export async function deleteSite(
             );
         }
 
-        let deletedNewtId: string | null = null;
+        const [deletedNewt] = await db
+            .select()
+            .from(newts)
+            .where(eq(newts.siteId, siteId))
+            .limit(1);
 
         await db.transaction(async (trx) => {
             if (site.type == "wireguard") {
@@ -71,56 +75,24 @@ export async function deleteSite(
                     await deletePeer(site.exitNodeId!, site.pubKey);
                 }
             } else if (site.type == "newt") {
-                const networks = await trx
+                // Clean up all client associations and send peer/proxy removal
-                    .select({ networkId: siteNetworks.networkId })
+                // messages in a single efficient pass before deleting the row.
-                    .from(siteNetworks)
+                await cleanupSiteAssociations(site, trx);
-                    .where(eq(siteNetworks.siteId, siteId));
-
-                // loop through them
-                const updatedSiteResources = await trx
-                    .select()
-                    .from(siteResources)
-                    .where(
-                        inArray(
-                            siteResources.networkId,
-                            networks.map((n) => n.networkId)
-                        )
-                    );
-                for (const siteResource of updatedSiteResources) {
-                    await rebuildClientAssociationsFromSiteResource(
-                        siteResource,
-                        trx
-                    );
-                }
-
-                // get the newt on the site by querying the newt table for siteId
-                const [deletedNewt] = await trx
-                    .delete(newts)
-                    .where(eq(newts.siteId, siteId))
-                    .returning();
-                if (deletedNewt) {
-                    deletedNewtId = deletedNewt.newtId;
-
-                    // delete all of the sessions for the newt
-                    await trx
-                        .delete(newtSessions)
-                        .where(eq(newtSessions.newtId, deletedNewt.newtId));
-                }
-            }
 
                 await trx.delete(sites).where(eq(sites.siteId, siteId));
+            }
 
             await usageService.add(site.orgId, FeatureId.SITES, -1, trx);
         });
 
         // Send termination message outside of transaction to prevent blocking
-        if (deletedNewtId) {
+        if (deletedNewt) {
             const payload = {
                 type: `newt/wg/terminate`,
                 data: {}
             };
             // Don't await this to prevent blocking the response
-            sendToClient(deletedNewtId, payload).catch((error) => {
+            sendToClient(deletedNewt.newtId, payload).catch((error) => {
                 logger.error(
                     "Failed to send termination message to newt:",
                     error

server/routers/siteResource/batchAddClientToSiteResources.ts

@@ -15,10 +15,7 @@ import logger from "@server/logger";
 import { fromError } from "zod-validation-error";
 import { eq, and, inArray } from "drizzle-orm";
 import { OpenAPITags, registry } from "@server/openApi";
-import {
+import { rebuildClientAssociationsFromClient } from "@server/lib/rebuildClientAssociations";
-    rebuildClientAssociationsFromClient,
-    rebuildClientAssociationsFromSiteResource
-} from "@server/lib/rebuildClientAssociations";
 
 const batchAddClientToSiteResourcesParamsSchema = z
     .object({