Merge branch 'main' into ENG-3586

2026-10-15 22:46:24 +00:00 · 2025-09-05 18:59:38 -04:00
parent 3858434135 55e89631aa
commit 48b9aeb0d1
306 changed files with 16755 additions and 7458 deletions
@@ -1,215 +0,0 @@
-import { z } from "zod";
-
-import { AUDIT_LOG_STREAMS } from "@app/lib/api-docs";
-import { readLimit } from "@app/server/config/rateLimiter";
-import { verifyAuth } from "@app/server/plugins/auth/verify-auth";
-import { SanitizedAuditLogStreamSchema } from "@app/server/routes/sanitizedSchemas";
-import { AuthMode } from "@app/services/auth/auth-type";
-
-export const registerAuditLogStreamRouter = async (server: FastifyZodProvider) => {
-  server.route({
-    method: "POST",
-    url: "/",
-    config: {
-      rateLimit: readLimit
-    },
-    schema: {
-      description: "Create an Audit Log Stream.",
-      security: [
-        {
-          bearerAuth: []
-        }
-      ],
-      body: z.object({
-        url: z.string().min(1).describe(AUDIT_LOG_STREAMS.CREATE.url),
-        headers: z
-          .object({
-            key: z.string().min(1).trim().describe(AUDIT_LOG_STREAMS.CREATE.headers.key),
-            value: z.string().min(1).trim().describe(AUDIT_LOG_STREAMS.CREATE.headers.value)
-          })
-          .describe(AUDIT_LOG_STREAMS.CREATE.headers.desc)
-          .array()
-          .optional()
-      }),
-      response: {
-        200: z.object({
-          auditLogStream: SanitizedAuditLogStreamSchema
-        })
-      }
-    },
-    onRequest: verifyAuth([AuthMode.JWT, AuthMode.IDENTITY_ACCESS_TOKEN]),
-    handler: async (req) => {
-      const auditLogStream = await server.services.auditLogStream.create({
-        actorId: req.permission.id,
-        actor: req.permission.type,
-        actorOrgId: req.permission.orgId,
-        actorAuthMethod: req.permission.authMethod,
-        url: req.body.url,
-        headers: req.body.headers
-      });
-
-      return { auditLogStream };
-    }
-  });
-
-  server.route({
-    method: "PATCH",
-    url: "/:id",
-    config: {
-      rateLimit: readLimit
-    },
-    schema: {
-      description: "Update an Audit Log Stream by ID.",
-      security: [
-        {
-          bearerAuth: []
-        }
-      ],
-      params: z.object({
-        id: z.string().describe(AUDIT_LOG_STREAMS.UPDATE.id)
-      }),
-      body: z.object({
-        url: z.string().optional().describe(AUDIT_LOG_STREAMS.UPDATE.url),
-        headers: z
-          .object({
-            key: z.string().min(1).trim().describe(AUDIT_LOG_STREAMS.UPDATE.headers.key),
-            value: z.string().min(1).trim().describe(AUDIT_LOG_STREAMS.UPDATE.headers.value)
-          })
-          .describe(AUDIT_LOG_STREAMS.UPDATE.headers.desc)
-          .array()
-          .optional()
-      }),
-      response: {
-        200: z.object({
-          auditLogStream: SanitizedAuditLogStreamSchema
-        })
-      }
-    },
-    onRequest: verifyAuth([AuthMode.JWT, AuthMode.IDENTITY_ACCESS_TOKEN]),
-    handler: async (req) => {
-      const auditLogStream = await server.services.auditLogStream.updateById({
-        actorId: req.permission.id,
-        actor: req.permission.type,
-        actorOrgId: req.permission.orgId,
-        actorAuthMethod: req.permission.authMethod,
-        id: req.params.id,
-        url: req.body.url,
-        headers: req.body.headers
-      });
-
-      return { auditLogStream };
-    }
-  });
-
-  server.route({
-    method: "DELETE",
-    url: "/:id",
-    config: {
-      rateLimit: readLimit
-    },
-    schema: {
-      description: "Delete an Audit Log Stream by ID.",
-      security: [
-        {
-          bearerAuth: []
-        }
-      ],
-      params: z.object({
-        id: z.string().describe(AUDIT_LOG_STREAMS.DELETE.id)
-      }),
-      response: {
-        200: z.object({
-          auditLogStream: SanitizedAuditLogStreamSchema
-        })
-      }
-    },
-    onRequest: verifyAuth([AuthMode.JWT, AuthMode.IDENTITY_ACCESS_TOKEN]),
-    handler: async (req) => {
-      const auditLogStream = await server.services.auditLogStream.deleteById({
-        actorId: req.permission.id,
-        actor: req.permission.type,
-        actorOrgId: req.permission.orgId,
-        actorAuthMethod: req.permission.authMethod,
-        id: req.params.id
-      });
-
-      return { auditLogStream };
-    }
-  });
-
-  server.route({
-    method: "GET",
-    url: "/:id",
-    config: {
-      rateLimit: readLimit
-    },
-    schema: {
-      description: "Get an Audit Log Stream by ID.",
-      security: [
-        {
-          bearerAuth: []
-        }
-      ],
-      params: z.object({
-        id: z.string().describe(AUDIT_LOG_STREAMS.GET_BY_ID.id)
-      }),
-      response: {
-        200: z.object({
-          auditLogStream: SanitizedAuditLogStreamSchema.extend({
-            headers: z
-              .object({
-                key: z.string(),
-                value: z.string()
-              })
-              .array()
-              .optional()
-          })
-        })
-      }
-    },
-    onRequest: verifyAuth([AuthMode.JWT, AuthMode.IDENTITY_ACCESS_TOKEN]),
-    handler: async (req) => {
-      const auditLogStream = await server.services.auditLogStream.getById({
-        actorId: req.permission.id,
-        actor: req.permission.type,
-        actorOrgId: req.permission.orgId,
-        actorAuthMethod: req.permission.authMethod,
-        id: req.params.id
-      });
-
-      return { auditLogStream };
-    }
-  });
-
-  server.route({
-    method: "GET",
-    url: "/",
-    config: {
-      rateLimit: readLimit
-    },
-    schema: {
-      description: "List Audit Log Streams.",
-      security: [
-        {
-          bearerAuth: []
-        }
-      ],
-      response: {
-        200: z.object({
-          auditLogStreams: SanitizedAuditLogStreamSchema.array()
-        })
-      }
-    },
-    onRequest: verifyAuth([AuthMode.JWT, AuthMode.IDENTITY_ACCESS_TOKEN]),
-    handler: async (req) => {
-      const auditLogStreams = await server.services.auditLogStream.list({
-        actorId: req.permission.id,
-        actor: req.permission.type,
-        actorOrgId: req.permission.orgId,
-        actorAuthMethod: req.permission.authMethod
-      });
-
-      return { auditLogStreams };
-    }
-  });
-};
@@ -0,0 +1,142 @@
+import { z } from "zod";
+
+import { LogProvider } from "@app/ee/services/audit-log-stream/audit-log-stream-enums";
+import { TAuditLogStream } from "@app/ee/services/audit-log-stream/audit-log-stream-types";
+import { readLimit, writeLimit } from "@app/server/config/rateLimiter";
+import { verifyAuth } from "@app/server/plugins/auth/verify-auth";
+import { AuthMode } from "@app/services/auth/auth-type";
+
+export const registerAuditLogStreamEndpoints = <T extends TAuditLogStream>({
+  server,
+  provider,
+  createSchema,
+  updateSchema,
+  sanitizedResponseSchema
+}: {
+  server: FastifyZodProvider;
+  provider: LogProvider;
+  createSchema: z.ZodType<{
+    credentials: T["credentials"];
+  }>;
+  updateSchema: z.ZodType<{
+    credentials: T["credentials"];
+  }>;
+  sanitizedResponseSchema: z.ZodTypeAny;
+}) => {
+  server.route({
+    method: "GET",
+    url: "/:logStreamId",
+    config: {
+      rateLimit: readLimit
+    },
+    schema: {
+      params: z.object({
+        logStreamId: z.string().uuid()
+      }),
+      response: {
+        200: z.object({
+          auditLogStream: sanitizedResponseSchema
+        })
+      }
+    },
+    onRequest: verifyAuth([AuthMode.JWT, AuthMode.IDENTITY_ACCESS_TOKEN]),
+    handler: async (req) => {
+      const { logStreamId } = req.params;
+
+      const auditLogStream = await server.services.auditLogStream.getById(logStreamId, provider, req.permission);
+
+      return { auditLogStream };
+    }
+  });
+
+  server.route({
+    method: "POST",
+    url: "/",
+    config: {
+      rateLimit: writeLimit
+    },
+    schema: {
+      body: createSchema,
+      response: {
+        200: z.object({
+          auditLogStream: sanitizedResponseSchema
+        })
+      }
+    },
+    onRequest: verifyAuth([AuthMode.JWT, AuthMode.IDENTITY_ACCESS_TOKEN]),
+    handler: async (req) => {
+      const { credentials } = req.body;
+
+      const auditLogStream = await server.services.auditLogStream.create(
+        {
+          provider,
+          credentials
+        },
+        req.permission
+      );
+
+      return { auditLogStream };
+    }
+  });
+
+  server.route({
+    method: "PATCH",
+    url: "/:logStreamId",
+    config: {
+      rateLimit: writeLimit
+    },
+    schema: {
+      params: z.object({
+        logStreamId: z.string().uuid()
+      }),
+      body: updateSchema,
+      response: {
+        200: z.object({
+          auditLogStream: sanitizedResponseSchema
+        })
+      }
+    },
+    onRequest: verifyAuth([AuthMode.JWT, AuthMode.IDENTITY_ACCESS_TOKEN]),
+    handler: async (req) => {
+      const { logStreamId } = req.params;
+      const { credentials } = req.body;
+
+      const auditLogStream = await server.services.auditLogStream.updateById(
+        {
+          logStreamId,
+          provider,
+          credentials
+        },
+        req.permission
+      );
+
+      return { auditLogStream };
+    }
+  });
+
+  server.route({
+    method: "DELETE",
+    url: "/:logStreamId",
+    config: {
+      rateLimit: writeLimit
+    },
+    schema: {
+      params: z.object({
+        logStreamId: z.string().uuid()
+      }),
+      response: {
+        200: z.object({
+          auditLogStream: sanitizedResponseSchema
+        })
+      }
+    },
+    onRequest: verifyAuth([AuthMode.JWT, AuthMode.IDENTITY_ACCESS_TOKEN]),
+    handler: async (req) => {
+      const { logStreamId } = req.params;
+
+      const auditLogStream = await server.services.auditLogStream.deleteById(logStreamId, provider, req.permission);
+
+      return { auditLogStream };
+    }
+  });
+};
@@ -0,0 +1,73 @@
+import { z } from "zod";
+
+import {
+  CustomProviderListItemSchema,
+  SanitizedCustomProviderSchema
+} from "@app/ee/services/audit-log-stream/custom/custom-provider-schemas";
+import {
+  DatadogProviderListItemSchema,
+  SanitizedDatadogProviderSchema
+} from "@app/ee/services/audit-log-stream/datadog/datadog-provider-schemas";
+import {
+  SanitizedSplunkProviderSchema,
+  SplunkProviderListItemSchema
+} from "@app/ee/services/audit-log-stream/splunk/splunk-provider-schemas";
+import { readLimit } from "@app/server/config/rateLimiter";
+import { verifyAuth } from "@app/server/plugins/auth/verify-auth";
+import { AuthMode } from "@app/services/auth/auth-type";
+
+const SanitizedAuditLogStreamSchema = z.union([
+  SanitizedCustomProviderSchema,
+  SanitizedDatadogProviderSchema,
+  SanitizedSplunkProviderSchema
+]);
+
+const ProviderOptionsSchema = z.discriminatedUnion("provider", [
+  CustomProviderListItemSchema,
+  DatadogProviderListItemSchema,
+  SplunkProviderListItemSchema
+]);
+
+export const registerAuditLogStreamRouter = async (server: FastifyZodProvider) => {
+  server.route({
+    method: "GET",
+    url: "/options",
+    config: {
+      rateLimit: readLimit
+    },
+    schema: {
+      response: {
+        200: z.object({
+          providerOptions: ProviderOptionsSchema.array()
+        })
+      }
+    },
+    onRequest: verifyAuth([AuthMode.JWT, AuthMode.IDENTITY_ACCESS_TOKEN]),
+    handler: () => {
+      const providerOptions = server.services.auditLogStream.listProviderOptions();
+
+      return { providerOptions };
+    }
+  });
+
+  server.route({
+    method: "GET",
+    url: "/",
+    config: {
+      rateLimit: readLimit
+    },
+    schema: {
+      response: {
+        200: z.object({
+          auditLogStreams: SanitizedAuditLogStreamSchema.array()
+        })
+      }
+    },
+    onRequest: verifyAuth([AuthMode.JWT, AuthMode.IDENTITY_ACCESS_TOKEN]),
+    handler: async (req) => {
+      const auditLogStreams = await server.services.auditLogStream.list(req.permission);
+
+      return { auditLogStreams };
+    }
+  });
+};
@@ -0,0 +1,51 @@
+import { LogProvider } from "@app/ee/services/audit-log-stream/audit-log-stream-enums";
+import {
+  CreateCustomProviderLogStreamSchema,
+  SanitizedCustomProviderSchema,
+  UpdateCustomProviderLogStreamSchema
+} from "@app/ee/services/audit-log-stream/custom/custom-provider-schemas";
+import {
+  CreateDatadogProviderLogStreamSchema,
+  SanitizedDatadogProviderSchema,
+  UpdateDatadogProviderLogStreamSchema
+} from "@app/ee/services/audit-log-stream/datadog/datadog-provider-schemas";
+import {
+  CreateSplunkProviderLogStreamSchema,
+  SanitizedSplunkProviderSchema,
+  UpdateSplunkProviderLogStreamSchema
+} from "@app/ee/services/audit-log-stream/splunk/splunk-provider-schemas";
+
+import { registerAuditLogStreamEndpoints } from "./audit-log-stream-endpoints";
+
+export * from "./audit-log-stream-router";
+
+export const AUDIT_LOG_STREAM_REGISTER_ROUTER_MAP: Record<LogProvider, (server: FastifyZodProvider) => Promise<void>> =
+  {
+    [LogProvider.Custom]: async (server: FastifyZodProvider) => {
+      registerAuditLogStreamEndpoints({
+        server,
+        provider: LogProvider.Custom,
+        sanitizedResponseSchema: SanitizedCustomProviderSchema,
+        createSchema: CreateCustomProviderLogStreamSchema,
+        updateSchema: UpdateCustomProviderLogStreamSchema
+      });
+    },
+    [LogProvider.Datadog]: async (server: FastifyZodProvider) => {
+      registerAuditLogStreamEndpoints({
+        server,
+        provider: LogProvider.Datadog,
+        sanitizedResponseSchema: SanitizedDatadogProviderSchema,
+        createSchema: CreateDatadogProviderLogStreamSchema,
+        updateSchema: UpdateDatadogProviderLogStreamSchema
+      });
+    },
+    [LogProvider.Splunk]: async (server: FastifyZodProvider) => {
+      registerAuditLogStreamEndpoints({
+        server,
+        provider: LogProvider.Splunk,
+        sanitizedResponseSchema: SanitizedSplunkProviderSchema,
+        createSchema: CreateSplunkProviderLogStreamSchema,
+        updateSchema: UpdateSplunkProviderLogStreamSchema
+      });
+    }
+  };
@@ -126,4 +126,39 @@ export const registerGithubOrgSyncRouter = async (server: FastifyZodProvider) =>
      return { githubOrgSyncConfig };
    }
  });
