feat: changed from token to headers for audit log streams api

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

@@ -13,6 +13,7 @@ import { OrgPermissionActions, OrgPermissionSubjects } from "../permission/org-p
 import { TPermissionServiceFactory } from "../permission/permission-service";
 import { TAuditLogStreamDALFactory } from "./audit-log-stream-dal";
 import {
+  LogStreamHeaders,
   TCreateAuditLogStreamDTO,
   TDeleteAuditLogStreamDTO,
   TGetDetailsAuditLogStreamDTO,
@@ -33,7 +34,14 @@ export const auditLogStreamServiceFactory = ({
   permissionService,
   licenseService
 }: TAuditLogStreamServiceFactoryDep) => {
-  const create = async ({ url, actor, token, actorId, actorOrgId, actorAuthMethod }: TCreateAuditLogStreamDTO) => {
+  const create = async ({
+    url,
+    actor,
+    headers = [],
+    actorId,
+    actorOrgId,
+    actorAuthMethod
+  }: TCreateAuditLogStreamDTO) => {
     if (!actorOrgId) throw new BadRequestError({ message: "Missing org id from token" });
 
     const plan = await licenseService.getPlan(actorOrgId);
@@ -62,30 +70,37 @@ export const auditLogStreamServiceFactory = ({
     }
 
     // testing connection first
-    const headers: RawAxiosRequestHeaders = { "Content-Type": "application/json" };
-    if (token) headers.Authorization = `Bearer ${token}`;
-    await request.post(
-      url,
-      { ping: "ok" },
-      {
-        headers,
-        // request timeout
-        timeout: AUDIT_LOG_STREAM_TIMEOUT,
-        // connection timeout
-        signal: AbortSignal.timeout(AUDIT_LOG_STREAM_TIMEOUT)
-      }
-    );
-    const encryptedToken = token ? infisicalSymmetricEncypt(token) : undefined;
+    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,
+          // 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}`);
+      });
+    const encryptedHeaders = headers ? infisicalSymmetricEncypt(JSON.stringify(headers)) : undefined;
     const logStream = await auditLogStreamDAL.create({
       orgId: actorOrgId,
       url,
-      ...(encryptedToken
+      ...(encryptedHeaders
         ? {
-            encryptedTokenCiphertext: encryptedToken.ciphertext,
-            encryptedTokenIV: encryptedToken.iv,
-            encryptedTokenTag: encryptedToken.tag,
-            encryptedTokenAlgorithm: encryptedToken.algorithm,
-            encryptedTokenKeyEncoding: encryptedToken.encoding
+            encryptedHeadersCiphertext: encryptedHeaders.ciphertext,
+            encryptedHeadersIV: encryptedHeaders.iv,
+            encryptedHeadersTag: encryptedHeaders.tag,
+            encryptedHeadersAlgorithm: encryptedHeaders.algorithm,
+            encryptedHeadersKeyEncoding: encryptedHeaders.encoding
           }
         : {})
     });
@@ -96,7 +111,7 @@ export const auditLogStreamServiceFactory = ({
     id,
     url,
     actor,
-    token,
+    headers = [],
     actorId,
     actorOrgId,
     actorAuthMethod
@@ -119,13 +134,17 @@ export const auditLogStreamServiceFactory = ({
     if (url) validateLocalIps(url);
 
     // testing connection first
-    const headers: RawAxiosRequestHeaders = { "Content-Type": "application/json" };
-    if (token) headers.Authorization = `Bearer ${token}`;
+    const streamHeaders: RawAxiosRequestHeaders = { "Content-Type": "application/json" };
+    if (headers.length)
+      headers.forEach(({ key, value }) => {
+        streamHeaders[key] = value;
+      });
+
     await request.post(
       url || logStream.url,
       { ping: "ok" },
       {
-        headers,
+        headers: streamHeaders,
         // request timeout
         timeout: AUDIT_LOG_STREAM_TIMEOUT,
         // connection timeout
@@ -133,16 +152,16 @@ export const auditLogStreamServiceFactory = ({
       }
     );
 
-    const encryptedToken = token ? infisicalSymmetricEncypt(token) : undefined;
+    const encryptedHeaders = headers ? infisicalSymmetricEncypt(JSON.stringify(headers)) : undefined;
     const updatedLogStream = await auditLogStreamDAL.updateById(id, {
       url,
-      ...(encryptedToken
+      ...(encryptedHeaders
         ? {
-            encryptedTokenCiphertext: encryptedToken.ciphertext,
-            encryptedTokenIV: encryptedToken.iv,
-            encryptedTokenTag: encryptedToken.tag,
-            encryptedTokenAlgorithm: encryptedToken.algorithm,
-            encryptedTokenKeyEncoding: encryptedToken.encoding
+            encryptedHeadersCiphertext: encryptedHeaders.ciphertext,
+            encryptedHeadersIV: encryptedHeaders.iv,
+            encryptedHeadersTag: encryptedHeaders.tag,
+            encryptedHeadersAlgorithm: encryptedHeaders.algorithm,
+            encryptedHeadersKeyEncoding: encryptedHeaders.encoding
           }
         : {})
     });
@@ -171,17 +190,19 @@ export const auditLogStreamServiceFactory = ({
     const { permission } = await permissionService.getOrgPermission(actor, actorId, orgId, actorAuthMethod, actorOrgId);
     ForbiddenError.from(permission).throwUnlessCan(OrgPermissionActions.Read, OrgPermissionSubjects.Settings);
 
-    const token =
-      logStream?.encryptedTokenCiphertext && logStream?.encryptedTokenIV && logStream?.encryptedTokenTag
-        ? infisicalSymmetricDecrypt({
-            tag: logStream.encryptedTokenTag,
-            iv: logStream.encryptedTokenIV,
-            ciphertext: logStream.encryptedTokenCiphertext,
-            keyEncoding: logStream.encryptedTokenKeyEncoding as SecretKeyEncoding
-          })
+    const headers =
+      logStream?.encryptedHeadersCiphertext && logStream?.encryptedHeadersIV && logStream?.encryptedHeadersTag
+        ? (JSON.parse(
+            infisicalSymmetricDecrypt({
+              tag: logStream.encryptedHeadersTag,
+              iv: logStream.encryptedHeadersIV,
+              ciphertext: logStream.encryptedHeadersCiphertext,
+              keyEncoding: logStream.encryptedHeadersKeyEncoding as SecretKeyEncoding
+            })
+          ) as LogStreamHeaders[])
         : undefined;
 
-    return { ...logStream, token };
+    return { ...logStream, headers };
   };
 
   const list = async ({ actor, actorId, actorOrgId, actorAuthMethod }: TListAuditLogStreamDTO) => {

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

@@ -1,14 +1,19 @@
 import { TOrgPermission } from "@app/lib/types";
 
+export type LogStreamHeaders = {
+  key: string;
+  value: string;
+};
+
 export type TCreateAuditLogStreamDTO = Omit<TOrgPermission, "orgId"> & {
   url: string;
-  token?: string;
+  headers?: LogStreamHeaders[];
 };
 
 export type TUpdateAuditLogStreamDTO = Omit<TOrgPermission, "orgId"> & {
   id: string;
   url?: string;
-  token?: string;
+  headers?: LogStreamHeaders[];
 };
 
 export type TDeleteAuditLogStreamDTO = Omit<TOrgPermission, "orgId"> & {

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

@@ -8,6 +8,7 @@ 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 { 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";
@@ -74,20 +75,31 @@ export const auditLogQueueServiceFactory = ({
     const logStreams = orgId ? await auditLogStreamDAL.find({ orgId }) : [];
     await Promise.allSettled(
       logStreams.map(
-        async ({ url, encryptedTokenTag, encryptedTokenIV, encryptedTokenKeyEncoding, encryptedTokenCiphertext }) => {
-          const token =
-            encryptedTokenIV && encryptedTokenCiphertext && encryptedTokenTag
-              ? infisicalSymmetricDecrypt({
-                  keyEncoding: encryptedTokenKeyEncoding as SecretKeyEncoding,
-                  iv: encryptedTokenIV,
-                  tag: encryptedTokenTag,
-                  ciphertext: encryptedTokenCiphertext
-                })
-              : undefined;
+        async ({
+          url,
+          encryptedHeadersTag,
+          encryptedHeadersIV,
+          encryptedHeadersKeyEncoding,
+          encryptedHeadersCiphertext
+        }) => {
+          const streamHeaders =
+            encryptedHeadersIV && encryptedHeadersCiphertext && encryptedHeadersTag
+              ? (JSON.parse(
+                  infisicalSymmetricDecrypt({
+                    keyEncoding: encryptedHeadersKeyEncoding as SecretKeyEncoding,
+                    iv: encryptedHeadersIV,
+                    tag: encryptedHeadersTag,
+                    ciphertext: encryptedHeadersCiphertext
+                  })
+                ) as LogStreamHeaders[])
+              : [];
 
           const headers: RawAxiosRequestHeaders = { "Content-Type": "application/json" };
 
-          if (token) headers.Authorization = `Bearer ${token}`;
+          if (headers.length)
+            streamHeaders.forEach(({ key, value }) => {
+              headers[key] = value;
+            });
 
           return request.post(url, auditLog, {
             headers,