[Analyze 2845] cdm sql audit logging (#2914)

* feat(analytics): add file-based SQL audit logging

* feat(analytics): cover direct CDM SQL paths

* fix(analytics): allow audit writes in Trex runtime

* refactor(analytics): fix audit log directory

* feat(analytics): support configurable audit output

* refactor(trex): remove audit directory chmod

* refactor(analytics-svc): remove SQL audit hash

* Include dataset config metadata in SQL audit logs

Author: jerome-ng

docker-compose.yml

@@ -17,6 +17,7 @@ volumes:
   hana-data:
   supabase-storage-data:
   hades-data:
+  audit-logs:
 
 x-envs:
   env_converter: &x-converter
@@ -95,6 +96,7 @@ services:
       - trex:/usr/src/data
       - cdw-config-cachedb-data-1:/usr/src/cdw_data/dynamically_generated
       - hades-data:/data/hades
+      - audit-logs:/var/log/d2e/audit
     ports:
       - "8090:8080"
     healthcheck:
@@ -445,6 +447,8 @@ services:
       CACHE_TASK_TIMEOUT: ${CACHE_TASK_TIMEOUT:-10800}
       ANALYTICS_STREAMING_CHUNK_SIZE_BY_DIALECT: '${ANALYTICS_STREAMING_CHUNK_SIZE_BY_DIALECT:-{"hana": 10000}}'
       IS_AUDIT_LOG_ENABLED: ${IS_AUDIT_LOG_ENABLED:-false}
+      IS_CDM_SQL_AUDIT_LOG_ENABLED: ${IS_CDM_SQL_AUDIT_LOG_ENABLED:-false}
+      AUDIT_LOG_TO_CONSOLE: ${AUDIT_LOG_TO_CONSOLE:-false}
       ANALYTICS_HANA_STREAMING_ENABLED: ${ANALYTICS_HANA_STREAMING_ENABLED:-false}
       # TREX_DEBUG_GC: ${TREX_DEBUG_GC:-0}
       # RUST_LOG: ${RUST_LOG:-info}

plugins/functions/analytics-svc/src/api/controllers/cohort.ts

@@ -18,6 +18,7 @@ import PortalServerAPI from "../PortalServerAPI";
 import { convertIFRToExtCohort } from "../../ifr-to-extcohort/main";
 import { dataflowRequest } from "../../utils/DataflowMgmtProxy";
 import { env } from "../../env";
+import { createCdmSqlAuditContext } from "../../utils/CdmSqlAuditLogger.ts";
 
 const language = "en";
 
@@ -314,6 +315,17 @@ export async function createCohort(req: IMRIRequest, res: Response) {
                     datasetId: req.selectedstudyDbMetadata.id,
                     token: req.headers.authorization,
                     dbCredential: req.dbCredentials.studyAnalyticsCredential,
+                    auditContext: createCdmSqlAuditContext({
+                        request: req,
+                        actorId: getUser(req)?.getUser() ?? "unknown",
+                        databaseCode:
+                            req.dbCredentials.studyAnalyticsCredential.code,
+                        databaseDialect:
+                            req.dbCredentials.studyAnalyticsCredential.dialect,
+                        databaseEngine: "hana",
+                        schemaName:
+                            req.dbCredentials.studyAnalyticsCredential.schema,
+                    }),
                 }
             );
         } else {

plugins/functions/analytics-svc/src/api/controllers/datasetFilter.ts

@@ -11,6 +11,10 @@ import QueryObject = qo.QueryObject;
 import { dataflowRequest } from "../../utils/DataflowMgmtProxy";
 import { FilterScopeQueryBuilder } from "../../utils/dataset-filter/filter-scope-query-builder";
 import { FilterQueryBuilder } from "../../utils/dataset-filter/query-builder";
+import {
+    createCdmSqlAuditConnection,
+    createCdmSqlAuditContext,
+} from "../../utils/CdmSqlAuditLogger.ts";
 import {
     DatabaseSchemaMap,
     IDatasetFilterScopesDto,
@@ -204,6 +208,36 @@ const createDatabaseSchemaMap = (
     }, {});
 };
 
