Generalising update detection to work with multiple integrations (elastic#237792)

### Summary 
This PR introduces support for multiple integrations by gentrifying the
query for update detection matchers fields, instead of just the okta
'roles'. The changes also include updating constants and bringing in AD
when loading in Saved Objects - monitoring sources.

**Index and Matcher Logic**

* Updated the logic for determining the last full sync markers index to
use the AD users index when the integration is AD, ensuring correct data
source usage.
* Modified the privileged user search query to dynamically include the
relevant matcher field (such as `user.group.name` for AD) in the
`_source` fields, supporting flexible matching across integrations.
[[1]](diffhunk://#diff-dae1d0e0452709a5c2d74497a2c3a2897584faeff73e3ce3f785044cde4673bcR54)
[[2]](diffhunk://#diff-dae1d0e0452709a5c2d74497a2c3a2897584faeff73e3ce3f785044cde4673bcL74-R75)
* Updated the pattern matcher service to pass the matcher field for AD
integrations when building the search body, ensuring correct field
matching for privileged status detection.

## How To Test 

If stuck at any point, I posted all the commands I have been using in a
google doc "paste bin" for reference [ --->
here<---](https://docs.google.com/document/d/1SG5NvSqiFx_-u_I9-TXI1JL7mF2s4pTrJaezatzhHFg/edit?tab=t.0#bookmark=id.bh4tzq75v6z)

Some commands are duplicated, but I hope something helps 🤘

### Update Detection 
1. Head over to Document Generator and pull the ad-data branch -
https://github.com/CAWilson94/security-documents-generator/tree/ad-data
2.Run Kibana and ES THEN generate a number of users (100 seems good)
from `yarn start privileged-user-monitoring`, choosing the integrations
option only.
3. Open Kibana dev tools - run engine init `POST
kbn:/api/entity_analytics/monitoring/engine/init {}`
4. You will get an error in the console about full sync - no such index
for okta entity-default, because these have not been created. Ignore
this, or create that index. Up to you.
5. Check the monitoring list and make sure there are privileged users
from your sync. Either through minitoring list endpoint or the internal
index itself:

`GET kbn:/api/entity_analytics/monitoring/users/list` OR ` POST
.entity_analytics.monitoring.users-default/_search`

From the index, the output looks like this:  
```
    "hits": [
      {
        "_index": ".entity_analytics.monitoring.users-default",
        "_id": "4vAjvpkBcs32I8L9OZG9",
        "_score": 1,
        "_source": {
          "entity_analytics_monitoring": {
            "labels": [
              {
                "field": "user.group.name",
                "source": "ff1ba7ff-e8cb-47e8-891d-787100ab7c1f",
                "value": "Enterprise Admins"
              }
            ]
          },
          "@timestamp": "2025-10-07T10:06:43.637Z",
          "event": {
            "ingested": "2025-10-07T10:06:43.639331Z"
          },
          "user": {
            "is_privileged": true,
            "name": "Alexandra.Abernathy",
            "entity": {
              "attributes": {
                "Privileged": true
              }
            }
          },
          "labels": {
            "source_ids": [
              "ff1ba7ff-e8cb-47e8-891d-787100ab7c1f"
            ],
            "sources": [
              "entity_analytics_integration"
            ]
          }
        }
      },
```

### Deletion Detection 
**Getting Some Data**
1. Head over to Document Generator and pull the ad-data branch -
https://github.com/CAWilson94/security-documents-generator/tree/ad-data
2.Run Kibana and ES THEN generate a number of users (100 seems good)
from yarn start privileged-user-monitoring, choosing the integrations
option only.
**Generating FIRST set of fullSync Markers**
1. Have a look at the ad data added, and get the latest and oldest
timestamps:
```
GET logs-entityanalytics_ad.user-default/_search
{
  "size": 1,
  "sort": [{ "@timestamp": "desc" }]
}

GET logs-entityanalytics_ad.user-default/_search
{
  "size": 1,
  "sort": [{ "@timestamp": "asc" }]
}
```

To create the fullSync Marker wrapper documents outside of these times,
make a started event right before the oldest and a completed event right
after the latest:

e.g. something like: 
```
POST logs-entityanalytics_ad.user-default/_bulk
{ "create": {} }
{ "@timestamp": "2025-09-28T04:00:19.836Z", // CHANGE THIS 
  "event": {
    "action": "started",
    "dataset": "entityanalytics_ad.user",
    "kind": "asset",
    "category": ["iam"],
    "type": ["info"]
  }
}
{ "create": {} }
{ "@timestamp": "2025-09-28T05:53:19.836Z", // CHANGE THIS 
  "event": {
    "action": "completed",
    "dataset": "entityanalytics_ad.user",
    "kind": "asset",
    "category": ["iam"],
    "type": ["info"]
  }
}
```

**Initalize the Engine, Confirm SO Markers:**
1. (optional) shorten the INTERVAL for the engine to something like 3m,
to give time to edit data between syncs:
_...entity_analytics/privilege_monitoring/constants.ts_
2. Initalize engine: `POST
kbn:/api/entity_analytics/monitoring/engine/init{}`
3. On FIRST RUN: the engine will: 
    * Look for a full sync in SO
    * If missing, fallback to the latest complete marker in the index. 
    VERIFY this by looking at the SO: 
    ```
    GET kbn:/api/entity_analytics/monitoring/entity_source/list
    ```
    
Expected Shape: 
    
    ```
    
    "integrations": {
    "syncMarkerIndex": "logs-entityanalytics_ad.user-default",
    "syncData": {
    "lastUpdateProcessed": "…",
    "lastFullSync": "2025-09-28T05:53:19.836Z" ...
     ```
     
  **No full sync - engine should skip**
* Basically, don't add a second set of markers here. It should do
nothing, logs should say _"No new full sync for source.."_
  
  **Full sync with NO users in window - ALL Stale**
1. Add new start and complete markers after all the users timestamps,
that wrap NO user docs.
2. Next engine run will detect a new full sync - find no users between
the markers, and treat all CURRENT privileged (in monitoring index)
users, as stale - soft delete.
  
  Validate: 
  
  ```
  POST .entity_analytics.monitoring.users-default/_search
{
  "size": 100,
"_source":
["user.*","labels.*","entity_analytics_monitoring.*","@timestamp","event.ingested"],
  "query": {
    "term": { "user.is_privileged": true }
  }
}
```

Expected: 0 hits

**Full sync with subset - only missing are stale**
1. Create Markers that wrap a time window you will populate 
2. Create some new documents in AD within these time windows, ensuring they have the same username as some already in the monitoring index. 
3. Check monitoring index after sync, see only missing names are soft deleted. 


## Testing Notes and Gotcha's 
Because this has been a long testing journey: 

* Please make sure when you copy pasta these examples, you are changing timestamps to work with the markers and users created. 

And thank you for testing 🧪 🥳

---------

Co-authored-by: Mark Hopkin <[email protected]>

Author: CAWilson94

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

@@ -275,7 +275,6 @@ export const bulkUpsertOperationsFactoryShared =
 
 export const makeIntegrationOpsBuilder = (dataClient: PrivilegeMonitoringDataClient) => {
   const buildOps = bulkUpsertOperationsFactoryShared(dataClient);
-
   return (usersChunk: PrivMonBulkUser[], source: MonitoringEntitySource) =>
     buildOps({
       users: usersChunk,
@@ -301,5 +300,5 @@ export const makeIndexOpsBuilder = (dataClient: PrivilegeMonitoringDataClient) =
       sourceLabel: 'index_sync',
       buildUpdateParams: (user) => ({ source_id: user.sourceId }),
     });
-  return indexOperations;
+  return indexOperations || [];
 };

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

@@ -5,62 +5,25 @@
  * 2.0.
  */
 import { defaultMonitoringUsersIndex } from '../../../../../common/entity_analytics/privileged_user_monitoring/utils';
-import type { IntegrationType } from './constants';
 import {
   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.entityanalytics_okta.values).toEqual(OKTA_ADMIN_ROLES);
-    // expect(INTEGRATION_MATCHERS_DETAILED.ad.values).toEqual(AD_ADMIN_ROLES);
-    expect(INTEGRATION_MATCHERS_DETAILED.entityanalytics_okta.fields).toEqual(['user.roles']);
-    // expect(INTEGRATION_MATCHERS_DETAILED.ad.fields).toEqual(['user.roles']);
-  });
 
   it('getMatchersFor returns correct matcher array', () => {
     expect(getMatchersFor('entityanalytics_okta')).toEqual([
       INTEGRATION_MATCHERS_DETAILED.entityanalytics_okta,
     ]);
-    // expect(getMatchersFor('ad')).toEqual([INTEGRATION_MATCHERS_DETAILED.ad]); add ad in follow-up PR
-  });
-
-  it('IntegrationType type should only allow okta and ad', () => {
-    const types: IntegrationType[] = ['entityanalytics_okta']; // add ad in follow-up PR
-    expect(types).toEqual(INTEGRATION_TYPES);
+    expect(getMatchersFor('entityanalytics_ad')).toEqual([
+      INTEGRATION_MATCHERS_DETAILED.entityanalytics_ad,
+    ]);
   });
 
   it('should generate defaultMonitoringUsersIndex', () => {
@@ -72,22 +35,26 @@ describe('constants', () => {
     expect(integrationsSourceIndex('default', 'entityanalytics_okta')).toBe(
       `${baseMonitoringUsersIndex}.sources.entityanalytics_okta-default`
     );
-    /* expect(integrationsSourceIndex('space1', 'ad')).toBe(
-      `${baseMonitoringUsersIndex}.sources.ad-space1`
-    ); */
+    expect(integrationsSourceIndex('space1', 'entityanalytics_ad')).toBe(
+      `${baseMonitoringUsersIndex}.sources.entityanalytics_ad-space1`
+    );
   });
 
   it('should generate correct stream index patterns', () => {
     expect(STREAM_INDEX_PATTERNS.entityanalytics_okta('default')).toBe(
       'logs-entityanalytics_okta.user-default'
     );
-    // expect(STREAM_INDEX_PATTERNS.ad('space1')).toBe('logs-entityanalytics_ad.user-space1');
+    expect(STREAM_INDEX_PATTERNS.entityanalytics_ad('space1')).toBe(
+      'logs-entityanalytics_ad.user-space1'
+    );
   });
 
   it('getStreamPatternFor returns correct pattern', () => {
     expect(getStreamPatternFor('entityanalytics_okta', 'default')).toBe(
       'logs-entityanalytics_okta.user-default'
     );
-    // expect(getStreamPatternFor('ad', 'space1')).toBe('logs-entityanalytics_ad.user-space1');
+    expect(getStreamPatternFor('entityanalytics_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

@@ -21,11 +21,11 @@ export const OKTA_ADMIN_ROLES: string[] = [
   'Read-only Administrator',
 ];
 
-export const AD_ADMIN_ROLES: string[] = ['Domain Admins', 'Enterprise Admins'];
+export const AD_ADMIN_GROUPS: string[] = ['Domain Admins', 'Enterprise Admins'];
 
 export const INTEGRATION_MATCHERS_DETAILED: Record<IntegrationType, Matcher> = {
   entityanalytics_okta: { fields: ['user.roles'], values: OKTA_ADMIN_ROLES },
-  // ad: { fields: ['user.roles'], values: AD_ADMIN_ROLES },
+  entityanalytics_ad: { fields: ['user.group.name'], values: AD_ADMIN_GROUPS },
 };
 
 export const getMatchersFor = (integration: IntegrationType): Matcher[] => [
@@ -41,12 +41,12 @@ export const integrationsSourceIndex = (namespace: string, integrationName: stri
 export const PRIVILEGE_MONITORING_PRIVILEGE_CHECK_API =
   '/api/entity_analytics/monitoring/privileges/privileges';
 
-export const INTEGRATION_TYPES = ['entityanalytics_okta' /* 'ad'*/] as const;
+export const INTEGRATION_TYPES = ['entityanalytics_okta', 'entityanalytics_ad'] as const;
 export type IntegrationType = (typeof INTEGRATION_TYPES)[number];
 
 export const STREAM_INDEX_PATTERNS: Record<IntegrationType, (namespace: string) => string> = {
   entityanalytics_okta: (namespace) => `logs-entityanalytics_okta.user-${namespace}`,
-  // ad: (namespace) => `logs-entityanalytics_ad.user-${namespace}`,
+  entityanalytics_ad: (namespace) => `logs-entityanalytics_ad.user-${namespace}`,
 };
 
 export const getStreamPatternFor = (integration: IntegrationType, namespace: string): string =>

x-pack/solutions/security/plugins/security_solution/server/lib/entity_analytics/privilege_monitoring/data_sources/sync/integrations/deletion_detection/deletion_detection.ts

@@ -45,6 +45,7 @@ export const createDeletionDetectionService = (
         'debug',
         `No new full sync for source ${source.id}; skipping deletion detection.`
       );
+      return;
     }
 
     const [completedEventTimeStamp, startedEventTimeStamp] = await Promise.all([
@@ -75,8 +76,13 @@ export const createDeletionDetectionService = (
       const staleUsers = await findStaleUsers(
         source.id,
         allIntegrationsUserNames,
-        'entity_analytics_integration' // TODO: confirm index/type constant
+        'entity_analytics_integration'
       );
+
+      if (staleUsers.length === 0) {
+        dataClient.log('debug', `No stale users to soft delete for source ${source.id}`);
+        return;
+      }
       const ops = bulkUtilsService.bulkSoftDeleteOperations(
         staleUsers,
         dataClient.index,
@@ -115,7 +121,6 @@ export const createDeletionDetectionService = (
     const usersToDelete: string[] = [];
     while (fetchMore) {
       const privilegedMonitoringUsers = await esClient.search<never, StaleUsersAggregations>({
-        // you need to change the name for this type
         index: source.indexPattern,
         ...buildFindUsersSearchBody({
           timeGte: startedEventTimeStamp,

x-pack/solutions/security/plugins/security_solution/server/lib/entity_analytics/privilege_monitoring/data_sources/sync/integrations/update_detection/privileged_status_match.ts

@@ -102,14 +102,19 @@ export const createPatternMatcherService = (
       while (fetchMore) {
         const response = await esClient.search<never, PrivMatchersAggregation>({
           index: source.indexPattern,
-          ...buildPrivilegedSearchBody(script, lastProcessedTimeStamp, afterKey, pageSize),
+          ...buildPrivilegedSearchBody(
+            script,
+            lastProcessedTimeStamp,
+            source.matchers[0].fields[0],
+            afterKey,
+            pageSize
+          ),
         });
 
         const aggregations = response.aggregations;
         const privUserAgg = response.aggregations?.privileged_user_status_since_last_run;
         const buckets = privUserAgg?.buckets ?? [];
 
-        // process current page
         if (buckets.length && aggregations) {
           const { users: privMonUsers, maxTimestamp } = await parseAggregationResponse(
             aggregations,

x-pack/solutions/security/plugins/security_solution/server/lib/entity_analytics/privilege_monitoring/data_sources/sync/integrations/update_detection/queries.ts

@@ -51,6 +51,7 @@ export const buildMatcherScript = (matcher?: Matcher): estypes.Script => {
 export const buildPrivilegedSearchBody = (
   script: estypes.Script,
   timeGte: string,
+  matchersField: string,
   afterKey?: AfterKey,
   pageSize: number = 100
 ): Omit<estypes.SearchRequest, 'index'> => ({
@@ -71,7 +72,7 @@ export const buildPrivilegedSearchBody = (
             size: 1,
             sort: [{ '@timestamp': { order: 'desc' as estypes.SortOrder } }],
             script_fields: { 'user.is_privileged': { script } },
-            _source: { includes: ['user', 'roles', '@timestamp'] },
+            _source: { includes: ['user', matchersField, '@timestamp'] },
           },
         },
       },
@@ -97,12 +98,14 @@ export const applyPrivilegedUpdates = async ({
     for (let start = 0; start < users.length; start += chunkSize) {
       const chunk = users.slice(start, start + chunkSize);
       const operations = opsForIntegration(chunk, source);
-      const resp = await esClient.bulk({
-        refresh: 'wait_for',
-        body: operations,
-      });
-      const errors = getErrorFromBulkResponse(resp);
-      dataClient.log('error', errorsMsg(errors));
+      if (operations.length > 0) {
+        const resp = await esClient.bulk({
+          refresh: 'wait_for',
+          body: operations,
+        });
+        const errors = getErrorFromBulkResponse(resp);
+        dataClient.log('error', errorsMsg(errors));
+      }
     }
   } catch (error) {
     dataClient.log('error', `Error applying privileged updates: ${error.message}`);

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

@@ -66,6 +66,7 @@ describe('createInitialisationSourcesService', () => {
     const existingSources = [
       { id: '1', name: '.entity_analytics.monitoring.users-default' },
       { id: '2', name: '.entity_analytics.monitoring.sources.entityanalytics_okta-default' },
+      { id: '2', name: '.entity_analytics.monitoring.sources.entityanalytics_ad-default' },
     ];
     mockFindAll.mockResolvedValue(existingSources);
     mockFindByQuery.mockResolvedValue(existingSources);

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

@@ -80,10 +80,9 @@ export const createInitialisationSourcesService = (deps: {
 };
 
 const getLastFullSyncMarkersIndex = (namespace: string, integration: IntegrationType) => {
-  // When using AD, will use the users index: TODO in: https://github.com/elastic/security-team/issues/13990
-  /* if (integration === 'ad') {
+  if (integration === 'entityanalytics_ad') {
     return getStreamPatternFor(integration, namespace);
-  }*/
+  }
   // okta has a dedicated index for last full sync markers
   return oktaLastFullSyncMarkersIndex(namespace);
 };
@@ -98,12 +97,12 @@ const makeIntegrationSource = (namespace: string, integration: IntegrationType)
   integrations: { syncMarkerIndex: getLastFullSyncMarkersIndex(namespace, integration) },
 });
 
-const buildRequiredSources = (namespace: string, indexPattern: string) => {
+function buildRequiredSources(namespace: string, indexPattern: string) {
   const integrations = INTEGRATION_TYPES.map((integration) =>
     makeIntegrationSource(namespace, integration)
   );
   return [makeDefaultIndexSource(namespace, indexPattern), ...integrations];
-};
+}
 
 const makeDefaultIndexSource = (namespace: string, name: string) => ({
   type: 'index' as const,

x-pack/solutions/security/test/security_solution_api_integration/test_suites/entity_analytics/monitoring/trial_license_complete_tier/privileged_users/migrations.ts

@@ -65,6 +65,19 @@ export default ({ getService }: FtrProviderContext) => {
       refresh: 'wait_for',
     });
 
+  const expectAllDefaultSourcesToExist = async (namespace: string) => {
+    const sources = await entityAnalyticsApi.listEntitySources({ query: {} }, namespace);
+    const names = (sources.body as ListEntitySourcesResponse).map((s) => s.name);
+
+    expect(names.sort()).toEqual(
+      [
+        `.entity_analytics.monitoring.users-${namespace}`,
+        `.entity_analytics.monitoring.sources.entityanalytics_okta-${namespace}`,
+        `.entity_analytics.monitoring.sources.entityanalytics_ad-${namespace}`,
+      ].sort()
+    );
+  };
+
   const SPACES = ['default', 'space1'];
 
   describe('@ess @serverless @skipInServerlessMKI Entity Analytics Privileged user monitoring Migrations', () => {
@@ -169,39 +182,21 @@ export default ({ getService }: FtrProviderContext) => {
 
           await entityAnalyticsRoutes.runMigrations();
 
-          const sources = await entityAnalyticsApi.listEntitySources({ query: {} }, namespace);
-          const names = (sources.body as ListEntitySourcesResponse).map((s) => s.name);
-
-          expect(names).toEqual([
-            `.entity_analytics.monitoring.users-${namespace}`,
-            `.entity_analytics.monitoring.sources.entityanalytics_okta-${namespace}`,
-          ]);
+          await expectAllDefaultSourcesToExist(namespace);
         });
 
         it(`should create missing entity source when migration runs one entity source doesn't exist`, async () => {
           await deleteEntitySources(namespace, [`.entity_analytics.monitoring.users-${namespace}`]);
 
           await entityAnalyticsRoutes.runMigrations();
 
-          const sources = await entityAnalyticsApi.listEntitySources({ query: {} }, namespace);
-          const names = (sources.body as ListEntitySourcesResponse).map((s) => s.name);
-
-          expect(names).toEqual([
-            `.entity_analytics.monitoring.users-${namespace}`,
-            `.entity_analytics.monitoring.sources.entityanalytics_okta-${namespace}`,
-          ]);
+          await expectAllDefaultSourcesToExist(namespace);
         });
 
         it(`should work as expected when migration runs and all entity sources exist`, async () => {
           await entityAnalyticsRoutes.runMigrations();
 
-          const sources = await entityAnalyticsApi.listEntitySources({ query: {} }, namespace);
-          const names = (sources.body as ListEntitySourcesResponse).map((s) => s.name);
-
-          expect(names).toEqual([
-            `.entity_analytics.monitoring.users-${namespace}`,
-            `.entity_analytics.monitoring.sources.entityanalytics_okta-${namespace}`,
-          ]);
+          await expectAllDefaultSourcesToExist(namespace);
         });
       });
     });