Monitoring sources integrations sync (#233311)

## Summary 
This PR introduces a new source initialisation service for privilege
monitoring, adding logic to handle both first-time initialisation (when
no sources exist) and re-initialisation (where existing sources are
updated and missing ones are created).

### Testing 
1. Enable privileged user monitoring in advanced settings
2. Enable experimental feature flag for integrations sync:
```
 xpack.securitySolution:
  enableExperimental: [integrationsSyncEnabled]
```
3. Dev tools 
4. `POST kbn:/api/entity_analytics/monitoring/engine/init {}`
5. `GET kbn:/api/entity_analytics/monitoring/entity_source/list`
6. Should have two integrations: okta, AD and the default index
installed.

Should then be able to call init again and see the same entity sources
from the list endpoint.
Should also be able to confirm integrations sync works only if
experimental features flag is set

Author: CAWilson94

x-pack/solutions/security/plugins/security_solution/common/experimental_features.ts

@@ -242,6 +242,11 @@ export const allowedExperimentalValues = Object.freeze({
    */
   privilegedUserMonitoringDisabled: false,
 
+  /**
+   * Enables Integrations Sync for Privileged User Monitoring
+   */
+  integrationsSyncEnabled: false,
+
   /**
    * Disables the siem migrations feature
    */

x-pack/solutions/security/plugins/security_solution/server/lib/entity_analytics/privilege_monitoring/data_sources/constants.test.ts

@@ -0,0 +1,89 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { defaultMonitoringUsersIndex } from '../../../../../common/entity_analytics/privileged_user_monitoring/utils';
+import type { IntegrationType } from './constants';
+import {
+  AD_ADMIN_ROLES,
+  getMatchersFor,
+  getStreamPatternFor,
+  INTEGRATION_MATCHERS_DETAILED,
+  INTEGRATION_TYPES,
+  integrationsSourceIndex,
+  OKTA_ADMIN_ROLES,
+  STREAM_INDEX_PATTERNS,
+} from './constants';
+
+describe('constants', () => {
+  const baseIndex = `entity_analytics.privileged_monitoring`;
+  const baseMonitoringUsersIndex = '.entity_analytics.monitoring';
+  it('should export all constants', () => {
+    expect(getMatchersFor).toBeDefined();
+  });
+  it('should have correct OKTA_ADMIN_ROLES', () => {
+    expect(OKTA_ADMIN_ROLES).toMatchInlineSnapshot(`
+  Array [
+    "Super Administrator",
+    "Organization Administrator",
+    "Group Administrator",
+    "Application Administrator",
+    "Mobile Administrator",
+    "Help Desk Administrator",
+    "Report Administrator",
+    "API Access Management Administrator",
+    "Group Membership Administrator",
+    "Read-only Administrator",
+  ]
+`);
+  });
+
+  it('should have correct AD_ADMIN_ROLES', () => {
+    expect(AD_ADMIN_ROLES).toEqual(['Domain Admins', 'Enterprise Admins']);
+    expect(AD_ADMIN_ROLES.length).toBe(2);
+  });
+
+  it('should have correct INTEGRATION_MATCHERS_DETAILED', () => {
+    expect(INTEGRATION_MATCHERS_DETAILED.okta.values).toEqual(OKTA_ADMIN_ROLES);
+    expect(INTEGRATION_MATCHERS_DETAILED.ad.values).toEqual(AD_ADMIN_ROLES);
+    expect(INTEGRATION_MATCHERS_DETAILED.okta.fields).toEqual(['user.roles']);
+    expect(INTEGRATION_MATCHERS_DETAILED.ad.fields).toEqual(['user.roles']);
+  });
+
+  it('getMatchersFor returns correct matcher array', () => {
+    expect(getMatchersFor('okta')).toEqual([INTEGRATION_MATCHERS_DETAILED.okta]);
+    expect(getMatchersFor('ad')).toEqual([INTEGRATION_MATCHERS_DETAILED.ad]);
+  });
+
+  it('IntegrationType type should only allow okta and ad', () => {
+    const types: IntegrationType[] = ['okta', 'ad'];
+    expect(types).toEqual(INTEGRATION_TYPES);
+  });
+
+  it('should generate defaultMonitoringUsersIndex', () => {
+    expect(defaultMonitoringUsersIndex('default')).toBe(`${baseIndex}.default`);
+    expect(defaultMonitoringUsersIndex('space1')).toBe(`${baseIndex}.space1`);
+  });
+
+  it('should generate integrationsSourceIndex', () => {
+    expect(integrationsSourceIndex('default', 'okta')).toBe(
+      `${baseMonitoringUsersIndex}.sources.okta-default`
+    );
+    expect(integrationsSourceIndex('space1', 'ad')).toBe(
+      `${baseMonitoringUsersIndex}.sources.ad-space1`
+    );
+  });
+
+  it('should generate correct stream index patterns', () => {
+    expect(STREAM_INDEX_PATTERNS.okta('default')).toBe('logs-entityanalytics_okta.user-default');
+    expect(STREAM_INDEX_PATTERNS.ad('space1')).toBe('logs-entityanalytics_ad.user-space1');
+  });
+
+  it('getStreamPatternFor returns correct pattern', () => {
+    expect(getStreamPatternFor('okta', 'default')).toBe('logs-entityanalytics_okta.user-default');
+    expect(getStreamPatternFor('ad', 'space1')).toBe('logs-entityanalytics_ad.user-space1');
+  });
+});

x-pack/solutions/security/plugins/security_solution/server/lib/entity_analytics/privilege_monitoring/data_sources/constants.ts

@@ -0,0 +1,57 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { PRIVMON_BASE_INDEX_NAME } from '../../../../../common/constants';
+
+export interface Matcher {
+  values: string[];
+  fields: string[];
+}
+
+export const OKTA_ADMIN_ROLES: string[] = [
+  'Super Administrator',
+  'Organization Administrator',
+  'Group Administrator',
+  'Application Administrator',
+  'Mobile Administrator',
+  'Help Desk Administrator',
+  'Report Administrator',
+  'API Access Management Administrator',
+  'Group Membership Administrator',
+  'Read-only Administrator',
+];
+
+export const AD_ADMIN_ROLES: string[] = ['Domain Admins', 'Enterprise Admins'];
+
+export const INTEGRATION_MATCHERS_DETAILED: Record<IntegrationType, Matcher> = {
+  okta: { fields: ['user.roles'], values: OKTA_ADMIN_ROLES },
+  ad: { fields: ['user.roles'], values: AD_ADMIN_ROLES },
+};
+
+export const getMatchersFor = (integration: IntegrationType): Matcher[] => [
+  INTEGRATION_MATCHERS_DETAILED[integration],
+];
+// TODO: this should be index source? If so, can follow pattern outlined below.
+// e.g. .entity_analytics.monitoring.sources.index-<space>
+
+// .entity_analytics.monitoring.sources.okta-<space>
+export const integrationsSourceIndex = (namespace: string, integrationName: string) =>
+  `${PRIVMON_BASE_INDEX_NAME}.sources.${integrationName}-${namespace}`;
+
+export const PRIVILEGE_MONITORING_PRIVILEGE_CHECK_API =
+  '/api/entity_analytics/monitoring/privileges/privileges';
+
+export const INTEGRATION_TYPES = ['okta', 'ad'] as const;
+export type IntegrationType = (typeof INTEGRATION_TYPES)[number];
+
+export const STREAM_INDEX_PATTERNS: Record<IntegrationType, (namespace: string) => string> = {
+  okta: (namespace) => `logs-entityanalytics_okta.user-${namespace}`,
+  ad: (namespace) => `logs-entityanalytics_ad.user-${namespace}`,
+};
+
+export const getStreamPatternFor = (integration: IntegrationType, namespace: string): string =>
+  STREAM_INDEX_PATTERNS[integration](namespace);

x-pack/solutions/security/plugins/security_solution/server/lib/entity_analytics/privilege_monitoring/data_sources/index.ts

@@ -0,0 +1,8 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+export * from './constants';

x-pack/solutions/security/plugins/security_solution/server/lib/entity_analytics/privilege_monitoring/engine/data_client.ts

@@ -16,6 +16,7 @@ import type {
   KibanaRequest,
 } from '@kbn/core/server';
 import type { TaskManagerStartContract } from '@kbn/task-manager-plugin/server';
+import type { ExperimentalFeatures } from '../../../../../common';
 import type { MonitoringEngineComponentResource } from '../../../../../common/api/entity_analytics';
 import { getPrivilegedMonitorUsersIndex } from '../../../../../common/entity_analytics/privileged_user_monitoring/utils';
 import type { ApiKeyManager } from '../auth/api_key';
@@ -36,6 +37,8 @@ export interface PrivilegeMonitoringGlobalDependencies {
 
   apiKeyManager?: ApiKeyManager;
   taskManager?: TaskManagerStartContract;
+
+  experimentalFeatures?: ExperimentalFeatures;
 }
 
 /**
@@ -46,6 +49,7 @@ export class PrivilegeMonitoringDataClient {
 
   constructor(public readonly deps: PrivilegeMonitoringGlobalDependencies) {
     this.index = getPrivilegedMonitorUsersIndex(deps.namespace);
+    this.deps.experimentalFeatures = deps.experimentalFeatures;
   }
 
   public getScopedSoClient(request: KibanaRequest, options?: SavedObjectsClientProviderOptions) {

x-pack/solutions/security/plugins/security_solution/server/lib/entity_analytics/privilege_monitoring/engine/initialisation_service.test.ts

@@ -19,6 +19,7 @@ import type { InitialisationService } from './initialisation_service';
 import { createInitialisationService } from './initialisation_service';
 import { MonitoringEngineComponentResourceEnum } from '../../../../../common/api/entity_analytics';
 import { PrivilegeMonitoringEngineActions } from '../auditing/actions';
+import { mockGlobalState } from '../../../../../public/common/mock';
 
 const mockUpsertIndex = jest.fn();
 jest.mock('./elasticsearch/indices', () => {
@@ -73,6 +74,7 @@ describe('Privileged User Monitoring: Index Sync Service', () => {
     auditLogger: auditMock,
     telemetry: telemetryMock,
     savedObjects: savedObjectServiceMock,
+    experimentalFeatures: mockGlobalState.app.enableExperimental,
   };
 
   let initService: InitialisationService;

x-pack/solutions/security/plugins/security_solution/server/lib/entity_analytics/privilege_monitoring/engine/initialisation_service.ts

@@ -26,16 +26,19 @@ import {
   PrivilegeMonitoringEngineDescriptorClient,
 } from '../saved_objects';
 import { startPrivilegeMonitoringTask } from '../tasks/privilege_monitoring_task';
+import { createInitialisationSourcesService } from './initialisation_sources_service';
 
 export type InitialisationService = ReturnType<typeof createInitialisationService>;
 export const createInitialisationService = (dataClient: PrivilegeMonitoringDataClient) => {
   const { deps } = dataClient;
   const { taskManager } = deps;
+
   if (!taskManager) {
     throw new Error('Task Manager is not available');
   }
 
   const IndexService = createPrivmonIndexService(dataClient);
+  const InitSourceCreationService = createInitialisationSourcesService(dataClient);
 
   const init = async (
     soClient: SavedObjectsClientContract
@@ -56,11 +59,17 @@ export const createInitialisationService = (dataClient: PrivilegeMonitoringDataC
       MonitoringEngineComponentResourceEnum.privmon_engine,
       'Initializing privilege monitoring engine'
     );
-
     const descriptor = await descriptorClient.init();
-    dataClient.log('debug', `Initialized privileged monitoring engine saved object`);
+    dataClient.log('info', `Initialized privileged monitoring engine saved object`);
+
+    if (deps.experimentalFeatures?.integrationsSyncEnabled ?? false) {
+      // upsert index AND integration sources
+      await InitSourceCreationService.upsertSources(monitoringIndexSourceClient);
+    } else {
+      // upsert ONLY index source
+      await createOrUpdateDefaultDataSource(monitoringIndexSourceClient);
+    }
 
-    await createOrUpdateDefaultDataSource(monitoringIndexSourceClient);
     try {
       dataClient.log('debug', 'Creating privilege user monitoring event.ingested pipeline');
       await IndexService.createIngestPipelineIfDoesNotExist();
@@ -148,6 +157,7 @@ export const createInitialisationService = (dataClient: PrivilegeMonitoringDataC
       dataClient.log('info', 'Creating default index source for privilege monitoring.');
 
       try {
+        // TODO: failing test, empty sources array. FIX
         const indexSourceDescriptor = monitoringIndexSourceClient.create(defaultIndexSource);
 
         dataClient.log(
@@ -168,6 +178,5 @@ export const createInitialisationService = (dataClient: PrivilegeMonitoringDataC
       }
     }
   };
-
   return { init };
 };

x-pack/solutions/security/plugins/security_solution/server/lib/entity_analytics/privilege_monitoring/engine/initialisation_sources_service.test.ts

@@ -0,0 +1,104 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import type { AuditLogger } from '@kbn/core/server';
+import type { PrivilegeMonitoringGlobalDependencies } from './data_client';
+import { PrivilegeMonitoringDataClient } from './data_client';
+import {
+  createInitialisationSourcesService,
+  type InitialisationSourcesService,
+} from './initialisation_sources_service';
+import {
+  elasticsearchServiceMock,
+  loggingSystemMock,
+  analyticsServiceMock,
+  savedObjectsServiceMock,
+} from '@kbn/core/server/mocks';
+import type { TaskManagerStartContract } from '@kbn/task-manager-plugin/server';
+import { MonitoringEntitySourceDescriptorClient } from '../saved_objects';
+import { mockGlobalState } from '../../../../../public/common/mock';
+
+jest.mock('../saved_objects', () => {
+  const mockEngineDescriptorInit = jest.fn();
+  const mockFind = jest.fn().mockResolvedValue({
+    saved_objects: [],
+    total: 0,
+  });
+  return {
+    MonitoringEntitySourceDescriptorClient: jest.fn().mockImplementation(() => ({
+      findByIndex: jest.fn(),
+      create: jest.fn(),
+      bulkCreate: jest.fn(),
+      update: jest.fn(),
+      find: mockFind,
+      findAll: jest.fn(),
+      bulkUpsert: jest.fn(),
+    })),
+    PrivilegeMonitoringEngineDescriptorClient: jest.fn().mockImplementation(() => ({
+      init: mockEngineDescriptorInit,
+      update: jest.fn(),
+    })),
+  };
+});
+
+describe('createInitialisationSourcesService', () => {
+  const clusterClientMock = elasticsearchServiceMock.createScopedClusterClient();
+  const loggerMock = loggingSystemMock.createLogger();
+  const auditMock = { log: jest.fn().mockReturnValue(undefined) } as unknown as AuditLogger;
+  const telemetryMock = analyticsServiceMock.createAnalyticsServiceSetup();
+
+  const savedObjectServiceMock = savedObjectsServiceMock.createStartContract();
+  const dataClientDeps: PrivilegeMonitoringGlobalDependencies = {
+    logger: loggerMock,
+    clusterClient: clusterClientMock,
+    namespace: 'default',
+    kibanaVersion: '9.0.0',
+    taskManager: {} as TaskManagerStartContract,
+    auditLogger: auditMock,
+    telemetry: telemetryMock,
+    savedObjects: savedObjectServiceMock,
+    experimentalFeatures: mockGlobalState.app.enableExperimental,
+  };
+
+  let dataClient: PrivilegeMonitoringDataClient;
+  let initSourcesService: InitialisationSourcesService;
+  let monitoringDescriptorClient: MonitoringEntitySourceDescriptorClient;
+
+  beforeEach(() => {
+    jest.clearAllMocks();
+    dataClient = new PrivilegeMonitoringDataClient(dataClientDeps);
+    initSourcesService = createInitialisationSourcesService(dataClient);
+    const mockLog = jest.fn();
+    dataClient.log = mockLog;
+    monitoringDescriptorClient = new (MonitoringEntitySourceDescriptorClient as jest.Mock)();
+  });
+
+  it('should create sources when none exist', async () => {
+    (monitoringDescriptorClient.findAll as jest.Mock).mockResolvedValue([]);
+    await initSourcesService.upsertSources(monitoringDescriptorClient);
+    expect(monitoringDescriptorClient.bulkCreate).toHaveBeenCalledTimes(1);
+    expect(monitoringDescriptorClient.update).not.toHaveBeenCalled();
+  });
+
+  it('should update sources when they already exist', async () => {
+    const existingSources = [
+      { id: '1', name: '.entity_analytics.monitoring.users-default' },
+      { id: '2', name: '.entity_analytics.monitoring.sources.okta-default' },
+      { id: '3', name: '.entity_analytics.monitoring.sources.ad-default' },
+    ];
+    (monitoringDescriptorClient.findAll as jest.Mock).mockResolvedValue(existingSources);
+    await initSourcesService.upsertSources(monitoringDescriptorClient);
+    expect(monitoringDescriptorClient.bulkUpsert).toHaveBeenCalledTimes(1);
+  });
+
+  it('should create missing sources and update existing ones', async () => {
+    const existingSources = [{ id: '3', name: '.entity_analytics.monitoring.sources.ad-default' }];
+    (monitoringDescriptorClient.findAll as jest.Mock).mockResolvedValue(existingSources);
+    await initSourcesService.upsertSources(monitoringDescriptorClient);
+    expect(monitoringDescriptorClient.bulkUpsert).toHaveBeenCalledTimes(1);
+  });
+});