+const getDatasetFilterDbConnection = async (databaseName, req) => {
+    const userObj = getUser(req);
+    const dbCredentials = getDatabaseCredentials()[databaseName];
+    const connection =
+        await dbConnectionUtil.DBConnectionUtil.getDBConnection({
+            credentials: dbCredentials,
+            schemaName: dbCredentials.schemaName,
+            vocabSchemaName: dbCredentials.vocabSchemaName,
+            userObj,
+        });
+    const databaseEngine =
+        dbCredentials.dialect === "hana"
+            ? "hana"
+            : dbCredentials.dialect === "postgresql"
+            ? "postgresql"
+            : "duckdb";
+
+    return createCdmSqlAuditConnection(
+        connection,
+        createCdmSqlAuditContext({
+            request: req,
+            actorId: userObj?.getUser() ?? "unknown",
+            databaseCode: databaseName,
+            databaseDialect: dbCredentials.dialect,
+            databaseEngine,
+            schemaName: dbCredentials.schema,
+        })
+    );
+};
+
 const queryFilterScopes = async (
     databaseName,
     schemas,
@@ -213,16 +247,7 @@ const queryFilterScopes = async (
     const builder = new FilterScopeQueryBuilder(dialect, schemas);
     const query = builder.build();
 
-    let userObj;
-    userObj = getUser(req);
-    const dbCredentials = getDatabaseCredentials()[databaseName];
-    const dbConnection =
-        await dbConnectionUtil.DBConnectionUtil.getDBConnection({
-            credentials: dbCredentials,
-            schemaName: dbCredentials.schemaName,
-            vocabSchemaName: dbCredentials.vocabSchemaName,
-            userObj,
-        });
+    const dbConnection = await getDatasetFilterDbConnection(databaseName, req);
 
     const queryobject = QueryObject.format(query);
     var rangeResults = queryobject.executeQuery(dbConnection);
@@ -261,17 +286,7 @@ const filter = async (databaseName, schemas, dialect, filterParams, req) => {
     const builder = new FilterQueryBuilder(dialect, schemas, filterParams);
     const query = builder.build();
 
-    let userObj;
-    userObj = getUser(req);
-    // retrieve the connection
-    const dbCredentials = getDatabaseCredentials()[databaseName];
-    const dbConnection =
-        await dbConnectionUtil.DBConnectionUtil.getDBConnection({
-            credentials: dbCredentials,
-            schemaName: dbCredentials.schemaName,
-            vocabSchemaName: dbCredentials.vocabSchemaName,
-            userObj,
-        });
+    const dbConnection = await getDatasetFilterDbConnection(databaseName, req);
 
     const queryobject = QueryObject.format(query);
 

plugins/functions/analytics-svc/src/env.ts

@@ -71,6 +71,8 @@ const Env = z
         DB_SVC__PATH: z.string().optional(),
         DB_SVC__PORT: z.string().optional(),
         IS_AUDIT_LOG_ENABLED: z.string().optional(),
+        IS_CDM_SQL_AUDIT_LOG_ENABLED: z.string().optional(),
+        AUDIT_LOG_TO_CONSOLE: z.string().optional(),
         ANALYTICS_HANA_STREAMING_ENABLED: z.string(),
         ANALYTICS_STREAMING_CHUNK_SIZE_BY_DIALECT: z
             .string()

plugins/functions/analytics-svc/src/main.ts

@@ -32,6 +32,11 @@ import PortalServerAPI from "./api/PortalServerAPI";
 import { env } from "./env";
 import addCorrelationIDToHeader from "./middleware/AddCorrelationId.ts";
 import { parseValueForPrototypePollutingAssignment } from "./utils/utils";
+import { getAuditUserIdFromRequest } from "./utils/AuditLogger.ts";
+import {
+    createCdmSqlAuditConnection,
+    createCdmSqlAuditContext,
+} from "./utils/CdmSqlAuditLogger.ts";
 dotenv.config();
 const log = console; //Logger.CreateLogger("analytics-log");
 const mriConfigConnection = new MriConfigConnection(
@@ -151,22 +156,43 @@ const initRoutes = async (app: express.Application) => {
                     credentials = req.dbCredentials.studyAnalyticsCredential;
                 }
 
-                if (credentials.dialect === ANALYTICS_DB_DIALECTS.HANA) {
-                    req.dbConnections = await getDBConnections({
-                        analyticsCredentials: credentials,
-                        userObj,
-                        sessionVariables: {
-                            paConfigId: req.paConfigId,
-                            paConfigVersion: req.paConfigVersion,
-                            cdmConfigId: req.cdmConfigId,
-                            cdmConfigVersion: req.cdmConfigVersion,
-                        },
-                    });
-                } else {
-                    req.dbConnections = getTrexDbConnection({
-                        analyticsCredentials: credentials,
-                    });
-                }
+                const dbConnections =
+                    credentials.dialect === ANALYTICS_DB_DIALECTS.HANA
+                        ? await getDBConnections({
+                              analyticsCredentials: credentials,
+                              userObj,
+                              sessionVariables: {
+                                  paConfigId: req.paConfigId,
+                                  paConfigVersion: req.paConfigVersion,
+                                  cdmConfigId: req.cdmConfigId,
+                                  cdmConfigVersion: req.cdmConfigVersion,
+                              },
+                          })
+                        : getTrexDbConnection({
+                              analyticsCredentials: credentials,
+                          });
+
+                req.dbConnections = {
+                    ...dbConnections,
+                    analyticsConnection: createCdmSqlAuditConnection(
+                        dbConnections.analyticsConnection,
+                        createCdmSqlAuditContext({
+                            request: req,
+                            actorId:
+                                userObj?.getUser() ??
+                                getAuditUserIdFromRequest(req) ??
+                                "unknown",
+                            databaseCode: credentials.code,
+                            databaseDialect: credentials.dialect,
+                            databaseEngine:
+                                credentials.dialect ===
+                                ANALYTICS_DB_DIALECTS.HANA
+                                    ? "hana"
+                                    : "duckdb",
+                            schemaName: credentials.schema,
+                        })
+                    ),
+                };
             }
 
             next();

plugins/functions/analytics-svc/src/mri/endpoint/CohortEndpoint.ts

@@ -9,6 +9,10 @@ import CreateLogger = Logger.CreateLogger;
 import QueryObject = qo.QueryObject;
 import { Connection as connLib } from "@alp/alp-base-utils";
 import ConnectionInterface = connLib.ConnectionInterface;
+import {
+    executeWithCdmSqlAudit,
+    type CdmSqlAuditContext,
+} from "../../utils/CdmSqlAuditLogger.ts";
 const logger = CreateLogger("analytics-log");
 
 declare const Trex: any;
@@ -562,7 +566,12 @@ export class CohortEndpoint {
         cohortDefinitionId: number,
         cohort: CohortType,
         queryObject: QueryObjectType,
-        metadata: { datasetId: string; token: string; dbCredential: any }
+        metadata: {
+            datasetId: string;
+            token: string;
+            dbCredential: any;
+            auditContext: CdmSqlAuditContext;
+        }
     ) {
         try {
             const partialInsertQuery = QueryObject.formatDict(
@@ -608,9 +617,16 @@ export class CohortEndpoint {
                     Number(cohortDefinitionId),
                     JSON.stringify(sessionVars)
                 );
-                const result = await materializeQuery.executeQuery<{ processed_rows: number }>(
-                    memConn
-                );
+                const result = await executeWithCdmSqlAudit({
+                    context: metadata.auditContext,
+                    operation: "executeQuery",
+                    sql: translatedSql,
+                    parameters: preparedQuery.placeholders ?? [],
+                    execute: () =>
+                        materializeQuery.executeQuery<{
+                            processed_rows: number;
+                        }>(memConn),
+                });
                 return result?.data?.[0]?.processed_rows;
             } finally {
                 if (typeof memConn?.close === "function") {

plugins/functions/analytics-svc/src/utils/AuditEventWriter.ts

@@ -0,0 +1,112 @@
+import { env } from "../env.ts";
+
+export const AUDIT_LOG_DIRECTORY = "/var/log/d2e/audit";
+export const PATIENT_ACCESS_AUDIT_FILE = "patient-access.ndjson";
+export const CDM_SQL_AUDIT_FILE = "cdm-sql-access.ndjson";
+
+export type AuditEvent = Record<string, unknown>;
+
+export type AuditTransport = {
+    audit(message: unknown, user: string): void | Promise<void>;
+};
+
+export interface AuditEventWriter {
+    append(fileName: string, event: AuditEvent): Promise<void>;
+}
+
+const encoder = new TextEncoder();
+
+function getAuditFilePath(directory: string, fileName: string): string {
+    if (!/^[a-z0-9][a-z0-9.-]*\.ndjson$/.test(fileName)) {
+        throw new Error("Invalid audit file name");
+    }
+
+    return `${directory.replace(/\/+$/, "")}/${fileName}`;
+}
+
+async function chmodWhenAvailable(path: string, mode: number): Promise<void> {
+    try {
+        await Deno.chmod(path, mode);
+    } catch (error) {
+        if (
+            error instanceof Error &&
+            error.name === "PermissionDenied" &&
+            error.message === "Deno.chmod is blocklisted"
+        ) {
+            return;
+        }
+        throw error;
+    }
+}
+
+export class NdjsonAuditEventWriter implements AuditEventWriter {
+    public constructor(private readonly directory = AUDIT_LOG_DIRECTORY) {}
+
+    public async append(fileName: string, event: AuditEvent): Promise<void> {
+        await Deno.mkdir(this.directory, { recursive: true, mode: 0o750 });
+        await chmodWhenAvailable(this.directory, 0o750);
+
+        const path = getAuditFilePath(this.directory, fileName);
+        const bytes = encoder.encode(`${JSON.stringify(event)}\n`);
+        const file = await Deno.open(path, {
+            append: true,
+            create: true,
+            mode: 0o640,
+            write: true,
+        });
+
+        try {
+            await chmodWhenAvailable(path, 0o640);
+            // Keep one event in one append operation so concurrent Trex workers
+            // cannot interleave portions of separate NDJSON records.
+            const bytesWritten = await file.write(bytes);
+            if (bytesWritten !== bytes.length) {
+                throw new Error("Incomplete audit event append");
+            }
+        } finally {
+            file.close();
+        }
+    }
+}
+
+export class ConsoleAuditEventWriter implements AuditEventWriter {
+    public append(_fileName: string, event: AuditEvent): Promise<void> {
+        console.info(JSON.stringify(event));
+        return Promise.resolve();
+    }
+}
+
+export function createAuditEventWriter(): AuditEventWriter {
+    return env.AUDIT_LOG_TO_CONSOLE?.toLowerCase() === "true"
+        ? new ConsoleAuditEventWriter()
+        : new NdjsonAuditEventWriter();
+}
+
+export function createPatientAccessAuditTransport(
+    writer: AuditEventWriter = createAuditEventWriter()
+): AuditTransport {
+    return {
+        async audit(message: unknown, user: string): Promise<void> {
+            const eventData =
+                message && typeof message === "object"
+                    ? (message as AuditEvent)
+                    : { message: String(message) };
+
+            try {
+                await writer.append(PATIENT_ACCESS_AUDIT_FILE, {
+                    ...eventData,
+                    schemaVersion: 1,
+                    eventType: "patient.access",
+                    actor: {
+                        type: "user",
+                        id: user,
+                    },
+                });
+            } catch (_error) {
+                // Audit persistence is fail-open. Never echo the event payload
+                // or patient identifier into operational container logs.
+                console.error("Patient audit event could not be written");
+            }
+        },
+    };
+}