feat(pam): real-time session log sync via incremental batch uploads

backend/src/@types/knex.d.ts

@@ -712,6 +712,11 @@ import {
   TPamResourceRotationRulesUpdate
 } from "@app/db/schemas/pam-resource-rotation-rules";
 import { TPamResources, TPamResourcesInsert, TPamResourcesUpdate } from "@app/db/schemas/pam-resources";
+import {
+  TPamSessionEventBatches,
+  TPamSessionEventBatchesInsert,
+  TPamSessionEventBatchesUpdate
+} from "@app/db/schemas/pam-session-event-batches";
 import { TPamSessions, TPamSessionsInsert, TPamSessionsUpdate } from "@app/db/schemas/pam-sessions";
 import {
   TProjectMicrosoftTeamsConfigs,
@@ -1592,6 +1597,11 @@ declare module "knex/types/tables" {
     >;
     [TableName.PamAccount]: KnexOriginal.CompositeTableType<TPamAccounts, TPamAccountsInsert, TPamAccountsUpdate>;
     [TableName.PamSession]: KnexOriginal.CompositeTableType<TPamSessions, TPamSessionsInsert, TPamSessionsUpdate>;
+    [TableName.PamSessionEventBatch]: KnexOriginal.CompositeTableType<
+      TPamSessionEventBatches,
+      TPamSessionEventBatchesInsert,
+      TPamSessionEventBatchesUpdate
+    >;
     [TableName.PamDiscoverySource]: KnexOriginal.CompositeTableType<
       TPamDiscoverySources,
       TPamDiscoverySourcesInsert,

backend/src/db/migrations/20260407000001_pam-session-event-batches.ts

@@ -0,0 +1,30 @@
+import { Knex } from "knex";
+
+import { TableName } from "../schemas";
+import { createOnUpdateTrigger, dropOnUpdateTrigger } from "../utils";
+
+export async function up(knex: Knex): Promise<void> {
+  if (!(await knex.schema.hasTable(TableName.PamSessionEventBatch))) {
+    await knex.schema.createTable(TableName.PamSessionEventBatch, (t) => {
+      t.uuid("id", { primaryKey: true }).defaultTo(knex.fn.uuid());
+
+      t.uuid("sessionId").notNullable();
+      t.foreign("sessionId").references("id").inTable(TableName.PamSession).onDelete("CASCADE");
+      t.index("sessionId");
+
+      t.bigInteger("startOffset").notNullable();
+      t.binary("encryptedEventsBlob").notNullable();
+
+      t.timestamps(true, true, true);
+
+      t.unique(["sessionId", "startOffset"]);
+    });
+
+    await createOnUpdateTrigger(knex, TableName.PamSessionEventBatch);
+  }
+}
+
+export async function down(knex: Knex): Promise<void> {
+  await dropOnUpdateTrigger(knex, TableName.PamSessionEventBatch);
+  await knex.schema.dropTableIfExists(TableName.PamSessionEventBatch);
+}

backend/src/db/schemas/index.ts

@@ -120,6 +120,7 @@ export * from "./pam-folders";
 export * from "./pam-resource-favorites";
 export * from "./pam-resource-rotation-rules";
 export * from "./pam-resources";
+export * from "./pam-session-event-batches";
 export * from "./pam-sessions";
 export * from "./pki-acme-accounts";
 export * from "./pki-acme-auths";

backend/src/db/schemas/models.ts

@@ -227,6 +227,7 @@ export enum TableName {
   PamResource = "pam_resources",
   PamAccount = "pam_accounts",
   PamSession = "pam_sessions",
+  PamSessionEventBatch = "pam_session_event_batches",
   PamDiscoverySource = "pam_discovery_sources",
   PamDiscoverySourceRun = "pam_discovery_source_runs",
   PamDiscoverySourceResource = "pam_discovery_source_resources",

backend/src/db/schemas/pam-session-event-batches.ts

@@ -0,0 +1,25 @@
+// Code generated by automation script, DO NOT EDIT.
+// Automated by pulling database and generating zod schema
+// To update. Just run npm run generate:schema
+// Written by akhilmhdh.
+
+import { z } from "zod";
+
+import { zodBuffer } from "@app/lib/zod";
+
+import { TImmutableDBKeys } from "./models";
+
+export const PamSessionEventBatchesSchema = z.object({
+  id: z.string().uuid(),
+  sessionId: z.string().uuid(),
+  startOffset: z.coerce.number(),
+  encryptedEventsBlob: zodBuffer,
+  createdAt: z.date(),
+  updatedAt: z.date()
+});
+
+export type TPamSessionEventBatches = z.infer<typeof PamSessionEventBatchesSchema>;
+export type TPamSessionEventBatchesInsert = Omit<z.input<typeof PamSessionEventBatchesSchema>, TImmutableDBKeys>;
+export type TPamSessionEventBatchesUpdate = Partial<
+  Omit<z.input<typeof PamSessionEventBatchesSchema>, TImmutableDBKeys>
+>;

backend/src/ee/routes/v1/pam-session-router.ts

@@ -30,6 +30,10 @@ const SessionCredentialsSchema = z.union([
 ]);
 
 export const registerPamSessionRouter = async (server: FastifyZodProvider) => {
+  server.addContentTypeParser("application/octet-stream", { parseAs: "buffer" }, (_req, body, done) => {
+    done(null, body);
+  });
+
   // Meant to be hit solely by gateway identities
   server.route({
     method: "GET",
@@ -269,19 +273,6 @@ export const registerPamSessionRouter = async (server: FastifyZodProvider) => {
     onRequest: verifyAuth([AuthMode.JWT]),
     handler: async (req) => {
       const response = await server.services.pamSession.getById(req.params.sessionId, req.permission);
-
-      await server.services.auditLog.createAuditLog({
-        ...req.auditLogInfo,
-        orgId: req.permission.orgId,
-        projectId: response.session.projectId,
-        event: {
-          type: EventType.PAM_SESSION_GET,
-          metadata: {
-            sessionId: req.params.sessionId
-          }
-        }
-      });
-
       return response;
     }
   });
@@ -322,4 +313,52 @@ export const registerPamSessionRouter = async (server: FastifyZodProvider) => {
       return response;
     }
   });
+
+  // Meant to be hit solely by gateway identities
+  server.route({
+    method: "POST",
+    url: "/:sessionId/event-batches",
+    config: {
+      rateLimit: writeLimit
+    },
+    schema: {
+      description: "Upload a PAM session event batch",
+      params: z.object({
+        sessionId: z.string().uuid()
+      }),
+      querystring: z.object({
+        startOffset: z.coerce.number().int().nonnegative()
+      }),
+      body: z.instanceof(Buffer),
+      response: {
+        200: z.object({ ok: z.literal(true) })
+      }
+    },
+    onRequest: verifyAuth([AuthMode.IDENTITY_ACCESS_TOKEN]),
+    handler: async (req) => {
+      const { projectId } = await server.services.pamSession.uploadEventBatch(
+        {
+          sessionId: req.params.sessionId,
+          startOffset: req.query.startOffset,
+          events: req.body
+        },
+        req.permission
+      );
+
+      await server.services.auditLog.createAuditLog({
+        ...req.auditLogInfo,
+        orgId: req.permission.orgId,
+        projectId,
+        event: {
+          type: EventType.PAM_SESSION_EVENT_BATCH_UPLOAD,
+          metadata: {
+            sessionId: req.params.sessionId,
+            startOffset: req.query.startOffset
+          }
+        }
+      });
+
+      return { ok: true as const };
+    }
+  });
 };

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

@@ -596,6 +596,7 @@ export enum EventType {
   PAM_SESSION_TERMINATE = "pam-session-terminate",
   PAM_SESSION_GET = "pam-session-get",
   PAM_SESSION_LIST = "pam-session-list",
+  PAM_SESSION_EVENT_BATCH_UPLOAD = "pam-session-event-batch-upload",
   PAM_FOLDER_CREATE = "pam-folder-create",
   PAM_FOLDER_UPDATE = "pam-folder-update",
   PAM_FOLDER_DELETE = "pam-folder-delete",
@@ -4747,6 +4748,14 @@ interface PamSessionListEvent {
   };
 }
 
+interface PamSessionEventBatchUploadEvent {
+  type: EventType.PAM_SESSION_EVENT_BATCH_UPLOAD;
+  metadata: {
+    sessionId: string;
+    startOffset: number;
+  };
+}
+
 interface PamFolderCreateEvent {
   type: EventType.PAM_FOLDER_CREATE;
   metadata: {
@@ -6163,6 +6172,7 @@ export type Event =
   | PamSessionTerminateEvent
   | PamSessionGetEvent
   | PamSessionListEvent
+  | PamSessionEventBatchUploadEvent
   | PamFolderCreateEvent
   | PamFolderUpdateEvent
   | PamFolderDeleteEvent

backend/src/ee/services/pam-session/pam-session-event-batch-dal.ts

@@ -0,0 +1,27 @@
+import { Knex } from "knex";
+
+import { TDbClient } from "@app/db";
+import { TableName } from "@app/db/schemas";
+import { ormify } from "@app/lib/knex";
+
+export type TPamSessionEventBatchDALFactory = ReturnType<typeof pamSessionEventBatchDALFactory>;
+
+export const pamSessionEventBatchDALFactory = (db: TDbClient) => {
+  const orm = ormify(db, TableName.PamSessionEventBatch);
+
+  const findBySessionId = async (sessionId: string, tx?: Knex) => {
+    return (tx || db.replicaNode())(TableName.PamSessionEventBatch)
+      .where("sessionId", sessionId)
+      .orderBy("startOffset", "asc")
+      .select("*");
+  };
+
+  const upsertBatch = async (sessionId: string, startOffset: number, encryptedEventsBlob: Buffer, tx?: Knex) => {
+    return (tx || db)(TableName.PamSessionEventBatch)
+      .insert({ sessionId, startOffset, encryptedEventsBlob })
+      .onConflict(["sessionId", "startOffset"])
+      .merge(["encryptedEventsBlob"]); // on re-upload of the same offset, overwrite the blob instead of erroring or skipping
+  };
+
+  return { ...orm, findBySessionId, upsertBatch };
+};

backend/src/ee/services/pam-session/pam-session-fns.ts

@@ -1,4 +1,4 @@
-import { TPamSessions } from "@app/db/schemas";
+import { TPamSessionEventBatches, TPamSessions } from "@app/db/schemas";
 import { TKmsServiceFactory } from "@app/services/kms/kms-service";
 import { KmsDataKey } from "@app/services/kms/kms-types";
 
@@ -25,6 +25,25 @@ export const decryptSessionCommandLogs = async ({
   return JSON.parse(decryptedPlainTextBlob.toString()) as (TPamSessionCommandLog | TTerminalEvent)[];
 };
 
+export const decryptBatches = async (
+  batches: TPamSessionEventBatches[],
+  projectId: string,
+  kmsService: Pick<TKmsServiceFactory, "createCipherPairWithDataKey">
+) => {
+  const { decryptor } = await kmsService.createCipherPairWithDataKey({
+    type: KmsDataKey.SecretManager,
+    projectId
+  });
+
+  const events: (TPamSessionCommandLog | TTerminalEvent)[] = [];
+  for (const batch of batches) {
+    const plain = decryptor({ cipherTextBlob: batch.encryptedEventsBlob });
+    const batchEvents = JSON.parse(plain.toString()) as (TPamSessionCommandLog | TTerminalEvent)[];
+    events.push(...batchEvents);
+  }
+  return events;
+};
+
 export const decryptSession = async (
   session: TPamSessions,
   projectId: string,

backend/src/ee/services/pam-session/pam-session-service.ts

@@ -20,11 +20,13 @@ import { OrgPermissionGatewayActions, OrgPermissionSubjects } from "../permissio
 import { ProjectPermissionPamSessionActions, ProjectPermissionSub } from "../permission/project-permission";
 import { TPamSessionDALFactory } from "./pam-session-dal";
 import { PamSessionStatus } from "./pam-session-enums";
-import { decryptSession } from "./pam-session-fns";
-import { TUpdateSessionLogsDTO } from "./pam-session-types";
+import { TPamSessionEventBatchDALFactory } from "./pam-session-event-batch-dal";
+import { decryptBatches, decryptSession } from "./pam-session-fns";
+import { TUpdateSessionLogsDTO, TUploadEventBatchDTO } from "./pam-session-types";
 
 type TPamSessionServiceFactoryDep = {
   pamSessionDAL: TPamSessionDALFactory;
+  pamSessionEventBatchDAL: TPamSessionEventBatchDALFactory;
   projectDAL: TProjectDALFactory;
   userDAL: Pick<TUserDALFactory, "findById">;
   permissionService: Pick<TPermissionServiceFactory, "getProjectPermission" | "getOrgPermission">;
@@ -36,6 +38,7 @@ export type TPamSessionServiceFactory = ReturnType<typeof pamSessionServiceFacto
 
 export const pamSessionServiceFactory = ({
   pamSessionDAL,
+  pamSessionEventBatchDAL,
   projectDAL,
   userDAL,
   permissionService,
@@ -91,6 +94,16 @@ export const pamSessionServiceFactory = ({
       ProjectPermissionSub.PamSessions
     );
 
+    const batches = await pamSessionEventBatchDAL.findBySessionId(sessionId);
+    if (batches.length > 0) {
+      const logs = await decryptBatches(batches, session.projectId, kmsService);
+      return {
+        session: { ...session, logs, encryptedLogsBlob: undefined } as unknown as Awaited<
+          ReturnType<typeof decryptSession>
+        >
+      };
+    }
+
     return {
       session: await decryptSession(session, session.projectId, kmsService)
     };
@@ -288,5 +301,46 @@ export const pamSessionServiceFactory = ({
     return { session: updatedSession, projectId: project.id, alreadyEnded: false };
   };
 
-  return { getById, list, updateLogsById, endSessionById, terminateSessionById };
+  const uploadEventBatch = async ({ sessionId, startOffset, events }: TUploadEventBatchDTO, actor: OrgServiceActor) => {
+    if (actor.type !== ActorType.IDENTITY) {
+      throw new ForbiddenRequestError({ message: "Only gateways can perform this action" });
+    }
+
+    const session = await pamSessionDAL.findById(sessionId);
+    if (!session) throw new NotFoundError({ message: `Session with ID '${sessionId}' not found` });
+
+    const project = await projectDAL.findById(session.projectId);
+    if (!project) throw new NotFoundError({ message: `Project with ID '${session.projectId}' not found` });
+
+    const { permission } = await permissionService.getOrgPermission({
+      actor: actor.type,
+      actorId: actor.id,
+      orgId: project.orgId,
+      actorAuthMethod: actor.authMethod,
+      actorOrgId: actor.orgId,
+      scope: OrganizationActionScope.Any
+    });
+
+    ForbiddenError.from(permission).throwUnlessCan(
+      OrgPermissionGatewayActions.CreateGateways,
+      OrgPermissionSubjects.Gateway
+    );
+
+    if (session.gatewayIdentityId && session.gatewayIdentityId !== actor.id) {
+      throw new ForbiddenRequestError({ message: "Identity does not have access to upload events for this session" });
+    }
+
+    const { encryptor } = await kmsService.createCipherPairWithDataKey({
+      type: KmsDataKey.SecretManager,
+      projectId: session.projectId
+    });
+
+    const { cipherTextBlob } = encryptor({ plainText: events });
+
+    await pamSessionEventBatchDAL.upsertBatch(sessionId, startOffset, cipherTextBlob);
+
+    return { projectId: project.id };
+  };
+
+  return { getById, list, updateLogsById, endSessionById, terminateSessionById, uploadEventBatch };
 };

backend/src/ee/services/pam-session/pam-session-types.ts

@@ -17,3 +17,9 @@ export type TUpdateSessionLogsDTO = {
   sessionId: string;
   logs: (TPamSessionCommandLog | TTerminalEvent | THttpEvent)[];
 };
+
+export type TUploadEventBatchDTO = {
+  sessionId: string;
+  startOffset: number;
+  events: Buffer;
+};

backend/src/server/routes/index.ts

@@ -100,6 +100,7 @@ import { pamResourceRotationRulesDALFactory } from "@app/ee/services/pam-resourc
 import { pamResourceRotationRulesServiceFactory } from "@app/ee/services/pam-resource/pam-resource-rotation-rules-service";
 import { pamResourceServiceFactory } from "@app/ee/services/pam-resource/pam-resource-service";
 import { pamSessionDALFactory } from "@app/ee/services/pam-session/pam-session-dal";
+import { pamSessionEventBatchDALFactory } from "@app/ee/services/pam-session/pam-session-event-batch-dal";
 import { pamSessionServiceFactory } from "@app/ee/services/pam-session/pam-session-service";
 import { pamWebAccessServiceFactory } from "@app/ee/services/pam-web-access/pam-web-access-service";
 import { permissionDALFactory } from "@app/ee/services/permission/permission-dal";
@@ -2787,6 +2788,7 @@ export const registerRoutes = async (
   const pamResourceFavoriteDAL = pamResourceFavoriteDALFactory(db);
   const pamAccountDAL = pamAccountDALFactory(db);
   const pamSessionDAL = pamSessionDALFactory(db);
+  const pamSessionEventBatchDAL = pamSessionEventBatchDALFactory(db);
   const pamDiscoveryRunDAL = pamDiscoveryRunDALFactory(db);
   const pamDiscoverySourceResourcesDAL = pamDiscoverySourceResourcesDALFactory(db);
   const pamDiscoverySourceAccountsDAL = pamDiscoverySourceAccountsDALFactory(db);
@@ -2863,6 +2865,7 @@ export const registerRoutes = async (
 
   const pamSessionService = pamSessionServiceFactory({
     pamSessionDAL,
+    pamSessionEventBatchDAL,
     projectDAL,
     userDAL,
     permissionService,