feat(server): audit log streams services and api routes

backend/src/ee/services/audit-log-stream/audit-log-stream-dal.ts

@@ -0,0 +1,11 @@
+import { TDbClient } from "@app/db";
+import { TableName } from "@app/db/schemas";
+import { ormify } from "@app/lib/knex";
+
+export type TAuditLogStreamDALFactory = ReturnType<typeof auditLogStreamDALFactory>;
+
+export const auditLogStreamDALFactory = (db: TDbClient) => {
+  const orm = ormify(db, TableName.AuditLogStream);
+
+  return orm;
+};

backend/src/ee/services/audit-log-stream/audit-log-stream-service.ts

@@ -0,0 +1,205 @@
+import { ForbiddenError } from "@casl/ability";
+
+import { SecretKeyEncoding } from "@app/db/schemas";
+import { infisicalSymmetricDecrypt, infisicalSymmetricEncypt } from "@app/lib/crypto/encryption";
+import { BadRequestError } from "@app/lib/errors";
+import { validateLocalIps } from "@app/lib/validator";
+import { TProjectDALFactory } from "@app/services/project/project-dal";
+
+import { TLicenseServiceFactory } from "../license/license-service";
+import { TPermissionServiceFactory } from "../permission/permission-service";
+import { ProjectPermissionActions, ProjectPermissionSub } from "../permission/project-permission";
+import { TAuditLogStreamDALFactory } from "./audit-log-stream-dal";
+import {
+  TCreateAuditLogStreamDTO,
+  TDeleteAuditLogStreamDTO,
+  TGetDetailsAuditLogStreamDTO,
+  TListAuditLogStreamDTO,
+  TUpdateAuditLogStreamDTO
+} from "./audit-log-stream-types";
+
+type TAuditLogStreamServiceFactoryDep = {
+  auditLogStreamDAL: TAuditLogStreamDALFactory;
+  permissionService: Pick<TPermissionServiceFactory, "getProjectPermission">;
+  projectDAL: Pick<TProjectDALFactory, "findProjectBySlug">;
+  licenseService: Pick<TLicenseServiceFactory, "getPlan">;
+};
+
+export type TAuditLogStreamServiceFactory = ReturnType<typeof auditLogStreamServiceFactory>;
+
+export const auditLogStreamServiceFactory = ({
+  auditLogStreamDAL,
+  permissionService,
+  projectDAL,
+  licenseService
+}: TAuditLogStreamServiceFactoryDep) => {
+  const create = async ({
+    projectSlug,
+    url,
+    actor,
+    token,
+    actorId,
+    actorOrgId,
+    actorAuthMethod
+  }: TCreateAuditLogStreamDTO) => {
+    const plan = await licenseService.getPlan(actorOrgId);
+    if (!plan.auditLogStreams)
+      throw new BadRequestError({
+        message: "Failed to create audit log streams due to plan restriction. Upgrade plan to create group."
+      });
+
+    const project = await projectDAL.findProjectBySlug(projectSlug, actorOrgId);
+    if (!project) throw new BadRequestError({ message: "Project not found" });
+    const projectId = project.id;
+
+    const { permission } = await permissionService.getProjectPermission(
+      actor,
+      actorId,
+      projectId,
+      actorAuthMethod,
+      actorOrgId
+    );
+    ForbiddenError.from(permission).throwUnlessCan(ProjectPermissionActions.Create, ProjectPermissionSub.Settings);
+
+    validateLocalIps(url);
+
+    const totalStreams = await auditLogStreamDAL.find({ projectId });
+    if (totalStreams.length >= plan.auditLogStreamLimit) {
+      throw new BadRequestError({
+        message:
+          "Failed to create audit log streams due to plan limit reached. Kindly contact Infisical to add more streams."
+      });
+    }
+    const encryptedToken = token ? infisicalSymmetricEncypt(token) : undefined;
+    const logStream = await auditLogStreamDAL.create({
+      projectId,
+      url,
+      ...(encryptedToken
+        ? {
+            encryptedTokenCiphertext: encryptedToken.ciphertext,
+            encryptedTokenIV: encryptedToken.iv,
+            encryptedTokenTag: encryptedToken.tag,
+            encryptedTokenAlgorithm: encryptedToken.algorithm,
+            encryptedTokenKeyEncoding: encryptedToken.encoding
+          }
+        : {})
+    });
+    return logStream;
+  };
+
+  const updateById = async ({
+    id,
+    url,
+    actor,
+    token,
+    actorId,
+    actorOrgId,
+    actorAuthMethod
+  }: TUpdateAuditLogStreamDTO) => {
+    const plan = await licenseService.getPlan(actorOrgId);
+    if (!plan.auditLogStreams)
+      throw new BadRequestError({
+        message: "Failed to update audit log streams due to plan restriction. Upgrade plan to create group."
+      });
+
+    const logStream = await auditLogStreamDAL.findById(id);
+    if (!logStream) throw new BadRequestError({ message: "Audit log stream not found" });
+
+    const { projectId } = logStream;
+    const { permission } = await permissionService.getProjectPermission(
+      actor,
+      actorId,
+      projectId,
+      actorAuthMethod,
+      actorOrgId
+    );
+    ForbiddenError.from(permission).throwUnlessCan(ProjectPermissionActions.Edit, ProjectPermissionSub.Settings);
+
+    if (url) validateLocalIps(url);
+    const encryptedToken = token ? infisicalSymmetricEncypt(token) : undefined;
+    const updatedLogStream = await auditLogStreamDAL.updateById(id, {
+      projectId,
+      url,
+      ...(encryptedToken
+        ? {
+            encryptedTokenCiphertext: encryptedToken.ciphertext,
+            encryptedTokenIV: encryptedToken.iv,
+            encryptedTokenTag: encryptedToken.tag,
+            encryptedTokenAlgorithm: encryptedToken.algorithm,
+            encryptedTokenKeyEncoding: encryptedToken.encoding
+          }
+        : {})
+    });
+    return updatedLogStream;
+  };
+
+  const deleteById = async ({ id, actor, actorId, actorOrgId, actorAuthMethod }: TDeleteAuditLogStreamDTO) => {
+    const logStream = await auditLogStreamDAL.findById(id);
+    if (!logStream) throw new BadRequestError({ message: "Audit log stream not found" });
+
+    const { projectId } = logStream;
+    const { permission } = await permissionService.getProjectPermission(
+      actor,
+      actorId,
+      projectId,
+      actorAuthMethod,
+      actorOrgId
+    );
+    ForbiddenError.from(permission).throwUnlessCan(ProjectPermissionActions.Delete, ProjectPermissionSub.Settings);
+
+    const deletedLogStream = await auditLogStreamDAL.deleteById(id);
+    return deletedLogStream;
+  };
+
+  const getById = async ({ id, actor, actorId, actorOrgId, actorAuthMethod }: TGetDetailsAuditLogStreamDTO) => {
+    const logStream = await auditLogStreamDAL.findById(id);
+    if (!logStream) throw new BadRequestError({ message: "Audit log stream not found" });
+
+    const { projectId } = logStream;
+    const { permission } = await permissionService.getProjectPermission(
+      actor,
+      actorId,
+      projectId,
+      actorAuthMethod,
+      actorOrgId
+    );
+    ForbiddenError.from(permission).throwUnlessCan(ProjectPermissionActions.Read, ProjectPermissionSub.Settings);
+    const token =
+      logStream?.encryptedTokenCiphertext && logStream?.encryptedTokenIV && logStream?.encryptedTokenTag
+        ? infisicalSymmetricDecrypt({
+            tag: logStream.encryptedTokenTag,
+            iv: logStream.encryptedTokenIV,
+            ciphertext: logStream.encryptedTokenCiphertext,
+            keyEncoding: logStream.encryptedTokenKeyEncoding as SecretKeyEncoding
+          })
+        : undefined;
+
+    return { ...logStream, token };
+  };
+
+  const list = async ({ projectSlug, actor, actorId, actorOrgId, actorAuthMethod }: TListAuditLogStreamDTO) => {
+    const project = await projectDAL.findProjectBySlug(projectSlug, actorOrgId);
+    if (!project) throw new BadRequestError({ message: "Project not found" });
+    const projectId = project.id;
+
+    const { permission } = await permissionService.getProjectPermission(
+      actor,
+      actorId,
+      projectId,
+      actorAuthMethod,
+      actorOrgId
+    );
+    ForbiddenError.from(permission).throwUnlessCan(ProjectPermissionActions.Create, ProjectPermissionSub.Settings);
+
+    const logStreams = await auditLogStreamDAL.find({ projectId });
+    return logStreams;
+  };
+
+  return {
+    create,
+    updateById,
+    deleteById,
+    getById,
+    list
+  };
+};

