feat: events system implementation (#4246)

* chore: save poc

* chore: save wip

* fix: undo cors

* fix: impl changes

* fix: PR changes

* fix: mocks

* fix: connection tracking and auth changes

* fix: PR changes

* fix: revert license

* feat: frontend change

* fix: revert docker compose.dev

* fix: duplicate publisher connection

* fix: pr changes

* chore: move event impl to `ee`

* fix: lint errors

* fix: check length of events

* fix: static permissions matching

* fix: secretPath

* fix: remove source prefix in bus event name

* fix: license check

backend/src/server/plugins/auth/inject-identity.ts

@@ -22,6 +22,7 @@ export type TAuthMode =
       orgId: string;
       authMethod: AuthMethod;
       isMfaVerified?: boolean;
+      token: AuthModeJwtTokenPayload;
     }
   | {
       authMode: AuthMode.API_KEY;
@@ -30,6 +31,7 @@ export type TAuthMode =
       userId: string;
       user: TUsers;
       orgId: string;
+      token: string;
     }
   | {
       authMode: AuthMode.SERVICE_TOKEN;
@@ -38,6 +40,7 @@ export type TAuthMode =
       serviceTokenId: string;
       orgId: string;
       authMethod: null;
+      token: string;
     }
   | {
       authMode: AuthMode.IDENTITY_ACCESS_TOKEN;
@@ -47,6 +50,7 @@ export type TAuthMode =
       orgId: string;
       authMethod: null;
       isInstanceAdmin?: boolean;
+      token: TIdentityAccessTokenJwtPayload;
     }
   | {
       authMode: AuthMode.SCIM_TOKEN;
@@ -56,7 +60,7 @@ export type TAuthMode =
       authMethod: null;
     };
 
