Add Config updates via SSE

end2end/server/app.ts

@@ -10,6 +10,8 @@ import { updateConfig } from "./src/handlers/updateConfig.ts";
 import { lists } from "./src/handlers/lists.ts";
 import { updateIPLists } from "./src/handlers/updateLists.ts";
 import { realtimeConfig } from "./src/handlers/realtimeConfig.ts";
+import { stream, disconnectStreams } from "./src/handlers/stream.ts";
+import { deleteApp } from "./src/handlers/deleteApp.ts";
 
 const app = express();
 app.set("trust proxy", false);
@@ -24,6 +26,8 @@ app.post("/api/runtime/config", checkToken, updateConfig);
 
 // Realtime polling endpoint
 app.get("/config", checkToken, realtimeConfig);
+app.get("/api/runtime/stream", checkToken, stream);
+app.post("/api/runtime/stream/disconnect", checkToken, disconnectStreams);
 
 app.get("/api/runtime/events", checkToken, listEvents);
 app.post("/api/runtime/events", checkToken, captureEvent);
@@ -32,6 +36,7 @@ app.get("/api/runtime/firewall/lists", checkToken, lists);
 app.post("/api/runtime/firewall/lists", checkToken, updateIPLists);
 
 app.post("/api/runtime/apps", createApp);