backend/src/ee/services/audit-log-stream/audit-log-stream-types.ts

@@ -0,0 +1,25 @@
+import { TProjectPermission } from "@app/lib/types";
+
+export type TCreateAuditLogStreamDTO = Omit<TProjectPermission, "projectId"> & {
+  projectSlug: string;
+  url: string;
+  token?: string;
+};
+
+export type TUpdateAuditLogStreamDTO = Omit<TProjectPermission, "projectId"> & {
+  id: string;
+  url?: string;
+  token?: string;
+};
+
+export type TDeleteAuditLogStreamDTO = Omit<TProjectPermission, "projectId"> & {
+  id: string;
+};
+
+export type TListAuditLogStreamDTO = Omit<TProjectPermission, "projectId"> & {
+  projectSlug: string;
+};
+
+export type TGetDetailsAuditLogStreamDTO = Omit<TProjectPermission, "projectId"> & {
+  id: string;
+};

backend/src/ee/services/audit-log/audit-log-queue.ts

@@ -1,13 +1,20 @@
+import { RawAxiosRequestHeaders } from "axios";
+
+import { SecretKeyEncoding } from "@app/db/schemas";
+import { request } from "@app/lib/config/request";
+import { infisicalSymmetricDecrypt } from "@app/lib/crypto/encryption";
 import { logger } from "@app/lib/logger";
 import { QueueJobs, QueueName, TQueueServiceFactory } from "@app/queue";
 import { TProjectDALFactory } from "@app/services/project/project-dal";
 