-const extractAuth = async (req: FastifyRequest, jwtSecret: string) => {
+export const extractAuth = async (req: FastifyRequest, jwtSecret: string) => {
   const apiKey = req.headers?.["x-api-key"];
   if (apiKey) {
     return { authMode: AuthMode.API_KEY, token: apiKey, actor: ActorType.USER } as const;
@@ -133,7 +137,8 @@ export const injectIdentity = fp(async (server: FastifyZodProvider) => {
           actor,
           orgId: orgId as string,
           authMethod: token.authMethod,
-          isMfaVerified: token.isMfaVerified
+          isMfaVerified: token.isMfaVerified,
+          token
         };
         break;
       }
@@ -148,7 +153,8 @@ export const injectIdentity = fp(async (server: FastifyZodProvider) => {
           identityId: identity.identityId,
           identityName: identity.name,
           authMethod: null,
-          isInstanceAdmin: serverCfg?.adminIdentityIds?.includes(identity.identityId)
+          isInstanceAdmin: serverCfg?.adminIdentityIds?.includes(identity.identityId),
+          token
         };
         if (token?.identityAuth?.oidc) {
           requestContext.set("identityAuthInfo", {
@@ -179,7 +185,8 @@ export const injectIdentity = fp(async (server: FastifyZodProvider) => {
           serviceToken,
           serviceTokenId: serviceToken.id,
           actor,
-          authMethod: null
+          authMethod: null,
+          token
         };
         break;
       }
@@ -191,7 +198,8 @@ export const injectIdentity = fp(async (server: FastifyZodProvider) => {
           actor,
           user,
           orgId: "API_KEY", // We set the orgId to an arbitrary value, since we can't link an API key to a specific org. We have to deprecate API keys soon!
-          authMethod: null
+          authMethod: null,
+          token: token as string
         };
         break;
       }

backend/src/server/routes/index.ts

@@ -31,6 +31,8 @@ import { buildDynamicSecretProviders } from "@app/ee/services/dynamic-secret/pro
 import { dynamicSecretLeaseDALFactory } from "@app/ee/services/dynamic-secret-lease/dynamic-secret-lease-dal";
 import { dynamicSecretLeaseQueueServiceFactory } from "@app/ee/services/dynamic-secret-lease/dynamic-secret-lease-queue";
 import { dynamicSecretLeaseServiceFactory } from "@app/ee/services/dynamic-secret-lease/dynamic-secret-lease-service";
+import { eventBusFactory } from "@app/ee/services/event/event-bus-service";
+import { sseServiceFactory } from "@app/ee/services/event/event-sse-service";
 import { externalKmsDALFactory } from "@app/ee/services/external-kms/external-kms-dal";
 import { externalKmsServiceFactory } from "@app/ee/services/external-kms/external-kms-service";
 import { gatewayDALFactory } from "@app/ee/services/gateway/gateway-dal";
@@ -495,6 +497,9 @@ export const registerRoutes = async (
   const projectMicrosoftTeamsConfigDAL = projectMicrosoftTeamsConfigDALFactory(db);
   const secretScanningV2DAL = secretScanningV2DALFactory(db);
 
+  const eventBusService = eventBusFactory(server.redis);
+  const sseService = sseServiceFactory(eventBusService, server.redis);
+
   const permissionService = permissionServiceFactory({
     permissionDAL,
     orgRoleDAL,
@@ -552,7 +557,8 @@ export const registerRoutes = async (
     queueService,
     projectDAL,
     licenseService,
-    auditLogStreamDAL
+    auditLogStreamDAL,
+    eventBusService
   });
 
   const auditLogService = auditLogServiceFactory({ auditLogDAL, permissionService, auditLogQueue });
@@ -1968,6 +1974,7 @@ export const registerRoutes = async (
   await kmsService.startService();
   await microsoftTeamsService.start();
   await dynamicSecretQueueService.init();
+  await eventBusService.init();
 
   // inject all services
   server.decorate<FastifyZodProvider["services"]>("services", {
@@ -2074,7 +2081,9 @@ export const registerRoutes = async (
     githubOrgSync: githubOrgSyncConfigService,
     folderCommit: folderCommitService,
     secretScanningV2: secretScanningV2Service,
-    reminder: reminderService
+    reminder: reminderService,
+    bus: eventBusService,
+    sse: sseService
   });
 
   const cronJobs: CronJob[] = [];
@@ -2190,5 +2199,7 @@ export const registerRoutes = async (
   server.addHook("onClose", async () => {
     cronJobs.forEach((job) => job.stop());
     await telemetryService.flushAll();
+    await eventBusService.close();
+    sseService.close();
   });
 };

backend/src/server/routes/v1/event-router.ts

@@ -0,0 +1,118 @@
+/* eslint-disable @typescript-eslint/no-floating-promises */
+import { subject } from "@casl/ability";
+import { pipeline } from "stream/promises";
+import { z } from "zod";
+
+import { ActionProjectType, ProjectType } from "@app/db/schemas";
+import { getServerSentEventsHeaders } from "@app/ee/services/event/event-sse-stream";
+import { EventRegisterSchema } from "@app/ee/services/event/types";
+import { ProjectPermissionSecretActions, ProjectPermissionSub } from "@app/ee/services/permission/project-permission";
+import { BadRequestError, ForbiddenRequestError, RateLimitError } from "@app/lib/errors";
+import { readLimit } from "@app/server/config/rateLimiter";
+import { verifyAuth } from "@app/server/plugins/auth/verify-auth";
+import { AuthMode } from "@app/services/auth/auth-type";
+
+export const registerEventRouter = async (server: FastifyZodProvider) => {
+  server.route({
+    method: "POST",
+    url: "/subscribe/project-events",
+    config: {
+      rateLimit: readLimit
+    },
+    schema: {
+      body: z.object({
+        projectId: z.string().trim(),
+        register: z.array(EventRegisterSchema).max(10)
+      })
+    },
+    onRequest: verifyAuth([AuthMode.JWT, AuthMode.IDENTITY_ACCESS_TOKEN]),
+    handler: async (req, reply) => {
+      try {
+        const { sse, permission, identityAccessToken, authToken, license } = req.server.services;
+
+        const plan = await license.getPlan(req.auth.orgId);
+
+        if (!plan.eventSubscriptions) {
+          throw new BadRequestError({
+            message:
+              "Failed to use event subscriptions due to plan restriction. Upgrade plan to access enterprise event subscriptions."
+          });
+        }
+
+        const count = await sse.getActiveConnectionsCount(req.body.projectId, req.permission.id);
+
+        if (count >= 5) {
+          throw new RateLimitError({
+            message: `Too many active connections for project ${req.body.projectId}. Please close some connections before opening a new one.`
+          });
+        }
+
+        const client = await sse.subscribe({
+          type: ProjectType.SecretManager,
+          registered: req.body.register,
+          async getAuthInfo() {
+            const ability = await permission.getProjectPermission({
+              actor: req.auth.actor,
+              projectId: req.body.projectId,
+              actionProjectType: ActionProjectType.Any,
+              actorAuthMethod: req.auth.authMethod,
+              actorId: req.permission.id,
+              actorOrgId: req.permission.orgId
+            });
+
+            return { permission: ability.permission, actorId: req.permission.id, projectId: req.body.projectId };
+          },
+          async onAuthRefresh(info) {
+            switch (req.auth.authMode) {
+              case AuthMode.JWT:
+                await authToken.fnValidateJwtIdentity(req.auth.token);
+                break;
+              case AuthMode.IDENTITY_ACCESS_TOKEN:
+                await identityAccessToken.fnValidateIdentityAccessToken(req.auth.token, req.realIp);
+                break;
+              default:
+                throw new Error("Unsupported authentication method");
+            }
+
+            req.body.register.forEach((r) => {
+              const allowed = info.permission.can(
+                ProjectPermissionSecretActions.Subscribe,
+                subject(ProjectPermissionSub.Secrets, {
+                  environment: r.conditions?.environmentSlug ?? "",
+                  secretPath: r.conditions?.secretPath ?? "/",
+                  eventType: r.event
+                })
+              );
+
+              if (!allowed) {
+                throw new ForbiddenRequestError({
+                  name: "PermissionDenied",
+                  message: `You are not allowed to subscribe on secrets`,
+                  details: {
+                    event: r.event,
+                    environmentSlug: r.conditions?.environmentSlug,
+                    secretPath: r.conditions?.secretPath ?? "/"
+                  }
+                });
+              }
+            });
+          }
+        });
+
+        // Switches to manual response and enable SSE streaming
+        reply.hijack();
+        reply.raw.writeHead(200, getServerSentEventsHeaders()).flushHeaders();
+        reply.raw.on("close", client.abort);
+
+        await pipeline(client.stream, reply.raw, { signal: client.signal });
+      } catch (error) {
+        if (error instanceof Error && error.name === "AbortError") {
+          // If the stream is aborted, we don't need to do anything
+          return;
+        }
+
+        throw error;
+      }
+    }
+  });
+};

backend/src/server/routes/v1/index.ts

@@ -13,6 +13,7 @@ import { registerCaRouter } from "./certificate-authority-router";
 import { CERTIFICATE_AUTHORITY_REGISTER_ROUTER_MAP } from "./certificate-authority-routers";
 import { registerCertRouter } from "./certificate-router";
 import { registerCertificateTemplateRouter } from "./certificate-template-router";
+import { registerEventRouter } from "./event-router";
 import { registerExternalGroupOrgRoleMappingRouter } from "./external-group-org-role-mapping-router";
 import { registerIdentityAccessTokenRouter } from "./identity-access-token-router";
 import { registerIdentityAliCloudAuthRouter } from "./identity-alicloud-auth-router";
@@ -183,4 +184,6 @@ export const registerV1Routes = async (server: FastifyZodProvider) => {
     },
     { prefix: "/reminders" }
   );
+
+  await server.register(registerEventRouter, { prefix: "/events" });
 };