+app.delete("/api/runtime/apps", checkToken, deleteApp);
 
 app.listen(port, () => {
   console.log(`Server is running on port ${port}`);

end2end/server/src/handlers/deleteApp.ts

@@ -0,0 +1,14 @@
+import type { Response } from "express";
+import { removeApp } from "../zen/apps.ts";
+import { closeStreams } from "./stream.ts";
+import type { ZenRequest } from "../types.ts";
+
+export function deleteApp(req: ZenRequest, res: Response) {
+  if (!req.zenApp) {
+    throw new Error("App is missing");
+  }
+
+  removeApp(req.zenApp);
+  closeStreams(req.zenApp.id);
+  res.json({ ok: true });
+}

end2end/server/src/handlers/stream.ts

@@ -0,0 +1,64 @@
+import type { Response } from "express";
+import { getAppConfig, configEvents } from "../zen/config.ts";
+import type { ZenRequest } from "../types.ts";
+
+const connections = new Map<number, Set<Response>>();
+
+export function stream(req: ZenRequest, res: Response) {
+  if (!req.zenApp) {
+    throw new Error("App is missing");
+  }
+
+  const app = req.zenApp;
+
+  if (!connections.has(app.id)) {
+    connections.set(app.id, new Set());
+  }
+  connections.get(app.id)!.add(res);
+
+  res.writeHead(200, {
+    "Content-Type": "text/event-stream",
+    "Cache-Control": "no-cache",
+    Connection: "keep-alive",
+  });
+
+  function sendConfig() {
+    const config = getAppConfig(app);
+    const data = { serviceId: app.id, configUpdatedAt: config.configUpdatedAt };
+    res.write(`event: config-updated\ndata: ${JSON.stringify(data)}\n\n`);
+  }
+
+  sendConfig();
+
+  const eventName = `config-updated:${app.id}`;
+  configEvents.on(eventName, sendConfig);
+
+  const ping = setInterval(() => {
+    res.write(": ping\n\n");
+  }, 30_000);
+
+  req.on("close", () => {
+    connections.get(app.id)?.delete(res);
+    configEvents.off(eventName, sendConfig);
+    clearInterval(ping);
+  });
+}
+
+export function closeStreams(appId: number) {
+  const appConnections = connections.get(appId);
+  if (appConnections) {
+    for (const conn of appConnections) {
+      conn.end();
+    }
+    appConnections.clear();
+  }
+}
+
+export function disconnectStreams(req: ZenRequest, res: Response) {
+  if (!req.zenApp) {
+    throw new Error("App is missing");
+  }
+
+  closeStreams(req.zenApp.id);
+  res.json({ ok: true });
+}

end2end/server/src/zen/apps.ts

@@ -6,7 +6,7 @@ export type App = {
   configUpdatedAt: number;
 };
 
-const apps: App[] = [];
+let apps: App[] = [];
 
 let id = 1;
 export function createApp(): string {
@@ -20,6 +20,10 @@ export function createApp(): string {
   return token;
 }
 
+export function removeApp(app: App): void {
+  apps = apps.filter((a) => a.id !== app.id);
+}
+
 export function getByToken(token: string): App | undefined {
   return apps.find((app) => {
     if (app.token.length !== token.length) {

end2end/server/src/zen/config.ts

@@ -1,5 +1,8 @@
+import { EventEmitter } from "node:events";
 import type { App } from "./apps.ts";
 
+export const configEvents = new EventEmitter();
+
 type AppConfig = {
   success: boolean;
   serviceId: number;
@@ -53,6 +56,7 @@ export function updateAppConfig(app: App, newConfig: Partial<AppConfig>) {
     ...newConfig,
     configUpdatedAt: Date.now(),
   };
+  configEvents.emit(`config-updated:${app.id}`);
   return true;
 }
 

end2end/tests-new/realtime-config-updates.test.mjs

@@ -0,0 +1,224 @@
+import { spawn } from "child_process";
+import { resolve } from "path";
+import { test } from "node:test";
+import { equal, doesNotMatch, match, fail } from "node:assert";
+import { getRandomPort } from "./utils/get-port.mjs";
+import { timeout } from "./utils/timeout.mjs";
+
+const pathToAppDir = resolve(
+  import.meta.dirname,
+  "../../sample-apps/hono-pg-ts-esm"
+);
+
+const testServerUrl = "http://localhost:5874";
+
+function spawnApp(token, port) {
+  return spawn(
+    `node`,
+    [
+      "--require",
+      "@aikidosec/firewall/instrument",
+      "--experimental-strip-types",
+      "./app.ts",
+      port,
+    ],
+    {
+      cwd: pathToAppDir,
+      env: {
+        ...process.env,
+        AIKIDO_TOKEN: token,
+        AIKIDO_ENDPOINT: testServerUrl,
+        AIKIDO_REALTIME_ENDPOINT: testServerUrl,
+        AIKIDO_DEBUG: "true",
+        AIKIDO_DEBUG_SSE: "true",
+        AIKIDO_BLOCK: "true",
+        AIKIDO_FEATURE_SSE: "true",
+      },
+    }
+  );
+}
+
+test("it picks up blocked IP via SSE config update", async () => {
+  const response = await fetch(`${testServerUrl}/api/runtime/apps`, {
+    method: "POST",
+  });
+  const body = await response.json();
+  const token = body.token;
+  const port = await getRandomPort();
+
+  const server = spawnApp(token, port);
+
+  try {
+    server.on("error", (err) => {
+      fail(err);
+    });
+
+    let stdout = "";
+    server.stdout.on("data", (data) => {
+      stdout += data.toString();
+    });
+
+    let stderr = "";
+    server.stderr.on("data", (data) => {
+      stderr += data.toString();
+    });
+
+    // Wait for the server to start and SSE to connect
+    await timeout(3000);
+
+    // Verify request from 5.6.7.8 is allowed before blocking
+    const before = await fetch(`http://127.0.0.1:${port}/`, {
+      headers: { "x-forwarded-for": "5.6.7.8" },
+      signal: AbortSignal.timeout(5000),
+    });
+    equal(before.status, 200);
+
+    // Block IP 5.6.7.8 via the test server API
+    await fetch(`${testServerUrl}/api/runtime/firewall/lists`, {
+      method: "POST",
+      headers: {
+        "Content-Type": "application/json",
+        Authorization: token,
+      },
+      body: JSON.stringify({
+        blockedIPAddresses: ["5.6.7.8"],
+      }),
+    });
+
+    // Wait for SSE config-updated event to propagate
+    await timeout(2000);
+
+    // Verify request from 5.6.7.8 is now blocked
+    const after = await fetch(`http://127.0.0.1:${port}/`, {
+      headers: { "x-forwarded-for": "5.6.7.8" },
+      signal: AbortSignal.timeout(5000),
+    });
+    equal(after.status, 403);
+  } catch (err) {
+    fail(err);
+  } finally {
+    server.kill();
+  }
+});
+
+test("it reconnects SSE after server disconnects", async () => {
+  const response = await fetch(`${testServerUrl}/api/runtime/apps`, {
+    method: "POST",
+  });
+  const body = await response.json();
+  const token = body.token;
+  const port = await getRandomPort();
+
+  const server = spawnApp(token, port);
+
+  try {
+    server.on("error", (err) => {
+      fail(err);
+    });
+
+    let stdout = "";
+    server.stdout.on("data", (data) => {
+      stdout += data.toString();
+    });
+
+    let stderr = "";
+    server.stderr.on("data", (data) => {
+      stderr += data.toString();
+    });
+
+    // Wait for the server to start and SSE to connect
+    await timeout(3000);
+    match(stdout, /SSE connected successfully/);
+
+    // Disconnect SSE from the server side
+    await fetch(`${testServerUrl}/api/runtime/stream/disconnect`, {
+      method: "POST",
+      headers: { Authorization: token },
+    });
+
+    // Wait for reconnect (initial reconnect delay is 5s + jitter up to 7.5s)
+    await timeout(10000);
+    match(stdout, /SSE connection closed by server/);
+
+    // Verify SSE reconnected
+    const connectedCount =
+      stdout.split("SSE connected successfully").length - 1;
+    equal(connectedCount >= 2, true);
+
+    // Block IP 9.8.7.6 after reconnect to verify the new connection works
+    await fetch(`${testServerUrl}/api/runtime/firewall/lists`, {
+      method: "POST",
+      headers: {
+        "Content-Type": "application/json",
+        Authorization: token,
+      },
+      body: JSON.stringify({
+        blockedIPAddresses: ["9.8.7.6"],
+      }),
+    });
+
+    // Wait for SSE config-updated event to propagate
+    await timeout(2000);
+
+    // Verify the blocked IP is picked up via the reconnected SSE
+    const blocked = await fetch(`http://127.0.0.1:${port}/`, {
+      headers: { "x-forwarded-for": "9.8.7.6" },
+      signal: AbortSignal.timeout(5000),
+    });
+    equal(blocked.status, 403);
+  } catch (err) {
+    fail(err);
+  } finally {
+    server.kill();
+  }
+});
+
+test("it stops SSE reconnect on 401", async () => {
+  const response = await fetch(`${testServerUrl}/api/runtime/apps`, {
+    method: "POST",
+  });
+  const body = await response.json();
+  const token = body.token;
+  const port = await getRandomPort();
+
+  const server = spawnApp(token, port);
+
+  try {
+    server.on("error", (err) => {
+      fail(err);
+    });
+
+    let stdout = "";
+    server.stdout.on("data", (data) => {
+      stdout += data.toString();
+    });
+
+    let stderr = "";
+    server.stderr.on("data", (data) => {
+      stderr += data.toString();
+    });
+
+    // Wait for the server to start and SSE to connect
+    await timeout(3000);
+    match(stdout, /SSE connected successfully/);
+
+    // Revoke the token and disconnect SSE so it tries to reconnect with 401
+    await fetch(`${testServerUrl}/api/runtime/apps`, {
+      method: "DELETE",
+      headers: { Authorization: token },
+    });
+
+    // Wait for reconnect attempts (may take multiple due to backoff + jitter)
+    await timeout(15000);
+
+    match(stdout, /SSE connection rejected with status 401, stopping/);
+    // Should not schedule a reconnect after 401
+    const rejectedIndex = stdout.indexOf("SSE connection rejected");
+    const afterRejected = stdout.slice(rejectedIndex);
+    doesNotMatch(afterRejected, /SSE scheduling reconnect/);
+  } catch (err) {
+    fail(err);
+  } finally {
+    server.kill();
+  }
+});