+import { TAuditLogStreamDALFactory } from "../audit-log-stream/audit-log-stream-dal";
 import { TLicenseServiceFactory } from "../license/license-service";
 import { TAuditLogDALFactory } from "./audit-log-dal";
 import { TCreateAuditLogDTO } from "./audit-log-types";
 
 type TAuditLogQueueServiceFactoryDep = {
   auditLogDAL: TAuditLogDALFactory;
+  auditLogStreamDAL: Pick<TAuditLogStreamDALFactory, "find">;
   queueService: TQueueServiceFactory;
   projectDAL: Pick<TProjectDALFactory, "findById">;
   licenseService: Pick<TLicenseServiceFactory, "getPlan">;
@@ -15,11 +22,15 @@ type TAuditLogQueueServiceFactoryDep = {
 
 export type TAuditLogQueueServiceFactory = ReturnType<typeof auditLogQueueServiceFactory>;
 
+// keep this timeout 5s it must be fast because else the queue will take time to finish
+// audit log is a crowded queue thus needs to be fast
+const AUDIT_LOG_STREAM_TIMEOUT = 5 * 1000;
 export const auditLogQueueServiceFactory = ({
   auditLogDAL,
   queueService,
   projectDAL,
-  licenseService
+  licenseService,
+  auditLogStreamDAL
 }: TAuditLogQueueServiceFactoryDep) => {
   const pushToLog = async (data: TCreateAuditLogDTO) => {
     await queueService.queue(QueueName.AuditLog, QueueJobs.AuditLog, data, {
@@ -47,7 +58,7 @@ export const auditLogQueueServiceFactory = ({
     // skip inserting if audit log retention is 0 meaning its not supported
     if (ttl === 0) return;
 
-    await auditLogDAL.create({
+    const auditLog = await auditLogDAL.create({
       actor: actor.type,
       actorMetadata: actor.metadata,
       userAgent,
@@ -59,6 +70,32 @@ export const auditLogQueueServiceFactory = ({
       eventMetadata: event.metadata,
       userAgentType
     });
+
+    const logStreams = await auditLogStreamDAL.find({ projectId });
+    await Promise.allSettled(
+      logStreams.map(
+        async ({ url, encryptedTokenTag, encryptedTokenIV, encryptedTokenKeyEncoding, encryptedTokenCiphertext }) => {
+          const token =
+            encryptedTokenIV && encryptedTokenCiphertext && encryptedTokenTag
+              ? infisicalSymmetricDecrypt({
+                  keyEncoding: encryptedTokenKeyEncoding as SecretKeyEncoding,
+                  iv: encryptedTokenIV,
+                  tag: encryptedTokenTag,
+                  ciphertext: encryptedTokenCiphertext
+                })
+              : undefined;
+          const headers: RawAxiosRequestHeaders = { "Content-Type": "application/json" };
+          if (token) headers.Authorization = `Bearer ${token}`;
+          return request.post(url, auditLog, {
+            headers,
+            // request timeout
+            timeout: AUDIT_LOG_STREAM_TIMEOUT,
+            // connection timeout
+            signal: AbortSignal.timeout(AUDIT_LOG_STREAM_TIMEOUT)
+          });
+        }
+      )
+    );
   });
 
   queueService.start(QueueName.AuditLogPrune, async () => {

backend/src/ee/services/license/licence-fns.ts

@@ -24,6 +24,8 @@ export const getDefaultOnPremFeatures = (): TFeatureSet => ({
   customAlerts: false,
   auditLogs: false,
   auditLogsRetentionDays: 0,
+  auditLogStreams: false,
+  auditLogStreamLimit: 3,
   samlSSO: false,
   scim: false,
   ldap: false,

backend/src/ee/services/license/license-types.ts

@@ -40,6 +40,8 @@ export type TFeatureSet = {
   customAlerts: false;
   auditLogs: false;
   auditLogsRetentionDays: 0;
+  auditLogStreams: false;
+  auditLogStreamLimit: 3;
   samlSSO: false;
   scim: false;
   ldap: false;