+
+  server.route({
+    url: "/sync-all-teams",
+    method: "POST",
+    config: {
+      rateLimit: writeLimit
+    },
+    onRequest: verifyAuth([AuthMode.JWT]),
+    schema: {
+      response: {
+        200: z.object({
+          totalUsers: z.number(),
+          errors: z.array(z.string()),
+          createdTeams: z.array(z.string()),
+          updatedTeams: z.array(z.string()),
+          removedMemberships: z.number(),
+          syncDuration: z.number()
+        })
+      }
+    },
+    handler: async (req) => {
+      const result = await server.services.githubOrgSync.syncAllTeams({
+        orgPermission: req.permission
+      });
+
+      return {
+        totalUsers: result.totalUsers,
+        errors: result.errors,
+        createdTeams: result.createdTeams,
+        updatedTeams: result.updatedTeams,
+        removedMemberships: result.removedMemberships,
+        syncDuration: result.syncDuration
+      };
+    }
+  });
 };
@@ -3,7 +3,7 @@ import { registerProjectTemplateRouter } from "@app/ee/routes/v1/project-templat
 import { registerAccessApprovalPolicyRouter } from "./access-approval-policy-router";
 import { registerAccessApprovalRequestRouter } from "./access-approval-request-router";
 import { registerAssumePrivilegeRouter } from "./assume-privilege-router";
-import { registerAuditLogStreamRouter } from "./audit-log-stream-router";
+import { AUDIT_LOG_STREAM_REGISTER_ROUTER_MAP, registerAuditLogStreamRouter } from "./audit-log-stream-routers";
 import { registerCaCrlRouter } from "./certificate-authority-crl-router";
 import { registerDynamicSecretLeaseRouter } from "./dynamic-secret-lease-router";
 import { registerKubernetesDynamicSecretLeaseRouter } from "./dynamic-secret-lease-routers/kubernetes-lease-router";
