Merge pull request #173 from Infisical/feat/pam-session-real-time-log-sync

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

Author: bernie-g

packages/api/api.go

@@ -54,6 +54,7 @@ const (
 	operationCallGetPamSessionKey                  = "CallGetPamSessionKey"
 	operationCallUploadPamSessionLog               = "CallUploadPamSessionLog"
 	operationCallPAMSessionTermination             = "CallPAMSessionTermination"
+	operationCallUploadPamSessionEventBatch        = "CallUploadPamSessionEventBatch"
 	operationCallGetMFASessionStatus               = "CallGetMFASessionStatus"
 	operationCallOrgRelayHeartBeat                 = "CallOrgRelayHeartBeat"
 	operationCallInstanceRelayHeartBeat            = "CallInstanceRelayHeartBeat"
@@ -1008,6 +1009,23 @@ func CallUploadPamSessionLogs(httpClient *resty.Client, sessionId string, reques
 	return nil
 }
 
+func CallUploadPamSessionEventBatch(httpClient *resty.Client, sessionId string, startOffset int64, data []byte) error {
+	response, err := httpClient.
+		R().
+		SetHeader("User-Agent", USER_AGENT).
+		SetHeader("Content-Type", "application/octet-stream").
+		SetBody(data).
+		Post(fmt.Sprintf("%v/v1/pam/sessions/%s/event-batches?startOffset=%d", config.INFISICAL_URL, sessionId, startOffset))
+
+	if err != nil {
+		return NewGenericRequestError(operationCallUploadPamSessionEventBatch, err)
+	}
+	if response.IsError() {
+		return NewAPIErrorWithResponse(operationCallUploadPamSessionEventBatch, response, nil)
+	}
+	return nil
+}
+
 func CallPAMSessionTermination(httpClient *resty.Client, sessionId string) error {
 	response, err := httpClient.
 		R().

packages/pam/pam-proxy.go

@@ -158,6 +158,7 @@ func HandlePAMProxy(ctx context.Context, conn *tls.Conn, pamConfig *GatewayPAMCo
 	if err != nil {
 		return fmt.Errorf("failed to create session logger: %w", err)
 	}
+	pamConfig.SessionUploader.RegisterSession(pamConfig.SessionId)
 
 	serverName := credentials.Host
 	switch pamConfig.ResourceType {