@@ -114,7 +114,21 @@ export const registerV1EERoutes = async (server: FastifyZodProvider) => {
  await server.register(registerSecretRouter, { prefix: "/secrets" });
  await server.register(registerSecretVersionRouter, { prefix: "/secret" });
  await server.register(registerGroupRouter, { prefix: "/groups" });
-  await server.register(registerAuditLogStreamRouter, { prefix: "/audit-log-streams" });
+
+  await server.register(
+    async (auditLogStreamRouter) => {
+      await auditLogStreamRouter.register(registerAuditLogStreamRouter);
+
+      // Provider-specific endpoints
+      await Promise.all(
+        Object.entries(AUDIT_LOG_STREAM_REGISTER_ROUTER_MAP).map(([provider, router]) =>
+          auditLogStreamRouter.register(router, { prefix: `/${provider}` })
+        )
+      );
+    },
+    { prefix: "/audit-log-streams" }
+  );
+
  await server.register(registerUserAdditionalPrivilegeRouter, { prefix: "/user-project-additional-privilege" });
  await server.register(
    async (privilegeRouter) => {
@@ -0,0 +1,5 @@
+export enum LogProvider {
+  Datadog = "datadog",
+  Splunk = "splunk",
+  Custom = "custom"
+}
@@ -0,0 +1,13 @@
+import { LogProvider } from "./audit-log-stream-enums";
+import { TAuditLogStreamCredentials, TLogStreamFactory } from "./audit-log-stream-types";
+import { CustomProviderFactory } from "./custom/custom-provider-factory";
+import { DatadogProviderFactory } from "./datadog/datadog-provider-factory";
+import { SplunkProviderFactory } from "./splunk/splunk-provider-factory";
+
+type TLogStreamFactoryImplementation = TLogStreamFactory<TAuditLogStreamCredentials>;
+
+export const LOG_STREAM_FACTORY_MAP: Record<LogProvider, TLogStreamFactoryImplementation> = {
+  [LogProvider.Datadog]: DatadogProviderFactory as TLogStreamFactoryImplementation,
+  [LogProvider.Splunk]: SplunkProviderFactory as TLogStreamFactoryImplementation,
+  [LogProvider.Custom]: CustomProviderFactory as TLogStreamFactoryImplementation
+};
@@ -1,21 +1,70 @@
-export function providerSpecificPayload(url: string) {
-  const { hostname } = new URL(url);
+import { TAuditLogStreams } from "@app/db/schemas";
+import { TKmsServiceFactory } from "@app/services/kms/kms-service";
+import { KmsDataKey } from "@app/services/kms/kms-types";

-  const payload: Record<string, string> = {};
+import { TAuditLogStream, TAuditLogStreamCredentials } from "./audit-log-stream-types";
+import { getCustomProviderListItem } from "./custom/custom-provider-fns";
+import { getDatadogProviderListItem } from "./datadog/datadog-provider-fns";
+import { getSplunkProviderListItem } from "./splunk/splunk-provider-fns";

-  switch (hostname) {
-    case "http-intake.logs.datadoghq.com":
-    case "http-intake.logs.us3.datadoghq.com":
-    case "http-intake.logs.us5.datadoghq.com":
-    case "http-intake.logs.datadoghq.eu":
-    case "http-intake.logs.ap1.datadoghq.com":
-    case "http-intake.logs.ddog-gov.com":
-      payload.ddsource = "infisical";
-      payload.service = "audit-logs";
-      break;
-    default:
-      break;
-  }
+export const listProviderOptions = () => {
+  return [getDatadogProviderListItem(), getSplunkProviderListItem(), getCustomProviderListItem()].sort((a, b) =>
+    a.name.localeCompare(b.name)
+  );
+};

-  return payload;
-}
+export const encryptLogStreamCredentials = async ({
+  orgId,
+  credentials,
+  kmsService
+}: {
+  orgId: string;
+  credentials: TAuditLogStreamCredentials;
+  kmsService: Pick<TKmsServiceFactory, "createCipherPairWithDataKey">;
+}) => {
+  const { encryptor } = await kmsService.createCipherPairWithDataKey({
+    type: KmsDataKey.Organization,
+    orgId
+  });
+
+  const { cipherTextBlob: encryptedCredentialsBlob } = encryptor({
+    plainText: Buffer.from(JSON.stringify(credentials))
+  });
+
+  return encryptedCredentialsBlob;
+};
+
+export const decryptLogStreamCredentials = async ({
+  orgId,
+  encryptedCredentials,
+  kmsService
+}: {
+  orgId: string;
+  encryptedCredentials: Buffer;
+  kmsService: Pick<TKmsServiceFactory, "createCipherPairWithDataKey">;
+}) => {
+  const { decryptor } = await kmsService.createCipherPairWithDataKey({
+    type: KmsDataKey.Organization,
+    orgId
+  });
+
+  const decryptedPlainTextBlob = decryptor({
+    cipherTextBlob: encryptedCredentials
+  });
+
+  return JSON.parse(decryptedPlainTextBlob.toString()) as TAuditLogStreamCredentials;
+};
+
+export const decryptLogStream = async (
+  logStream: TAuditLogStreams,
+  kmsService: Pick<TKmsServiceFactory, "createCipherPairWithDataKey">
+) => {
+  return {
+    ...logStream,
+    credentials: await decryptLogStreamCredentials({
+      encryptedCredentials: logStream.encryptedCredentials,
+      orgId: logStream.orgId,
+      kmsService
+    })
+  } as TAuditLogStream;
+};
@@ -0,0 +1,14 @@
+import { AuditLogStreamsSchema } from "@app/db/schemas";
+
+export const BaseProviderSchema = AuditLogStreamsSchema.omit({
+  encryptedCredentials: true,
+  provider: true,
+
+  // Old "archived" values
+  encryptedHeadersAlgorithm: true,
+  encryptedHeadersCiphertext: true,
+  encryptedHeadersIV: true,
+  encryptedHeadersKeyEncoding: true,
+  encryptedHeadersTag: true,
+  url: true
+});
@@ -1,242 +1,252 @@
 import { ForbiddenError } from "@casl/ability";
-import { RawAxiosRequestHeaders } from "axios";
+import { AxiosError } from "axios";

-import { SecretKeyEncoding } from "@app/db/schemas";
-import { getConfig } from "@app/lib/config/env";
-import { request } from "@app/lib/config/request";
-import { crypto } from "@app/lib/crypto/cryptography";
-import { BadRequestError, NotFoundError, UnauthorizedError } from "@app/lib/errors";
-import { blockLocalAndPrivateIpAddresses } from "@app/lib/validator";
+import { TAuditLogs } from "@app/db/schemas";
+import {
+  decryptLogStream,
+  decryptLogStreamCredentials,
+  encryptLogStreamCredentials,
+  listProviderOptions
+} from "@app/ee/services/audit-log-stream/audit-log-stream-fns";
+import { BadRequestError, NotFoundError } from "@app/lib/errors";
+import { logger } from "@app/lib/logger";
+import { OrgServiceActor } from "@app/lib/types";
+import { TKmsServiceFactory } from "@app/services/kms/kms-service";

-import { AUDIT_LOG_STREAM_TIMEOUT } from "../audit-log/audit-log-queue";
 import { TLicenseServiceFactory } from "../license/license-service";
 import { OrgPermissionActions, OrgPermissionSubjects } from "../permission/org-permission";
 import { TPermissionServiceFactory } from "../permission/permission-service-types";
 import { TAuditLogStreamDALFactory } from "./audit-log-stream-dal";
-import { providerSpecificPayload } from "./audit-log-stream-fns";
-import { LogStreamHeaders, TAuditLogStreamServiceFactory } from "./audit-log-stream-types";
+import { LogProvider } from "./audit-log-stream-enums";
+import { LOG_STREAM_FACTORY_MAP } from "./audit-log-stream-factory";
+import { TAuditLogStream, TCreateAuditLogStreamDTO, TUpdateAuditLogStreamDTO } from "./audit-log-stream-types";
+import { TCustomProviderCredentials } from "./custom/custom-provider-types";

-type TAuditLogStreamServiceFactoryDep = {
+export type TAuditLogStreamServiceFactoryDep = {
  auditLogStreamDAL: TAuditLogStreamDALFactory;
  permissionService: Pick<TPermissionServiceFactory, "getOrgPermission">;
  licenseService: Pick<TLicenseServiceFactory, "getPlan">;
+  kmsService: Pick<TKmsServiceFactory, "createCipherPairWithDataKey">;
 };

+export type TAuditLogStreamServiceFactory = ReturnType<typeof auditLogStreamServiceFactory>;
+
 export const auditLogStreamServiceFactory = ({
  auditLogStreamDAL,
  permissionService,
-  licenseService
-}: TAuditLogStreamServiceFactoryDep): TAuditLogStreamServiceFactory => {
-  const create: TAuditLogStreamServiceFactory["create"] = async ({
-    url,
-    actor,
-    headers = [],
-    actorId,
-    actorOrgId,
-    actorAuthMethod
-  }) => {
-    if (!actorOrgId) throw new UnauthorizedError({ message: "No organization ID attached to authentication token" });
-
-    const plan = await licenseService.getPlan(actorOrgId);
+  licenseService,
+  kmsService
+}: TAuditLogStreamServiceFactoryDep) => {
+  const create = async ({ provider, credentials }: TCreateAuditLogStreamDTO, actor: OrgServiceActor) => {
+    const plan = await licenseService.getPlan(actor.orgId);
    if (!plan.auditLogStreams) {
      throw new BadRequestError({
-        message: "Failed to create audit log streams due to plan restriction. Upgrade plan to create group."
+        message: "Failed to create Audit Log Stream: Plan restriction. Upgrade plan to continue."
      });
    }

    const { permission } = await permissionService.getOrgPermission(
-      actor,
-      actorId,
-      actorOrgId,
-      actorAuthMethod,
-      actorOrgId
+      actor.type,
+      actor.id,
+      actor.orgId,
+      actor.authMethod,
+      actor.orgId
    );
+
    ForbiddenError.from(permission).throwUnlessCan(OrgPermissionActions.Create, OrgPermissionSubjects.Settings);

-    const appCfg = getConfig();
-    if (appCfg.isCloud) await blockLocalAndPrivateIpAddresses(url);
-
-    const totalStreams = await auditLogStreamDAL.find({ orgId: actorOrgId });
+    const totalStreams = await auditLogStreamDAL.find({ orgId: actor.orgId });
    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."
+        message: "Failed to create Audit Log Stream: Plan limit reached. Contact Infisical to increase quota."
      });
    }

-    // testing connection first
-    const streamHeaders: RawAxiosRequestHeaders = { "Content-Type": "application/json" };
-    if (headers.length)
-      headers.forEach(({ key, value }) => {
-        streamHeaders[key] = value;
-      });
+    const factory = LOG_STREAM_FACTORY_MAP[provider]();
+    const validatedCredentials = await factory.validateCredentials({ credentials });

-    await request
-      .post(
-        url,
-        { ...providerSpecificPayload(url), ping: "ok" },
-        {
-          headers: streamHeaders,
-          // request timeout
-          timeout: AUDIT_LOG_STREAM_TIMEOUT,
-          // connection timeout
-          signal: AbortSignal.timeout(AUDIT_LOG_STREAM_TIMEOUT)
-        }
-      )
-      .catch((err) => {
-        throw new BadRequestError({ message: `Failed to connect with upstream source: ${(err as Error)?.message}` });
-      });
+    const encryptedCredentials = await encryptLogStreamCredentials({
+      credentials: validatedCredentials,
+      orgId: actor.orgId,
+      kmsService
+    });

-    const encryptedHeaders = headers
-      ? crypto.encryption().symmetric().encryptWithRootEncryptionKey(JSON.stringify(headers))
-      : undefined;
    const logStream = await auditLogStreamDAL.create({
-      orgId: actorOrgId,
-      url,
-      ...(encryptedHeaders
-        ? {
-            encryptedHeadersCiphertext: encryptedHeaders.ciphertext,
-            encryptedHeadersIV: encryptedHeaders.iv,
-            encryptedHeadersTag: encryptedHeaders.tag,
-            encryptedHeadersAlgorithm: encryptedHeaders.algorithm,
-            encryptedHeadersKeyEncoding: encryptedHeaders.encoding
-          }
-        : {})
+      orgId: actor.orgId,
+      provider,
+      encryptedCredentials
    });
-    return logStream;
+
+    return { ...logStream, credentials: validatedCredentials } as TAuditLogStream;
  };

-  const updateById: TAuditLogStreamServiceFactory["updateById"] = async ({
-    id,
-    url,
-    actor,
-    headers = [],
-    actorId,
-    actorOrgId,
-    actorAuthMethod
-  }) => {
-    if (!actorOrgId) throw new UnauthorizedError({ message: "No organization ID attached to authentication token" });
-
-    const plan = await licenseService.getPlan(actorOrgId);
-    if (!plan.auditLogStreams)
+  const updateById = async (
+    { logStreamId, provider, credentials }: TUpdateAuditLogStreamDTO,
+    actor: OrgServiceActor
+  ) => {
+    const plan = await licenseService.getPlan(actor.orgId);
+    if (!plan.auditLogStreams) {
      throw new BadRequestError({
-        message: "Failed to update audit log streams due to plan restriction. Upgrade plan to create group."
+        message: "Failed to update Audit Log Stream: Plan restriction. Upgrade plan to continue."
      });
+    }

-    const logStream = await auditLogStreamDAL.findById(id);
-    if (!logStream) throw new NotFoundError({ message: `Audit log stream with ID '${id}' not found` });
+    const logStream = await auditLogStreamDAL.findById(logStreamId);
+    if (!logStream) throw new NotFoundError({ message: `Audit Log Stream with ID '${logStreamId}' not found` });
+
+    const { permission } = await permissionService.getOrgPermission(
+      actor.type,
+      actor.id,
+      actor.orgId,
+      actor.authMethod,
+      logStream.orgId
+    );

-    const { orgId } = logStream;
-    const { permission } = await permissionService.getOrgPermission(actor, actorId, orgId, actorAuthMethod, actorOrgId);
    ForbiddenError.from(permission).throwUnlessCan(OrgPermissionActions.Edit, OrgPermissionSubjects.Settings);
-    const appCfg = getConfig();
-    if (url && appCfg.isCloud) await blockLocalAndPrivateIpAddresses(url);

-    // testing connection first
-    const streamHeaders: RawAxiosRequestHeaders = { "Content-Type": "application/json" };
-    if (headers.length)
-      headers.forEach(({ key, value }) => {
-        streamHeaders[key] = value;
-      });
+    const finalCredentials = { ...credentials };

-    await request
-      .post(
-        url || logStream.url,
-        { ...providerSpecificPayload(url || logStream.url), ping: "ok" },
-        {
-          headers: streamHeaders,
-          // request timeout
-          timeout: AUDIT_LOG_STREAM_TIMEOUT,
-          // connection timeout
-          signal: AbortSignal.timeout(AUDIT_LOG_STREAM_TIMEOUT)
-        }
-      )
-      .catch((err) => {
-        throw new Error(`Failed to connect with the source ${(err as Error)?.message}`);
-      });
+    // For the "Custom" provider, we must handle masked header values ('******').
+    // These are placeholders from the frontend for secrets that haven't been changed.
+    // We need to replace them with the original, unmasked values from the database.
+    if (
+      provider === LogProvider.Custom &&
+      "headers" in finalCredentials &&
+      Array.isArray(finalCredentials.headers) &&
+      finalCredentials.headers.some((header) => header.value === "******")
+    ) {
+      const decryptedOldCredentials = (await decryptLogStreamCredentials({
+        encryptedCredentials: logStream.encryptedCredentials,
+        orgId: logStream.orgId,
+        kmsService
+      })) as TCustomProviderCredentials;

-    const encryptedHeaders = headers
-      ? crypto.encryption().symmetric().encryptWithRootEncryptionKey(JSON.stringify(headers))
-      : undefined;
-    const updatedLogStream = await auditLogStreamDAL.updateById(id, {
-      url,
-      ...(encryptedHeaders
-        ? {
-            encryptedHeadersCiphertext: encryptedHeaders.ciphertext,
-            encryptedHeadersIV: encryptedHeaders.iv,
-            encryptedHeadersTag: encryptedHeaders.tag,
-            encryptedHeadersAlgorithm: encryptedHeaders.algorithm,
-            encryptedHeadersKeyEncoding: encryptedHeaders.encoding
+      const oldHeadersMap = decryptedOldCredentials.headers.reduce<Record<string, string>>((acc, header) => {
+        acc[header.key] = header.value;
+        return acc;
+      }, {});
+
+      const finalHeaders: { key: string; value: string }[] = [];
+      for (const header of finalCredentials.headers) {
+        if (header.value === "******") {
+          const oldValue = oldHeadersMap[header.key];
+          if (oldValue) {
+            finalHeaders.push({ key: header.key, value: oldValue });
          }
-        : {})
+        } else {
+          finalHeaders.push(header);
+        }
+      }
+      finalCredentials.headers = finalHeaders;
+    }
+
+    const factory = LOG_STREAM_FACTORY_MAP[provider]();
+    const validatedCredentials = await factory.validateCredentials({ credentials: finalCredentials });
+
+    const encryptedCredentials = await encryptLogStreamCredentials({
+      credentials: validatedCredentials,
+      orgId: actor.orgId,
+      kmsService
    });
-    return updatedLogStream;
+
+    const updatedLogStream = await auditLogStreamDAL.updateById(logStreamId, {
+      encryptedCredentials
+    });
+
+    return { ...updatedLogStream, credentials: validatedCredentials } as TAuditLogStream;
  };

-  const deleteById: TAuditLogStreamServiceFactory["deleteById"] = async ({
-    id,
-    actor,
-    actorId,
-    actorOrgId,
-    actorAuthMethod
-  }) => {
-    if (!actorOrgId) throw new UnauthorizedError({ message: "No organization ID attached to authentication token" });
+  const deleteById = async (logStreamId: string, provider: LogProvider, actor: OrgServiceActor) => {
+    const logStream = await auditLogStreamDAL.findById(logStreamId);
+    if (!logStream) throw new NotFoundError({ message: `Audit Log Stream with ID '${logStreamId}' not found` });

-    const logStream = await auditLogStreamDAL.findById(id);
-    if (!logStream) throw new NotFoundError({ message: `Audit log stream with ID '${id}' not found` });
+    const { permission } = await permissionService.getOrgPermission(
+      actor.type,
+      actor.id,
+      actor.orgId,
+      actor.authMethod,
+      logStream.orgId
+    );

-    const { orgId } = logStream;
-    const { permission } = await permissionService.getOrgPermission(actor, actorId, orgId, actorAuthMethod, actorOrgId);
    ForbiddenError.from(permission).throwUnlessCan(OrgPermissionActions.Delete, OrgPermissionSubjects.Settings);

-    const deletedLogStream = await auditLogStreamDAL.deleteById(id);
-    return deletedLogStream;
+    if (logStream.provider !== provider) {
+      throw new BadRequestError({
+        message: `Audit Log Stream with ID '${logStreamId}' is not for provider '${provider}'`
+      });
+    }
+
+    const deletedLogStream = await auditLogStreamDAL.deleteById(logStreamId);
+
+    return decryptLogStream(deletedLogStream, kmsService);
  };

-  const getById: TAuditLogStreamServiceFactory["getById"] = async ({
-    id,
-    actor,
-    actorId,
-    actorOrgId,
-    actorAuthMethod
-  }) => {
-    const logStream = await auditLogStreamDAL.findById(id);
-    if (!logStream) throw new NotFoundError({ message: `Audit log stream with ID '${id}' not found` });
+  const getById = async (logStreamId: string, provider: LogProvider, actor: OrgServiceActor) => {
+    const logStream = await auditLogStreamDAL.findById(logStreamId);

-    const { orgId } = logStream;
-    const { permission } = await permissionService.getOrgPermission(actor, actorId, orgId, actorAuthMethod, actorOrgId);
-    ForbiddenError.from(permission).throwUnlessCan(OrgPermissionActions.Read, OrgPermissionSubjects.Settings);
+    if (!logStream) throw new NotFoundError({ message: `Audit log stream with ID '${logStreamId}' not found` });

-    const headers =
-      logStream?.encryptedHeadersCiphertext && logStream?.encryptedHeadersIV && logStream?.encryptedHeadersTag
-        ? (JSON.parse(
-            crypto
-              .encryption()
-              .symmetric()
-              .decryptWithRootEncryptionKey({
-                tag: logStream.encryptedHeadersTag,
-                iv: logStream.encryptedHeadersIV,
-                ciphertext: logStream.encryptedHeadersCiphertext,
-                keyEncoding: logStream.encryptedHeadersKeyEncoding as SecretKeyEncoding
-              })
-          ) as LogStreamHeaders[])
-        : undefined;
-
-    return { ...logStream, headers };
-  };
-
-  const list: TAuditLogStreamServiceFactory["list"] = async ({ actor, actorId, actorOrgId, actorAuthMethod }) => {
    const { permission } = await permissionService.getOrgPermission(
-      actor,
-      actorId,
-      actorOrgId,
-      actorAuthMethod,
-      actorOrgId
+      actor.type,
+      actor.id,
+      logStream.orgId,
+      actor.authMethod,
+      actor.orgId
    );
+
    ForbiddenError.from(permission).throwUnlessCan(OrgPermissionActions.Read, OrgPermissionSubjects.Settings);

-    const logStreams = await auditLogStreamDAL.find({ orgId: actorOrgId });
-    return logStreams;
+    if (logStream.provider !== provider) {
+      throw new BadRequestError({
+        message: `Audit Log Stream with ID '${logStreamId}' is not for provider '${provider}'`
+      });
+    }
+
+    return decryptLogStream(logStream, kmsService);
+  };
+
+  const list = async (actor: OrgServiceActor) => {
+    const { permission } = await permissionService.getOrgPermission(
+      actor.type,
+      actor.id,
+      actor.orgId,
+      actor.authMethod,
+      actor.orgId
+    );
+
+    ForbiddenError.from(permission).throwUnlessCan(OrgPermissionActions.Read, OrgPermissionSubjects.Settings);
+
+    const logStreams = await auditLogStreamDAL.find({ orgId: actor.orgId });
+
+    return Promise.all(logStreams.map((stream) => decryptLogStream(stream, kmsService)));
+  };
+
+  const streamLog = async (orgId: string, auditLog: TAuditLogs) => {
+    const logStreams = await auditLogStreamDAL.find({ orgId });
+    await Promise.allSettled(
+      logStreams.map(async ({ provider, encryptedCredentials }) => {
+        const credentials = await decryptLogStreamCredentials({
+          encryptedCredentials,
+          orgId,
+          kmsService
+        });
+
+        const factory = LOG_STREAM_FACTORY_MAP[provider as LogProvider]();
+
+        try {
+          await factory.streamLog({
+            credentials,
+            auditLog
+          });
+        } catch (error) {
+          logger.error(
+            error,
+            `Failed to stream audit log [auditLogId=${auditLog.id}] [provider=${provider}] [orgId=${orgId}]${error instanceof AxiosError ? `: ${error.message}` : ""}`
+          );
+          throw error;
+        }
+      })
+    );
  };

  return {
@@ -244,6 +254,8 @@ export const auditLogStreamServiceFactory = ({
    updateById,
    deleteById,
    getById,
-    list
+    list,
+    listProviderOptions,
+    streamLog
  };
 };
@@ -1,48 +1,38 @@
-import { TAuditLogStreams } from "@app/db/schemas";
-import { TOrgPermission } from "@app/lib/types";
+import { TAuditLogs } from "@app/db/schemas";

-export type LogStreamHeaders = {
-  key: string;
-  value: string;
+import { LogProvider } from "./audit-log-stream-enums";
+import { TCustomProvider, TCustomProviderCredentials } from "./custom/custom-provider-types";
+import { TDatadogProvider, TDatadogProviderCredentials } from "./datadog/datadog-provider-types";
+import { TSplunkProvider, TSplunkProviderCredentials } from "./splunk/splunk-provider-types";
+
+export type TAuditLogStream = TDatadogProvider | TSplunkProvider | TCustomProvider;
+
+export type TAuditLogStreamCredentials =
+  | TDatadogProviderCredentials
+  | TSplunkProviderCredentials
+  | TCustomProviderCredentials;
+
+export type TCreateAuditLogStreamDTO = {
+  provider: LogProvider;
+  credentials: TAuditLogStreamCredentials;
 };

-export type TCreateAuditLogStreamDTO = Omit<TOrgPermission, "orgId"> & {
-  url: string;
-  headers?: LogStreamHeaders[];
+export type TUpdateAuditLogStreamDTO = {
+  logStreamId: string;
+  provider: LogProvider;
+  credentials: TAuditLogStreamCredentials;
 };

-export type TUpdateAuditLogStreamDTO = Omit<TOrgPermission, "orgId"> & {
-  id: string;
-  url?: string;
-  headers?: LogStreamHeaders[];
-};
+export type TLogStreamFactoryValidateCredentials<C extends TAuditLogStreamCredentials> = (input: {
+  credentials: C;
+}) => Promise<C>;

-export type TDeleteAuditLogStreamDTO = Omit<TOrgPermission, "orgId"> & {
-  id: string;
-};
+export type TLogStreamFactoryStreamLog<C extends TAuditLogStreamCredentials> = (input: {
+  credentials: C;
+  auditLog: TAuditLogs;
+}) => Promise<void>;

-export type TListAuditLogStreamDTO = Omit<TOrgPermission, "orgId">;
-
-export type TGetDetailsAuditLogStreamDTO = Omit<TOrgPermission, "orgId"> & {
-  id: string;
-};
-
-export type TAuditLogStreamServiceFactory = {
-  create: (arg: TCreateAuditLogStreamDTO) => Promise<TAuditLogStreams>;
-  updateById: (arg: TUpdateAuditLogStreamDTO) => Promise<TAuditLogStreams>;
-  deleteById: (arg: TDeleteAuditLogStreamDTO) => Promise<TAuditLogStreams>;
-  getById: (arg: TGetDetailsAuditLogStreamDTO) => Promise<{
-    headers: LogStreamHeaders[] | undefined;
-    orgId: string;
-    url: string;
-    id: string;
-    createdAt: Date;
-    updatedAt: Date;
-    encryptedHeadersCiphertext?: string | null | undefined;
-    encryptedHeadersIV?: string | null | undefined;
-    encryptedHeadersTag?: string | null | undefined;
-    encryptedHeadersAlgorithm?: string | null | undefined;
-    encryptedHeadersKeyEncoding?: string | null | undefined;
-  }>;
-  list: (arg: TListAuditLogStreamDTO) => Promise<TAuditLogStreams[]>;
+export type TLogStreamFactory<C extends TAuditLogStreamCredentials> = () => {
+  validateCredentials: TLogStreamFactoryValidateCredentials<C>;
+  streamLog: TLogStreamFactoryStreamLog<C>;
 };
@@ -0,0 +1,67 @@
+import { RawAxiosRequestHeaders } from "axios";
+
+import { request } from "@app/lib/config/request";
+import { BadRequestError } from "@app/lib/errors";
+import { blockLocalAndPrivateIpAddresses } from "@app/lib/validator";
+
+import { AUDIT_LOG_STREAM_TIMEOUT } from "../../audit-log/audit-log-queue";
+import { TLogStreamFactoryStreamLog, TLogStreamFactoryValidateCredentials } from "../audit-log-stream-types";
+import { TCustomProviderCredentials } from "./custom-provider-types";
+
+export const CustomProviderFactory = () => {
+  const validateCredentials: TLogStreamFactoryValidateCredentials<TCustomProviderCredentials> = async ({
+    credentials
+  }) => {
+    const { url, headers } = credentials;
+
+    await blockLocalAndPrivateIpAddresses(url);
+
+    const streamHeaders: RawAxiosRequestHeaders = { "Content-Type": "application/json" };
+    if (headers.length) {
+      headers.forEach(({ key, value }) => {
+        streamHeaders[key] = value;
+      });
+    }
+
+    await request
+      .post(
+        url,
+        { ping: "ok" },
+        {
+          headers: streamHeaders,
+          timeout: AUDIT_LOG_STREAM_TIMEOUT,
+          signal: AbortSignal.timeout(AUDIT_LOG_STREAM_TIMEOUT)
+        }
+      )
+      .catch((err) => {
+        throw new BadRequestError({ message: `Failed to connect with upstream source: ${(err as Error)?.message}` });
+      });
+
+    return credentials;
+  };
+
+  const streamLog: TLogStreamFactoryStreamLog<TCustomProviderCredentials> = async ({ credentials, auditLog }) => {
+    const { url, headers } = credentials;
+
+    await blockLocalAndPrivateIpAddresses(url);
+
+    const streamHeaders: RawAxiosRequestHeaders = { "Content-Type": "application/json" };
+
+    if (headers.length) {
+      headers.forEach(({ key, value }) => {
+        streamHeaders[key] = value;
+      });
+    }
+
+    await request.post(url, auditLog, {
+      headers: streamHeaders,
+      timeout: AUDIT_LOG_STREAM_TIMEOUT,
+      signal: AbortSignal.timeout(AUDIT_LOG_STREAM_TIMEOUT)
+    });
+  };
+
+  return {
+    validateCredentials,
+    streamLog
+  };
+};
@@ -0,0 +1,8 @@
+import { LogProvider } from "../audit-log-stream-enums";
+
+export const getCustomProviderListItem = () => {
+  return {
+    name: "Custom" as const,
+    provider: LogProvider.Custom as const
+  };
+};
@@ -0,0 +1,50 @@
+import RE2 from "re2";
+import { z } from "zod";
+
+import { LogProvider } from "../audit-log-stream-enums";
+import { BaseProviderSchema } from "../audit-log-stream-schemas";
+
+export const CustomProviderCredentialsSchema = z.object({
+  url: z.string().url().trim().min(1).max(255),
+  headers: z
+    .object({
+      key: z
+        .string()
+        .min(1)
+        .refine((val) => new RE2(/^[^\n\r]+$/).test(val), "Header keys cannot contain newlines or carriage returns"),
+      value: z
+        .string()
+        .min(1)
+        .refine((val) => new RE2(/^[^\n\r]+$/).test(val), "Header values cannot contain newlines or carriage returns")
+    })
+    .array()
+});
+
+const BaseCustomProviderSchema = BaseProviderSchema.extend({ provider: z.literal(LogProvider.Custom) });
+
+export const CustomProviderSchema = BaseCustomProviderSchema.extend({
+  credentials: CustomProviderCredentialsSchema
+});
+
+export const SanitizedCustomProviderSchema = BaseCustomProviderSchema.extend({
+  credentials: z.object({
+    url: CustomProviderCredentialsSchema.shape.url,
+    // Return header keys and a redacted value
+    headers: CustomProviderCredentialsSchema.shape.headers.transform((headers) =>
+      headers.map((header) => ({ ...header, value: "******" }))
+    )
+  })
+});
+
+export const CustomProviderListItemSchema = z.object({
+  name: z.literal("Custom"),
+  provider: z.literal(LogProvider.Custom)
+});
+
+export const CreateCustomProviderLogStreamSchema = z.object({
+  credentials: CustomProviderCredentialsSchema
+});
+
+export const UpdateCustomProviderLogStreamSchema = z.object({
+  credentials: CustomProviderCredentialsSchema
+});
@@ -0,0 +1,7 @@
+import { z } from "zod";
+
+import { CustomProviderCredentialsSchema, CustomProviderSchema } from "./custom-provider-schemas";
+
+export type TCustomProvider = z.infer<typeof CustomProviderSchema>;
+
+export type TCustomProviderCredentials = z.infer<typeof CustomProviderCredentialsSchema>;
@@ -0,0 +1,67 @@
+import { RawAxiosRequestHeaders } from "axios";
+
+import { getConfig } from "@app/lib/config/env";
+import { request } from "@app/lib/config/request";
+import { BadRequestError } from "@app/lib/errors";
+import { blockLocalAndPrivateIpAddresses } from "@app/lib/validator";
+
+import { AUDIT_LOG_STREAM_TIMEOUT } from "../../audit-log/audit-log-queue";
+import { TLogStreamFactoryStreamLog, TLogStreamFactoryValidateCredentials } from "../audit-log-stream-types";
+import { TDatadogProviderCredentials } from "./datadog-provider-types";
+
+function createPayload(event: Record<string, unknown>) {
+  const appCfg = getConfig();
+
+  const ddtags = [`env:${appCfg.NODE_ENV || "unknown"}`].join(",");
+
+  return {
+    ...event,
+    hostname: new URL(appCfg.SITE_URL || "http://infisical").hostname,
+    ddsource: "infisical",
+    service: "infisical",
+    ddtags
+  };
+}
+
+export const DatadogProviderFactory = () => {
+  const validateCredentials: TLogStreamFactoryValidateCredentials<TDatadogProviderCredentials> = async ({
+    credentials
+  }) => {
+    const { url, token } = credentials;
+
+    await blockLocalAndPrivateIpAddresses(url);
+
+    const streamHeaders: RawAxiosRequestHeaders = { "Content-Type": "application/json", "DD-API-KEY": token };
+
+    await request
+      .post(url, createPayload({ ping: "ok" }), {
+        headers: streamHeaders,
+        timeout: AUDIT_LOG_STREAM_TIMEOUT,
+        signal: AbortSignal.timeout(AUDIT_LOG_STREAM_TIMEOUT)
+      })
+      .catch((err) => {
+        throw new BadRequestError({ message: `Failed to connect with Datadog: ${(err as Error)?.message}` });
+      });
+
+    return credentials;
+  };
+
+  const streamLog: TLogStreamFactoryStreamLog<TDatadogProviderCredentials> = async ({ credentials, auditLog }) => {
+    const { url, token } = credentials;
+
+    await blockLocalAndPrivateIpAddresses(url);
+
+    const streamHeaders: RawAxiosRequestHeaders = { "Content-Type": "application/json", "DD-API-KEY": token };
+
+    await request.post(url, createPayload(auditLog), {
+      headers: streamHeaders,
+      timeout: AUDIT_LOG_STREAM_TIMEOUT,
+      signal: AbortSignal.timeout(AUDIT_LOG_STREAM_TIMEOUT)
+    });
+  };
+
+  return {
+    validateCredentials,
+    streamLog
+  };
+};
@@ -0,0 +1,8 @@
+import { LogProvider } from "../audit-log-stream-enums";
+
+export const getDatadogProviderListItem = () => {
+  return {
+    name: "Datadog" as const,
+    provider: LogProvider.Datadog as const
+  };
+};
@@ -0,0 +1,38 @@
+import RE2 from "re2";
+import { z } from "zod";
+
+import { LogProvider } from "../audit-log-stream-enums";
+import { BaseProviderSchema } from "../audit-log-stream-schemas";
+
+export const DatadogProviderCredentialsSchema = z.object({
+  url: z.string().url().trim().min(1).max(255),
+  token: z
+    .string()
+    .trim()
+    .refine((val) => new RE2(/^[a-fA-F0-9]{32}$/).test(val), "Invalid Datadog API key format")
+});
+
+const BaseDatadogProviderSchema = BaseProviderSchema.extend({ provider: z.literal(LogProvider.Datadog) });
+
+export const DatadogProviderSchema = BaseDatadogProviderSchema.extend({
+  credentials: DatadogProviderCredentialsSchema
+});
+
+export const SanitizedDatadogProviderSchema = BaseDatadogProviderSchema.extend({
+  credentials: DatadogProviderCredentialsSchema.pick({
+    url: true
+  })
+});
+
+export const DatadogProviderListItemSchema = z.object({
+  name: z.literal("Datadog"),
+  provider: z.literal(LogProvider.Datadog)
+});
+
+export const CreateDatadogProviderLogStreamSchema = z.object({
+  credentials: DatadogProviderCredentialsSchema
+});
+
+export const UpdateDatadogProviderLogStreamSchema = z.object({
+  credentials: DatadogProviderCredentialsSchema
+});
@@ -0,0 +1,7 @@
+import { z } from "zod";
+
+import { DatadogProviderCredentialsSchema, DatadogProviderSchema } from "./datadog-provider-schemas";
+
+export type TDatadogProvider = z.infer<typeof DatadogProviderSchema>;
+
+export type TDatadogProviderCredentials = z.infer<typeof DatadogProviderCredentialsSchema>;
@@ -0,0 +1,84 @@
+import { RawAxiosRequestHeaders } from "axios";
+
+import { getConfig } from "@app/lib/config/env";
+import { request } from "@app/lib/config/request";
+import { BadRequestError } from "@app/lib/errors";
+import { blockLocalAndPrivateIpAddresses } from "@app/lib/validator";
+
+import { AUDIT_LOG_STREAM_TIMEOUT } from "../../audit-log/audit-log-queue";
+import { TLogStreamFactoryStreamLog, TLogStreamFactoryValidateCredentials } from "../audit-log-stream-types";
+import { TSplunkProviderCredentials } from "./splunk-provider-types";
+
+function createPayload(event: Record<string, unknown>) {
+  const appCfg = getConfig();
+
+  return {
+    time: Math.floor(Date.now() / 1000),
+    ...(appCfg.SITE_URL && { host: new URL(appCfg.SITE_URL).host }),
+    source: "infisical",
+    sourcetype: "_json",
+    event
+  };
+}
+
+async function createSplunkUrl(hostname: string) {
+  let parsedHostname: string;
+  try {
+    parsedHostname = new URL(`https://${hostname}`).hostname;
+  } catch (error) {
+    throw new BadRequestError({ message: `Invalid Splunk hostname provided: ${(error as Error).message}` });
+  }
+
+  await blockLocalAndPrivateIpAddresses(`https://${parsedHostname}`);
+
+  return `https://${parsedHostname}:8088/services/collector/event`;
+}
+
+export const SplunkProviderFactory = () => {
+  const validateCredentials: TLogStreamFactoryValidateCredentials<TSplunkProviderCredentials> = async ({
+    credentials
+  }) => {
+    const { hostname, token } = credentials;
+
+    const url = await createSplunkUrl(hostname);
+
+    const streamHeaders: RawAxiosRequestHeaders = {
+      "Content-Type": "application/json",
+      Authorization: `Splunk ${token}`
+    };
+
+    await request
+      .post(url, createPayload({ ping: "ok" }), {
+        headers: streamHeaders,
+        timeout: AUDIT_LOG_STREAM_TIMEOUT,
+        signal: AbortSignal.timeout(AUDIT_LOG_STREAM_TIMEOUT)
+      })
+      .catch((err) => {
+        throw new BadRequestError({ message: `Failed to connect with Splunk: ${(err as Error)?.message}` });
+      });
+
+    return credentials;
+  };
+
+  const streamLog: TLogStreamFactoryStreamLog<TSplunkProviderCredentials> = async ({ credentials, auditLog }) => {
+    const { hostname, token } = credentials;
+
+    const url = await createSplunkUrl(hostname);
+
+    const streamHeaders: RawAxiosRequestHeaders = {
+      "Content-Type": "application/json",
+      Authorization: `Splunk ${token}`
+    };
+
+    await request.post(url, createPayload(auditLog), {
+      headers: streamHeaders,
+      timeout: AUDIT_LOG_STREAM_TIMEOUT,
+      signal: AbortSignal.timeout(AUDIT_LOG_STREAM_TIMEOUT)
+    });
+  };
+
+  return {
+    validateCredentials,
+    streamLog
+  };
+};
@@ -0,0 +1,8 @@
+import { LogProvider } from "../audit-log-stream-enums";
+
+export const getSplunkProviderListItem = () => {
+  return {
+    name: "Splunk" as const,
+    provider: LogProvider.Splunk as const
+  };
+};
@@ -0,0 +1,59 @@
+import { z } from "zod";
+
+import { LogProvider } from "../audit-log-stream-enums";
+import { BaseProviderSchema } from "../audit-log-stream-schemas";
+
+export const SplunkProviderCredentialsSchema = z.object({
+  hostname: z
+    .string()
+    .trim()
+    .min(1)
+    .max(255)
+    .superRefine((val, ctx) => {
+      if (val.includes("://")) {
+        ctx.addIssue({
+          code: "custom",
+          message: "Hostname should not include protocol"
+        });
+        return;
+      }
+
+      try {
+        const url = new URL(`https://${val}`);
+        if (url.hostname !== val) {
+          ctx.addIssue({
+            code: "custom",
+            message: "Must be a valid hostname without port or path"
+          });
+        }
+      } catch {
+        ctx.addIssue({ code: "custom", message: "Invalid hostname" });
+      }
+    }),
+  token: z.string().uuid().trim().min(1)
+});
+
+const BaseSplunkProviderSchema = BaseProviderSchema.extend({ provider: z.literal(LogProvider.Splunk) });
+
+export const SplunkProviderSchema = BaseSplunkProviderSchema.extend({
+  credentials: SplunkProviderCredentialsSchema
+});
+
+export const SanitizedSplunkProviderSchema = BaseSplunkProviderSchema.extend({
+  credentials: SplunkProviderCredentialsSchema.pick({
+    hostname: true
+  })
+});
+
+export const SplunkProviderListItemSchema = z.object({
+  name: z.literal("Splunk"),
+  provider: z.literal(LogProvider.Splunk)
+});
+
+export const CreateSplunkProviderLogStreamSchema = z.object({
+  credentials: SplunkProviderCredentialsSchema
+});
+
+export const UpdateSplunkProviderLogStreamSchema = z.object({
+  credentials: SplunkProviderCredentialsSchema
+});
@@ -0,0 +1,7 @@
+import { z } from "zod";
+
+import { SplunkProviderCredentialsSchema, SplunkProviderSchema } from "./splunk-provider-schemas";
+
+export type TSplunkProvider = z.infer<typeof SplunkProviderSchema>;
+
+export type TSplunkProviderCredentials = z.infer<typeof SplunkProviderCredentialsSchema>;
@@ -1,22 +1,14 @@
-import { AxiosError, RawAxiosRequestHeaders } from "axios";
-
-import { SecretKeyEncoding } from "@app/db/schemas";
-import { request } from "@app/lib/config/request";
-import { crypto } from "@app/lib/crypto/cryptography";
-import { logger } from "@app/lib/logger";
+import { TAuditLogStreamServiceFactory } from "@app/ee/services/audit-log-stream/audit-log-stream-service";
 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 { providerSpecificPayload } from "../audit-log-stream/audit-log-stream-fns";
-import { LogStreamHeaders } from "../audit-log-stream/audit-log-stream-types";
 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">;
+  auditLogStreamService: Pick<TAuditLogStreamServiceFactory, "streamLog">;
  queueService: TQueueServiceFactory;
  projectDAL: Pick<TProjectDALFactory, "findById">;
  licenseService: Pick<TLicenseServiceFactory, "getPlan">;
@@ -35,7 +27,7 @@ export const auditLogQueueServiceFactory = async ({
  queueService,
  projectDAL,
  licenseService,
-  auditLogStreamDAL
+  auditLogStreamService
 }: TAuditLogQueueServiceFactoryDep): Promise<TAuditLogQueueServiceFactory> => {
  const pushToLog = async (data: TCreateAuditLogDTO) => {
    await queueService.queue<QueueName.AuditLog>(QueueName.AuditLog, QueueJobs.AuditLog, data, {
@@ -86,60 +78,7 @@ export const auditLogQueueServiceFactory = async ({
        userAgentType
      });

-      const logStreams = orgId ? await auditLogStreamDAL.find({ orgId }) : [];
-      await Promise.allSettled(
-        logStreams.map(
-          async ({
-            url,
-            encryptedHeadersTag,
-            encryptedHeadersIV,
-            encryptedHeadersKeyEncoding,
-            encryptedHeadersCiphertext
-          }) => {
-            const streamHeaders =
-              encryptedHeadersIV && encryptedHeadersCiphertext && encryptedHeadersTag
-                ? (JSON.parse(
-                    crypto
-                      .encryption()
-                      .symmetric()
-                      .decryptWithRootEncryptionKey({
-                        keyEncoding: encryptedHeadersKeyEncoding as SecretKeyEncoding,
-                        iv: encryptedHeadersIV,
-                        tag: encryptedHeadersTag,
-                        ciphertext: encryptedHeadersCiphertext
-                      })
-                  ) as LogStreamHeaders[])
-                : [];
-
-            const headers: RawAxiosRequestHeaders = { "Content-Type": "application/json" };
-
-            if (streamHeaders.length)
-              streamHeaders.forEach(({ key, value }) => {
-                headers[key] = value;
-              });
-
-            try {
-              const response = await request.post(
-                url,
-                { ...providerSpecificPayload(url), ...auditLog },
-                {
-                  headers,
-                  // request timeout
-                  timeout: AUDIT_LOG_STREAM_TIMEOUT,
-                  // connection timeout
-                  signal: AbortSignal.timeout(AUDIT_LOG_STREAM_TIMEOUT)
-                }
-              );
-              return response;
-            } catch (error) {
-              logger.error(
-                `Failed to stream audit log [url=${url}] for org [orgId=${orgId}] [error=${(error as AxiosError).message}]`
-              );
-              return error;
-            }
-          }
-        )
-      );
+      await auditLogStreamService.streamLog(orgId, auditLog);
    }
  });

@@ -6,9 +6,9 @@ import { getConfig } from "@app/lib/config/env";
 import { BadRequestError } from "@app/lib/errors";
 import { ActorType } from "@app/services/auth/auth-type";

-import { OrgPermissionActions, OrgPermissionSubjects } from "../permission/org-permission";
+import { OrgPermissionAuditLogsActions, OrgPermissionSubjects } from "../permission/org-permission";
 import { TPermissionServiceFactory } from "../permission/permission-service-types";
-import { ProjectPermissionActions, ProjectPermissionSub } from "../permission/project-permission";
+import { ProjectPermissionAuditLogsActions, ProjectPermissionSub } from "../permission/project-permission";
 import { TAuditLogDALFactory } from "./audit-log-dal";
 import { TAuditLogQueueServiceFactory } from "./audit-log-queue";
 import { EventType, TAuditLogServiceFactory } from "./audit-log-types";
@@ -41,7 +41,10 @@ export const auditLogServiceFactory = ({
        actorOrgId,
        actionProjectType: ActionProjectType.Any
      });
-      ForbiddenError.from(permission).throwUnlessCan(ProjectPermissionActions.Read, ProjectPermissionSub.AuditLogs);
+      ForbiddenError.from(permission).throwUnlessCan(
+        ProjectPermissionAuditLogsActions.Read,
+        ProjectPermissionSub.AuditLogs
+      );
    } else {
      // Organization-wide logs
      const { permission } = await permissionService.getOrgPermission(
@@ -52,7 +55,10 @@ export const auditLogServiceFactory = ({
        actorOrgId
      );

-      ForbiddenError.from(permission).throwUnlessCan(OrgPermissionActions.Read, OrgPermissionSubjects.AuditLogs);
+      ForbiddenError.from(permission).throwUnlessCan(
+        OrgPermissionAuditLogsActions.Read,
+        OrgPermissionSubjects.AuditLogs
+      );
    }

    // If project ID is not provided, then we need to return all the audit logs for the organization itself.
@@ -30,7 +30,7 @@ const generateUsername = (usernameTemplate?: string | null, identity?: { name: s
 export const CassandraProvider = (): TDynamicProviderFns => {
  const validateProviderInputs = async (inputs: unknown) => {
    const providerInputs = await DynamicSecretCassandraSchema.parseAsync(inputs);
-    const hostIps = await Promise.all(
+    await Promise.all(
      providerInputs.host
        .split(",")
        .filter(Boolean)
@@ -48,10 +48,10 @@ export const CassandraProvider = (): TDynamicProviderFns => {
      allowedExpressions: (val) => ["username"].includes(val)
    });

-    return { ...providerInputs, hostIps };
+    return { ...providerInputs };
  };

-  const $getClient = async (providerInputs: z.infer<typeof DynamicSecretCassandraSchema> & { hostIps: string[] }) => {
+  const $getClient = async (providerInputs: z.infer<typeof DynamicSecretCassandraSchema>) => {
    const sslOptions = providerInputs.ca ? { rejectUnauthorized: false, ca: providerInputs.ca } : undefined;
    const client = new cassandra.Client({
      sslOptions,
@@ -64,7 +64,7 @@ export const CassandraProvider = (): TDynamicProviderFns => {
      },
      keyspace: providerInputs.keyspace,
      localDataCenter: providerInputs?.localDataCenter,
-      contactPoints: providerInputs.hostIps
+      contactPoints: providerInputs.host.split(",")
    });
    return client;
  };
@@ -28,14 +28,14 @@ const generateUsername = (usernameTemplate?: string | null, identity?: { name: s
 export const ElasticSearchProvider = (): TDynamicProviderFns => {
  const validateProviderInputs = async (inputs: unknown) => {
    const providerInputs = await DynamicSecretElasticSearchSchema.parseAsync(inputs);
-    const [hostIp] = await verifyHostInputValidity(providerInputs.host);
-    return { ...providerInputs, hostIp };
+    await verifyHostInputValidity(providerInputs.host);
+    return { ...providerInputs };
  };

-  const $getClient = async (providerInputs: z.infer<typeof DynamicSecretElasticSearchSchema> & { hostIp: string }) => {
+  const $getClient = async (providerInputs: z.infer<typeof DynamicSecretElasticSearchSchema>) => {
    const connection = new ElasticSearchClient({
      node: {
-        url: new URL(`${providerInputs.hostIp}:${providerInputs.port}`),
+        url: new URL(`${providerInputs.host}:${providerInputs.port}`),
        ...(providerInputs.ca && {
          ssl: {
            rejectUnauthorized: false,
@@ -28,15 +28,15 @@ const generateUsername = (usernameTemplate?: string | null, identity?: { name: s
 export const MongoDBProvider = (): TDynamicProviderFns => {
  const validateProviderInputs = async (inputs: unknown) => {
    const providerInputs = await DynamicSecretMongoDBSchema.parseAsync(inputs);
-    const [hostIp] = await verifyHostInputValidity(providerInputs.host);
-    return { ...providerInputs, hostIp };
+    await verifyHostInputValidity(providerInputs.host);
+    return { ...providerInputs };
  };

-  const $getClient = async (providerInputs: z.infer<typeof DynamicSecretMongoDBSchema> & { hostIp: string }) => {
+  const $getClient = async (providerInputs: z.infer<typeof DynamicSecretMongoDBSchema>) => {
    const isSrv = !providerInputs.port;
    const uri = isSrv
-      ? `mongodb+srv://${providerInputs.hostIp}`
-      : `mongodb://${providerInputs.hostIp}:${providerInputs.port}`;
+      ? `mongodb+srv://${providerInputs.host}`
+      : `mongodb://${providerInputs.host}:${providerInputs.port}`;

    const client = new MongoClient(uri, {
      auth: {
@@ -87,13 +87,13 @@ async function deleteRabbitMqUser({ axiosInstance, usernameToDelete }: TDeleteRa
 export const RabbitMqProvider = (): TDynamicProviderFns => {
  const validateProviderInputs = async (inputs: unknown) => {
    const providerInputs = await DynamicSecretRabbitMqSchema.parseAsync(inputs);
-    const [hostIp] = await verifyHostInputValidity(providerInputs.host);
-    return { ...providerInputs, hostIp };
+    await verifyHostInputValidity(providerInputs.host);
+    return { ...providerInputs };
  };

-  const $getClient = async (providerInputs: z.infer<typeof DynamicSecretRabbitMqSchema> & { hostIp: string }) => {
+  const $getClient = async (providerInputs: z.infer<typeof DynamicSecretRabbitMqSchema>) => {
    const axiosInstance = axios.create({
-      baseURL: `${providerInputs.hostIp}:${providerInputs.port}/api`,
+      baseURL: `${providerInputs.host}:${providerInputs.port}/api`,
      auth: {
        username: providerInputs.username,
        password: providerInputs.password
@@ -36,7 +36,7 @@ export const SapAseProvider = (): TDynamicProviderFns => {
  const validateProviderInputs = async (inputs: unknown) => {
    const providerInputs = await DynamicSecretSapAseSchema.parseAsync(inputs);

-    const [hostIp] = await verifyHostInputValidity(providerInputs.host);
+    await verifyHostInputValidity(providerInputs.host);
    validateHandlebarTemplate("SAP ASE creation", providerInputs.creationStatement, {
      allowedExpressions: (val) => ["username", "password"].includes(val)
    });
@@ -45,16 +45,13 @@ export const SapAseProvider = (): TDynamicProviderFns => {
        allowedExpressions: (val) => ["username"].includes(val)
      });
    }
-    return { ...providerInputs, hostIp };
+    return { ...providerInputs };
  };

-  const $getClient = async (
-    providerInputs: z.infer<typeof DynamicSecretSapAseSchema> & { hostIp: string },
-    useMaster?: boolean
-  ) => {
+  const $getClient = async (providerInputs: z.infer<typeof DynamicSecretSapAseSchema>, useMaster?: boolean) => {
    const connectionString =
      `DRIVER={FreeTDS};` +
-      `SERVER=${providerInputs.hostIp};` +
+      `SERVER=${providerInputs.host};` +
      `PORT=${providerInputs.port};` +
      `DATABASE=${useMaster ? "master" : providerInputs.database};` +
      `UID=${providerInputs.username};` +
@@ -37,7 +37,7 @@ export const SapHanaProvider = (): TDynamicProviderFns => {
  const validateProviderInputs = async (inputs: unknown) => {
    const providerInputs = await DynamicSecretSapHanaSchema.parseAsync(inputs);

-    const [hostIp] = await verifyHostInputValidity(providerInputs.host);
+    await verifyHostInputValidity(providerInputs.host);
    validateHandlebarTemplate("SAP Hana creation", providerInputs.creationStatement, {
      allowedExpressions: (val) => ["username", "password", "expiration"].includes(val)
    });
@@ -49,12 +49,12 @@ export const SapHanaProvider = (): TDynamicProviderFns => {
    validateHandlebarTemplate("SAP Hana revoke", providerInputs.revocationStatement, {
      allowedExpressions: (val) => ["username"].includes(val)
    });
-    return { ...providerInputs, hostIp };
+    return { ...providerInputs };
  };

-  const $getClient = async (providerInputs: z.infer<typeof DynamicSecretSapHanaSchema> & { hostIp: string }) => {
+  const $getClient = async (providerInputs: z.infer<typeof DynamicSecretSapHanaSchema>) => {
    const client = hdb.createClient({
-      host: providerInputs.hostIp,
+      host: providerInputs.host,
      port: providerInputs.port,
      user: providerInputs.username,
      password: providerInputs.password,
@@ -150,8 +150,10 @@ export const SqlDatabaseProvider = ({ gatewayService }: TSqlDatabaseProviderDTO)
    return { ...providerInputs, hostIp };
  };

-  const $getClient = async (providerInputs: z.infer<typeof DynamicSecretSqlDBSchema>) => {
-    const ssl = providerInputs.ca ? { rejectUnauthorized: false, ca: providerInputs.ca } : undefined;
+  const $getClient = async (providerInputs: z.infer<typeof DynamicSecretSqlDBSchema> & { hostIp: string }) => {
+    const ssl = providerInputs.ca
+      ? { rejectUnauthorized: false, ca: providerInputs.ca, servername: providerInputs.host }
+      : undefined;
    const isMsSQLClient = providerInputs.client === SqlProviders.MsSQL;

    const db = knex({
@@ -159,7 +161,7 @@ export const SqlDatabaseProvider = ({ gatewayService }: TSqlDatabaseProviderDTO)
      connection: {
        database: providerInputs.database,
        port: providerInputs.port,
-        host: providerInputs.host,
+        host: providerInputs.client === SqlProviders.Postgres ? providerInputs.hostIp : providerInputs.host,
        user: providerInputs.username,
        password: providerInputs.password,
        ssl,
@@ -209,8 +211,8 @@ export const SqlDatabaseProvider = ({ gatewayService }: TSqlDatabaseProviderDTO)
  const validateConnection = async (inputs: unknown) => {
    const providerInputs = await validateProviderInputs(inputs);
    let isConnected = false;
-    const gatewayCallback = async (host = providerInputs.hostIp, port = providerInputs.port) => {
-      const db = await $getClient({ ...providerInputs, port, host });
+    const gatewayCallback = async (host = providerInputs.host, port = providerInputs.port) => {
+      const db = await $getClient({ ...providerInputs, port, host, hostIp: providerInputs.hostIp });
      // oracle needs from keyword
      const testStatement = providerInputs.client === SqlProviders.Oracle ? "SELECT 1 FROM DUAL" : "SELECT 1";

@@ -1,11 +1,11 @@
-import Redis from "ioredis";
+import { Cluster, Redis } from "ioredis";
 import { z } from "zod";

 import { logger } from "@app/lib/logger";

 import { BusEventSchema, TopicName } from "./types";

-export const eventBusFactory = (redis: Redis) => {
+export const eventBusFactory = (redis: Redis | Cluster) => {
  const publisher = redis.duplicate();
  // Duplicate the publisher to create a subscriber.
  // This is necessary because Redis does not allow a single connection to both publish and subscribe.
@@ -1,6 +1,6 @@
 /* eslint-disable no-continue */
 import { subject } from "@casl/ability";
-import Redis from "ioredis";
+import { Cluster, Redis } from "ioredis";

 import { KeyStorePrefixes } from "@app/keystore/keystore";
 import { logger } from "@app/lib/logger";
@@ -12,7 +12,7 @@ import { BusEvent, RegisteredEvent } from "./types";
 const AUTH_REFRESH_INTERVAL = 60 * 1000;
 const HEART_BEAT_INTERVAL = 15 * 1000;

-export const sseServiceFactory = (bus: TEventBusService, redis: Redis) => {
+export const sseServiceFactory = (bus: TEventBusService, redis: Redis | Cluster) => {
  const clients = new Set<EventStreamClient>();

  const heartbeatInterval = setInterval(() => {
@@ -3,7 +3,7 @@ import { Readable } from "node:stream";

 import { MongoAbility, PureAbility } from "@casl/ability";
 import { MongoQuery } from "@ucast/mongo2js";
-import Redis from "ioredis";
+import { Cluster, Redis } from "ioredis";
 import { nanoid } from "nanoid";

 import { ProjectType } from "@app/db/schemas";
@@ -65,7 +65,7 @@ export type EventStreamClient = {
  matcher: PureAbility;
 };

-export function createEventStreamClient(redis: Redis, options: IEventStreamClientOpts): EventStreamClient {
+export function createEventStreamClient(redis: Redis | Cluster, options: IEventStreamClientOpts): EventStreamClient {
  const rules = options.registered.map((r) => {
    const secretPath = r.conditions?.secretPath;
    const hasConditions = r.conditions?.environmentSlug || r.conditions?.secretPath;
@@ -1,14 +1,19 @@
+/* eslint-disable @typescript-eslint/return-await */
+/* eslint-disable no-await-in-loop */
 import { ForbiddenError } from "@casl/ability";
 import { Octokit } from "@octokit/core";
 import { paginateGraphql } from "@octokit/plugin-paginate-graphql";
 import { Octokit as OctokitRest } from "@octokit/rest";
+import RE2 from "re2";

 import { OrgMembershipRole } from "@app/db/schemas";
 import { BadRequestError, NotFoundError } from "@app/lib/errors";
 import { groupBy } from "@app/lib/fn";
 import { logger } from "@app/lib/logger";
+import { retryWithBackoff } from "@app/lib/retry";
 import { TKmsServiceFactory } from "@app/services/kms/kms-service";
 import { KmsDataKey } from "@app/services/kms/kms-types";
+import { TOrgMembershipDALFactory } from "@app/services/org-membership/org-membership-dal";

 import { TGroupDALFactory } from "../group/group-dal";
 import { TUserGroupMembershipDALFactory } from "../group/user-group-membership-dal";
@@ -16,20 +21,67 @@ import { TLicenseServiceFactory } from "../license/license-service";
 import { OrgPermissionActions, OrgPermissionSubjects } from "../permission/org-permission";
 import { TPermissionServiceFactory } from "../permission/permission-service-types";
 import { TGithubOrgSyncDALFactory } from "./github-org-sync-dal";
-import { TCreateGithubOrgSyncDTO, TDeleteGithubOrgSyncDTO, TUpdateGithubOrgSyncDTO } from "./github-org-sync-types";
+import {
+  TCreateGithubOrgSyncDTO,
+  TDeleteGithubOrgSyncDTO,
+  TSyncAllTeamsDTO,
+  TSyncResult,
+  TUpdateGithubOrgSyncDTO,
+  TValidateGithubTokenDTO
+} from "./github-org-sync-types";

 const OctokitWithPlugin = Octokit.plugin(paginateGraphql);

+// Type definitions for GitHub API errors
+interface GitHubApiError extends Error {
+  status?: number;
+  response?: {
+    status?: number;
+    headers?: {
+      "x-ratelimit-reset"?: string;
+    };
+  };
+}
+
+interface OrgMembershipWithUser {
+  id: string;
+  orgId: string;
+  role: string;
+  status: string;
+  isActive: boolean;
+  inviteEmail: string | null;
+  user: {
+    id: string;
+    email: string;
+    username: string | null;
+    firstName: string | null;
+    lastName: string | null;
+  } | null;
+}
+
+interface GroupMembership {
+  id: string;
+  groupId: string;
+  groupName: string;
+  orgMembershipId: string;
+  firstName: string | null;
+  lastName: string | null;
+}
+
 type TGithubOrgSyncServiceFactoryDep = {
  githubOrgSyncDAL: TGithubOrgSyncDALFactory;
  permissionService: Pick<TPermissionServiceFactory, "getOrgPermission">;
  kmsService: Pick<TKmsServiceFactory, "createCipherPairWithDataKey">;
  userGroupMembershipDAL: Pick<
    TUserGroupMembershipDALFactory,
-    "findGroupMembershipsByUserIdInOrg" | "insertMany" | "delete"
+    "findGroupMembershipsByUserIdInOrg" | "findGroupMembershipsByGroupIdInOrg" | "insertMany" | "delete"
  >;
  groupDAL: Pick<TGroupDALFactory, "insertMany" | "transaction" | "find">;
  licenseService: Pick<TLicenseServiceFactory, "getPlan">;
+  orgMembershipDAL: Pick<
+    TOrgMembershipDALFactory,
+    "find" | "findOrgMembershipById" | "findOrgMembershipsWithUsersByOrgId"
+  >;
 };

 export type TGithubOrgSyncServiceFactory = ReturnType<typeof githubOrgSyncServiceFactory>;
@@ -40,7 +92,8 @@ export const githubOrgSyncServiceFactory = ({
  kmsService,
  userGroupMembershipDAL,
  groupDAL,
-  licenseService
+  licenseService,
+  orgMembershipDAL
 }: TGithubOrgSyncServiceFactoryDep) => {
  const createGithubOrgSync = async ({
    githubOrgName,
@@ -304,8 +357,8 @@ export const githubOrgSyncServiceFactory = ({
    const removeFromTeams = infisicalUserGroups.filter((el) => !githubUserTeamSet.has(el.groupName));

    if (newTeams.length || updateTeams.length || removeFromTeams.length) {
-      await groupDAL.transaction(async (tx) => {
-        if (newTeams.length) {
+      if (newTeams.length) {
+        await groupDAL.transaction(async (tx) => {
          const newGroups = await groupDAL.insertMany(
            newTeams.map((newGroupName) => ({
              name: newGroupName,
@@ -322,9 +375,11 @@ export const githubOrgSyncServiceFactory = ({
            })),
            tx
          );
-        }
+        });
+      }

-        if (updateTeams.length) {
+      if (updateTeams.length) {
+        await groupDAL.transaction(async (tx) => {
          await userGroupMembershipDAL.insertMany(
            updateTeams.map((el) => ({
              groupId: githubUserTeamOnInfisicalGroupByName[el][0].id,
@@ -332,16 +387,433 @@ export const githubOrgSyncServiceFactory = ({
            })),
            tx
          );
-        }
+        });
+      }

-        if (removeFromTeams.length) {
+      if (removeFromTeams.length) {
+        await groupDAL.transaction(async (tx) => {
          await userGroupMembershipDAL.delete(
            { userId, $in: { groupId: removeFromTeams.map((el) => el.groupId) } },
            tx
          );
-        }
+        });
+      }
+    }
+  };
+
+  const validateGithubToken = async ({ orgPermission, githubOrgAccessToken }: TValidateGithubTokenDTO) => {
+    const { permission } = await permissionService.getOrgPermission(
+      orgPermission.type,
+      orgPermission.id,
+      orgPermission.orgId,
+      orgPermission.authMethod,
+      orgPermission.orgId
+    );
+
+    ForbiddenError.from(permission).throwUnlessCan(OrgPermissionActions.Read, OrgPermissionSubjects.GithubOrgSync);
+
+    const plan = await licenseService.getPlan(orgPermission.orgId);
+    if (!plan.githubOrgSync) {
+      throw new BadRequestError({
+        message:
+          "Failed to validate GitHub token due to plan restriction. Upgrade plan to use GitHub organization sync."
      });
    }
+
+    const config = await githubOrgSyncDAL.findOne({ orgId: orgPermission.orgId });
+    if (!config) {
+      throw new BadRequestError({ message: "GitHub organization sync is not configured" });
+    }
+
+    try {
+      const testOctokit = new OctokitRest({
+        auth: githubOrgAccessToken,
+        request: {
+          signal: AbortSignal.timeout(10000)
+        }
+      });
+
+      const { data: org } = await testOctokit.rest.orgs.get({
+        org: config.githubOrgName
+      });
+
+      const octokitGraphQL = new OctokitWithPlugin({
+        auth: githubOrgAccessToken,
+        request: {
+          signal: AbortSignal.timeout(10000)
+        }
+      });
+
+      await octokitGraphQL.graphql(`query($org: String!) { organization(login: $org) { id name } }`, {
+        org: config.githubOrgName
+      });
+
+      return {
+        valid: true,
+        organizationInfo: {
+          id: org.id,
+          login: org.login,
+          name: org.name || org.login,
+          publicRepos: org.public_repos,
+          privateRepos: org.owned_private_repos || 0
+        }
+      };
+    } catch (error) {
+      logger.error(error, `GitHub token validation failed for org ${config.githubOrgName}`);
+
+      const gitHubError = error as GitHubApiError;
+      const statusCode = gitHubError.status || gitHubError.response?.status;
+      if (statusCode) {
+        if (statusCode === 401) {
+          throw new BadRequestError({
+            message: "GitHub access token is invalid or expired."
+          });
+        }
+        if (statusCode === 403) {
+          throw new BadRequestError({
+            message:
+              "GitHub access token lacks required permissions. Required: 1) 'read:org' scope for organization teams, 2) Token owner must be an organization member with team visibility access, 3) Organization settings must allow team visibility. Check GitHub token scopes and organization member permissions."
+          });
+        }
+        if (statusCode === 404) {
+          throw new BadRequestError({
+            message: `Organization '${config.githubOrgName}' not found or access token does not have access to it.`
+          });
+        }
+      }
+
+      throw new BadRequestError({
+        message: `GitHub token validation failed: ${(error as Error).message}`
+      });
+    }
+  };
+
+  const syncAllTeams = async ({ orgPermission }: TSyncAllTeamsDTO): Promise<TSyncResult> => {
+    const { permission } = await permissionService.getOrgPermission(
+      orgPermission.type,
+      orgPermission.id,
+      orgPermission.orgId,
+      orgPermission.authMethod,
+      orgPermission.orgId
+    );
+
+    ForbiddenError.from(permission).throwUnlessCan(
+      OrgPermissionActions.Edit,
+      OrgPermissionSubjects.GithubOrgSyncManual
+    );
+
+    const plan = await licenseService.getPlan(orgPermission.orgId);
+    if (!plan.githubOrgSync) {
+      throw new BadRequestError({
+        message:
+          "Failed to sync all GitHub teams due to plan restriction. Upgrade plan to use GitHub organization sync."
+      });
+    }
+
+    const config = await githubOrgSyncDAL.findOne({ orgId: orgPermission.orgId });
+    if (!config || !config?.isActive) {
+      throw new BadRequestError({ message: "GitHub organization sync is not configured or not active" });
+    }
+
+    const { decryptor } = await kmsService.createCipherPairWithDataKey({
+      type: KmsDataKey.Organization,
+      orgId: orgPermission.orgId
+    });
+
+    if (!config.encryptedGithubOrgAccessToken) {
+      throw new BadRequestError({
+        message: "GitHub organization access token is required. Please set a token first."
+      });
+    }
+
+    const orgAccessToken = decryptor({ cipherTextBlob: config.encryptedGithubOrgAccessToken }).toString();
+
+    try {
+      const testOctokit = new OctokitRest({
+        auth: orgAccessToken,
+        request: {
+          signal: AbortSignal.timeout(10000)
+        }
+      });
+
+      await testOctokit.rest.orgs.get({
+        org: config.githubOrgName
+      });
+
+      await testOctokit.rest.users.getAuthenticated();
+    } catch (error) {
+      throw new BadRequestError({
+        message: "Stored GitHub access token is invalid or expired. Please set a new token."
+      });
+    }
+
+    const allMembers = await orgMembershipDAL.findOrgMembershipsWithUsersByOrgId(orgPermission.orgId);
+    const activeMembers = allMembers.filter(
+      (member) => member.status === "accepted" && member.isActive
+    ) as OrgMembershipWithUser[];
+
+    const startTime = Date.now();
+    const syncErrors: string[] = [];
+
+    const octokit = new OctokitWithPlugin({
+      auth: orgAccessToken,
+      request: {
+        signal: AbortSignal.timeout(30000)
+      }
+    });
+
+    const data = await retryWithBackoff(async () => {
+      return octokit.graphql
+        .paginate<{
+          organization: {
+            teams: {
+              totalCount: number;
+              edges: {
+                node: {
+                  name: string;
+                  description: string;
+                  members: {
+                    edges: {
+                      node: {
+                        login: string;
+                      };
+                    }[];
+                  };
+                };
+              }[];
+            };
+          };
+        }>(
+          `
+        query orgTeams($cursor: String, $org: String!) {
+          organization(login: $org) {
+            teams(first: 100, after: $cursor) {
+              totalCount
+              edges {
+                node {
+                  name
+                  description
+                  members(first: 100) {
+                    edges {
+                      node {
+                        login
+                      }
+                    }
+                  }
+                }
+              }
+              pageInfo {
+                hasNextPage
+                endCursor
+              }
+            }
+          }
+        }
+        `,
+          {
+            org: config.githubOrgName
+          }
+        )
+        .catch((err) => {
+          logger.error(err, "GitHub GraphQL error for batched team sync");
+
+          const gitHubError = err as GitHubApiError;
+          const statusCode = gitHubError.status || gitHubError.response?.status;
+          if (statusCode) {
+            if (statusCode === 401) {
+              throw new BadRequestError({
+                message: "GitHub access token is invalid or expired. Please provide a new token."
+              });
+            }
+            if (statusCode === 403) {
+              throw new BadRequestError({
+                message:
+                  "GitHub access token lacks required permissions for organization team sync. Required: 1) 'admin:org' scope, 2) Token owner must be organization owner or have team read permissions, 3) Organization settings must allow team visibility. Check token scopes and user role."
+              });
+            }
+            if (statusCode === 404) {
+              throw new BadRequestError({
+                message: `Organization ${config.githubOrgName} not found or access token does not have sufficient permissions to read it.`
+              });
+            }
+          }
+
+          if ((err as Error)?.message?.includes("Although you appear to have the correct authorization credential")) {
+            throw new BadRequestError({
+              message:
+                "Organization has restricted OAuth app access. Please check that: 1) Your organization has approved the Infisical OAuth application, 2) The token owner has sufficient organization permissions."
+            });
+          }
+          throw new BadRequestError({ message: `GitHub GraphQL query failed: ${(err as Error)?.message}` });
+        });
+    });
+
+    const {
+      organization: { teams }
+    } = data;
+
+    const userTeamMap = new Map<string, string[]>();
+    const allGithubUsernamesInTeams = new Set<string>();
+
+    teams?.edges?.forEach((teamEdge) => {
+      const teamName = teamEdge.node.name.toLowerCase();
+
+      teamEdge.node.members.edges.forEach((memberEdge) => {
+        const username = memberEdge.node.login.toLowerCase();
+        allGithubUsernamesInTeams.add(username);
+
+        if (!userTeamMap.has(username)) {
+          userTeamMap.set(username, []);
+        }
+        userTeamMap.get(username)!.push(teamName);
+      });
+    });
+
+    const allGithubTeamNames = Array.from(new Set(teams?.edges?.map((edge) => edge.node.name.toLowerCase()) || []));
+
+    const existingTeamsOnInfisical = await groupDAL.find({
+      orgId: orgPermission.orgId,
+      $in: { name: allGithubTeamNames }
+    });
+    const existingTeamsMap = groupBy(existingTeamsOnInfisical, (i) => i.name);
+
+    const teamsToCreate = allGithubTeamNames.filter((teamName) => !(teamName in existingTeamsMap));
+    const createdTeams = new Set<string>();
+    const updatedTeams = new Set<string>();
+    const totalRemovedMemberships = 0;
+
+    await groupDAL.transaction(async (tx) => {
+      if (teamsToCreate.length > 0) {
+        const newGroups = await groupDAL.insertMany(
+          teamsToCreate.map((teamName) => ({
+            name: teamName,
+            role: OrgMembershipRole.Member,
+            slug: teamName,
+            orgId: orgPermission.orgId
+          })),
+          tx
+        );
+
+        newGroups.forEach((group) => {
+          if (!existingTeamsMap[group.name]) {
+            existingTeamsMap[group.name] = [];
+          }
+          existingTeamsMap[group.name].push(group);
+          createdTeams.add(group.name);
+        });
+      }
+
+      const allTeams = [...Object.values(existingTeamsMap).flat()];
+
+      for (const team of allTeams) {
+        const teamName = team.name.toLowerCase();
+
+        const currentMemberships = (await userGroupMembershipDAL.findGroupMembershipsByGroupIdInOrg(
+          team.id,
+          orgPermission.orgId
+        )) as GroupMembership[];
+
+        const expectedUserIds = new Set<string>();
+        teams?.edges?.forEach((teamEdge) => {
+          if (teamEdge.node.name.toLowerCase() === teamName) {
+            teamEdge.node.members.edges.forEach((memberEdge) => {
+              const githubUsername = memberEdge.node.login.toLowerCase();
+
+              const matchingMember = activeMembers.find((member) => {
+                const email = member.user?.email || member.inviteEmail;
+                if (!email) return false;
+
+                const emailPrefix = email.split("@")[0].toLowerCase();
+                const emailDomain = email.split("@")[1].toLowerCase();
+
+                if (emailPrefix === githubUsername) {
+                  return true;
+                }
+                const domainName = emailDomain.split(".")[0];
+                if (githubUsername.endsWith(domainName) && githubUsername.length > domainName.length) {
+                  const baseUsername = githubUsername.slice(0, -domainName.length);
+                  if (emailPrefix === baseUsername) {
+                    return true;
+                  }
+                }
+                const emailSplitRegex = new RE2(/[._-]/);
+                const emailParts = emailPrefix.split(emailSplitRegex);
+                const longestEmailPart = emailParts.reduce((a, b) => (a.length > b.length ? a : b), "");
+                if (longestEmailPart.length >= 4 && githubUsername.includes(longestEmailPart)) {
+                  return true;
+                }
+                return false;
+              });
+
+              if (matchingMember?.user?.id) {
+                expectedUserIds.add(matchingMember.user.id);
+                logger.info(
+                  `Matched GitHub user ${githubUsername} to email ${matchingMember.user?.email || matchingMember.inviteEmail}`
+                );
+              }
+            });
+          }
+        });
+
+        const currentUserIds = new Set<string>();
+        currentMemberships.forEach((membership) => {
+          const activeMember = activeMembers.find((am) => am.id === membership.orgMembershipId);
+          if (activeMember?.user?.id) {
+            currentUserIds.add(activeMember.user.id);
+          }
+        });
+
+        const usersToAdd = Array.from(expectedUserIds).filter((userId) => !currentUserIds.has(userId));
+
+        const membershipsToRemove = currentMemberships.filter((membership) => {
+          const activeMember = activeMembers.find((am) => am.id === membership.orgMembershipId);
+          return activeMember?.user?.id && !expectedUserIds.has(activeMember.user.id);
+        });
+
+        if (usersToAdd.length > 0) {
+          await userGroupMembershipDAL.insertMany(
+            usersToAdd.map((userId) => ({
+              userId,
+              groupId: team.id
+            })),
+            tx
+          );
+          updatedTeams.add(teamName);
+        }
+
+        if (membershipsToRemove.length > 0) {
+          await userGroupMembershipDAL.delete(
+            {
+              $in: {
+                id: membershipsToRemove.map((m) => m.id)
+              }
+            },
+            tx
+          );
+          updatedTeams.add(teamName);
+        }
+      }
+    });
+
+    const syncDuration = Date.now() - startTime;
+
+    logger.info(
+      {
+        orgId: orgPermission.orgId,
+        createdTeams: createdTeams.size,
+        syncDuration
+      },
+      "GitHub team sync completed"
+    );
+
+    return {
+      totalUsers: activeMembers.length,
+      errors: syncErrors,
+      createdTeams: Array.from(createdTeams),
+      updatedTeams: Array.from(updatedTeams),
+      removedMemberships: totalRemovedMemberships,
+      syncDuration
+    };
  };

  return {
@@ -349,6 +821,8 @@ export const githubOrgSyncServiceFactory = ({
    updateGithubOrgSync,
    deleteGithubOrgSync,
    getGithubOrgSync,
-    syncUserGroups
+    syncUserGroups,
+    syncAllTeams,
+    validateGithubToken
  };
 };
@@ -21,3 +21,21 @@ export interface TDeleteGithubOrgSyncDTO {
 export interface TGetGithubOrgSyncDTO {
  orgPermission: OrgServiceActor;
 }
+
+export interface TSyncAllTeamsDTO {
+  orgPermission: OrgServiceActor;
+}
+
+export interface TSyncResult {
+  totalUsers: number;
+  errors: string[];
+  createdTeams: string[];
+  updatedTeams: string[];
+  removedMemberships: number;
+  syncDuration: number;
+}
+
+export interface TValidateGithubTokenDTO {
+  orgPermission: OrgServiceActor;
+  githubOrgAccessToken: string;
+}
@@ -323,6 +323,8 @@ export const licenseServiceFactory = ({
      });
    }

+    await updateSubscriptionOrgMemberCount(orgId);
+
    const {
      data: { url }
    } = await licenseServerCloudApi.request.post(
@@ -722,6 +724,16 @@ export const licenseServiceFactory = ({
    await keyStore.deleteItem(FEATURE_CACHE_KEY(orgId));
  };

+  const getCustomerId = () => {
+    if (!selfHostedLicense) return "unknown";
+    return selfHostedLicense?.customerId;
+  };
+
+  const getLicenseId = () => {
+    if (!selfHostedLicense) return "unknown";
+    return selfHostedLicense?.licenseId;
+  };
+
  return {
    generateOrgCustomerId,
    removeOrgCustomer,
@@ -736,6 +748,8 @@ export const licenseServiceFactory = ({
      return onPremFeatures;
    },
    getPlan,
+    getCustomerId,
+    getLicenseId,
    invalidateGetPlan,
    updateSubscriptionOrgMemberCount,
    refreshPlan,
@@ -2,6 +2,7 @@ import { AbilityBuilder, createMongoAbility, MongoAbility } from "@casl/ability"

 import {
  ProjectPermissionActions,
+  ProjectPermissionAuditLogsActions,
  ProjectPermissionCertificateActions,
  ProjectPermissionCmekActions,
  ProjectPermissionCommitsActions,
@@ -394,7 +395,7 @@ const buildMemberPermissionRules = () => {
  );

  can([ProjectPermissionActions.Read], ProjectPermissionSub.Role);
-  can([ProjectPermissionActions.Read], ProjectPermissionSub.AuditLogs);
+  can([ProjectPermissionAuditLogsActions.Read], ProjectPermissionSub.AuditLogs);
  can([ProjectPermissionActions.Read], ProjectPermissionSub.IpAllowList);

  // double check if all CRUD are needed for CA and Certificates
@@ -502,7 +503,7 @@ const buildViewerPermissionRules = () => {
  can(ProjectPermissionActions.Read, ProjectPermissionSub.Settings);
  can(ProjectPermissionActions.Read, ProjectPermissionSub.Environments);
  can(ProjectPermissionActions.Read, ProjectPermissionSub.Tags);
-  can(ProjectPermissionActions.Read, ProjectPermissionSub.AuditLogs);
+  can(ProjectPermissionAuditLogsActions.Read, ProjectPermissionSub.AuditLogs);
  can(ProjectPermissionActions.Read, ProjectPermissionSub.IpAllowList);
  can(ProjectPermissionActions.Read, ProjectPermissionSub.CertificateAuthorities);
  can(ProjectPermissionCertificateActions.Read, ProjectPermissionSub.Certificates);
@@ -23,6 +23,10 @@ export enum OrgPermissionAppConnectionActions {
  Connect = "connect"
 }

+export enum OrgPermissionAuditLogsActions {
+  Read = "read"
+}
+
 export enum OrgPermissionKmipActions {
  Proxy = "proxy",
  Setup = "setup"
@@ -90,6 +94,7 @@ export enum OrgPermissionSubjects {
  Sso = "sso",
  Scim = "scim",
  GithubOrgSync = "github-org-sync",
+  GithubOrgSyncManual = "github-org-sync-manual",
  Ldap = "ldap",
  Groups = "groups",
  Billing = "billing",
@@ -119,13 +124,14 @@ export type OrgPermissionSet =
  | [OrgPermissionActions, OrgPermissionSubjects.Sso]
  | [OrgPermissionActions, OrgPermissionSubjects.Scim]
  | [OrgPermissionActions, OrgPermissionSubjects.GithubOrgSync]
+  | [OrgPermissionActions, OrgPermissionSubjects.GithubOrgSyncManual]
  | [OrgPermissionActions, OrgPermissionSubjects.Ldap]
  | [OrgPermissionGroupActions, OrgPermissionSubjects.Groups]
  | [OrgPermissionActions, OrgPermissionSubjects.SecretScanning]
  | [OrgPermissionBillingActions, OrgPermissionSubjects.Billing]
  | [OrgPermissionIdentityActions, OrgPermissionSubjects.Identity]
  | [OrgPermissionActions, OrgPermissionSubjects.Kms]
-  | [OrgPermissionActions, OrgPermissionSubjects.AuditLogs]
+  | [OrgPermissionAuditLogsActions, OrgPermissionSubjects.AuditLogs]
  | [OrgPermissionActions, OrgPermissionSubjects.ProjectTemplates]
  | [OrgPermissionGatewayActions, OrgPermissionSubjects.Gateway]
  | [
@@ -188,6 +194,10 @@ export const OrgPermissionSchema = z.discriminatedUnion("subject", [
    subject: z.literal(OrgPermissionSubjects.GithubOrgSync).describe("The entity this permission pertains to."),
    action: CASL_ACTION_SCHEMA_NATIVE_ENUM(OrgPermissionActions).describe("Describe what action an entity can take.")
  }),
+  z.object({
+    subject: z.literal(OrgPermissionSubjects.GithubOrgSyncManual).describe("The entity this permission pertains to."),
+    action: CASL_ACTION_SCHEMA_NATIVE_ENUM(OrgPermissionActions).describe("Describe what action an entity can take.")
+  }),
  z.object({
    subject: z.literal(OrgPermissionSubjects.Ldap).describe("The entity this permission pertains to."),
    action: CASL_ACTION_SCHEMA_NATIVE_ENUM(OrgPermissionActions).describe("Describe what action an entity can take.")
@@ -214,7 +224,9 @@ export const OrgPermissionSchema = z.discriminatedUnion("subject", [
  }),
  z.object({
    subject: z.literal(OrgPermissionSubjects.AuditLogs).describe("The entity this permission pertains to."),
-    action: CASL_ACTION_SCHEMA_NATIVE_ENUM(OrgPermissionActions).describe("Describe what action an entity can take.")
+    action: CASL_ACTION_SCHEMA_NATIVE_ENUM(OrgPermissionAuditLogsActions).describe(
+      "Describe what action an entity can take."
+    )
  }),
  z.object({
    subject: z.literal(OrgPermissionSubjects.ProjectTemplates).describe("The entity this permission pertains to."),
@@ -309,6 +321,11 @@ const buildAdminPermission = () => {
  can(OrgPermissionActions.Edit, OrgPermissionSubjects.GithubOrgSync);
  can(OrgPermissionActions.Delete, OrgPermissionSubjects.GithubOrgSync);

+  can(OrgPermissionActions.Read, OrgPermissionSubjects.GithubOrgSyncManual);
+  can(OrgPermissionActions.Create, OrgPermissionSubjects.GithubOrgSyncManual);
+  can(OrgPermissionActions.Edit, OrgPermissionSubjects.GithubOrgSyncManual);
+  can(OrgPermissionActions.Delete, OrgPermissionSubjects.GithubOrgSyncManual);
+
  can(OrgPermissionActions.Read, OrgPermissionSubjects.Ldap);
  can(OrgPermissionActions.Create, OrgPermissionSubjects.Ldap);
  can(OrgPermissionActions.Edit, OrgPermissionSubjects.Ldap);
@@ -340,10 +357,7 @@ const buildAdminPermission = () => {
  can(OrgPermissionActions.Edit, OrgPermissionSubjects.Kms);
  can(OrgPermissionActions.Delete, OrgPermissionSubjects.Kms);

-  can(OrgPermissionActions.Read, OrgPermissionSubjects.AuditLogs);
-  can(OrgPermissionActions.Create, OrgPermissionSubjects.AuditLogs);
-  can(OrgPermissionActions.Edit, OrgPermissionSubjects.AuditLogs);
-  can(OrgPermissionActions.Delete, OrgPermissionSubjects.AuditLogs);
+  can(OrgPermissionAuditLogsActions.Read, OrgPermissionSubjects.AuditLogs);

  can(OrgPermissionActions.Read, OrgPermissionSubjects.ProjectTemplates);
  can(OrgPermissionActions.Create, OrgPermissionSubjects.ProjectTemplates);
@@ -416,7 +430,7 @@ const buildMemberPermission = () => {
  can(OrgPermissionIdentityActions.Edit, OrgPermissionSubjects.Identity);
  can(OrgPermissionIdentityActions.Delete, OrgPermissionSubjects.Identity);

-  can(OrgPermissionActions.Read, OrgPermissionSubjects.AuditLogs);
+  can(OrgPermissionAuditLogsActions.Read, OrgPermissionSubjects.AuditLogs);

  can(OrgPermissionAppConnectionActions.Connect, OrgPermissionSubjects.AppConnections);
  can(OrgPermissionGatewayActions.ListGateways, OrgPermissionSubjects.Gateway);
@@ -164,6 +164,10 @@ export enum ProjectPermissionSecretEventActions {
  SubscribeImportMutations = "subscribe-on-import-mutations"
 }

+export enum ProjectPermissionAuditLogsActions {
+  Read = "read"
+}
+
 export enum ProjectPermissionSub {
  Role = "role",
  Member = "member",
@@ -304,7 +308,7 @@ export type ProjectPermissionSet =
  | [ProjectPermissionGroupActions, ProjectPermissionSub.Groups]
  | [ProjectPermissionActions, ProjectPermissionSub.Integrations]
  | [ProjectPermissionActions, ProjectPermissionSub.Webhooks]
-  | [ProjectPermissionActions, ProjectPermissionSub.AuditLogs]
+  | [ProjectPermissionAuditLogsActions, ProjectPermissionSub.AuditLogs]
  | [ProjectPermissionActions, ProjectPermissionSub.Environments]
  | [ProjectPermissionActions, ProjectPermissionSub.IpAllowList]
  | [ProjectPermissionActions, ProjectPermissionSub.Settings]
@@ -645,7 +649,7 @@ const GeneralPermissionSchema = [
  }),
  z.object({
    subject: z.literal(ProjectPermissionSub.AuditLogs).describe("The entity this permission pertains to."),
-    action: CASL_ACTION_SCHEMA_NATIVE_ENUM(ProjectPermissionActions).describe(
+    action: CASL_ACTION_SCHEMA_NATIVE_ENUM(ProjectPermissionAuditLogsActions).describe(
      "Describe what action an entity can take."
    )
  }),
@@ -59,7 +59,7 @@ type TSecretReplicationServiceFactoryDep = {
    TSecretVersionV2DALFactory,
    "find" | "insertMany" | "update" | "findLatestVersionMany"
  >;
-  secretImportDAL: Pick<TSecretImportDALFactory, "find" | "updateById" | "findByFolderIds">;
+  secretImportDAL: Pick<TSecretImportDALFactory, "find" | "updateById" | "findByFolderIds" | "findByIds">;
  folderDAL: Pick<
    TSecretFolderDALFactory,
    "findSecretPathByFolderIds" | "findBySecretPath" | "create" | "findOne" | "findByManySecretPath"