[Security Solution][Detections] Modify threshold rule synthetic signal generation to use data from last hit in bucket (#82444)

* Fix threshold rule synthetic signal generation

* Use top_hits aggregation

* Add timestampOverride

* Account for when threshold.field is not supplied

* Ensure we're getting the last event when threshold.field is not provided

* Add missing import

x-pack/plugins/security_solution/server/lib/detection_engine/signals/build_events_query.ts

@@ -4,7 +4,10 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
-import { TimestampOverrideOrUndefined } from '../../../../common/detection_engine/schemas/common/schemas';
+import {
+  SortOrderOrUndefined,
+  TimestampOverrideOrUndefined,
+} from '../../../../common/detection_engine/schemas/common/schemas';
 
 interface BuildEventsSearchQuery {
   aggregations?: unknown;
@@ -13,6 +16,7 @@ interface BuildEventsSearchQuery {
   to: string;
   filter: unknown;
   size: number;
+  sortOrder?: SortOrderOrUndefined;
   searchAfterSortId: string | number | undefined;
   timestampOverride: TimestampOverrideOrUndefined;
 }
@@ -25,6 +29,7 @@ export const buildEventsSearchQuery = ({
   filter,
   size,
   searchAfterSortId,
+  sortOrder,
   timestampOverride,
 }: BuildEventsSearchQuery) => {
   const timestamp = timestampOverride ?? '@timestamp';
@@ -108,7 +113,7 @@ export const buildEventsSearchQuery = ({
       sort: [
         {
           [timestamp]: {
-            order: 'asc',
+            order: sortOrder ?? 'asc',
           },
         },
       ],

x-pack/plugins/security_solution/server/lib/detection_engine/signals/bulk_create_threshold_signals.test.ts

@@ -4,10 +4,29 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
+import { sampleDocNoSortIdNoVersion } from './__mocks__/es_results';
 import { getThresholdSignalQueryFields } from './bulk_create_threshold_signals';
 
 describe('getThresholdSignalQueryFields', () => {
   it('should return proper fields for match_phrase filters', () => {
+    const mockHit = {
+      ...sampleDocNoSortIdNoVersion(),
+      _source: {
+        '@timestamp': '2020-11-03T02:31:47.431Z',
+        event: {
+          dataset: 'traefik.access',
+          module: 'traefik',
+        },
+        traefik: {
+          access: {
+            entryPointName: 'web-secure',
+          },
+        },
+        url: {
+          domain: 'kibana.siem.estc.dev',
+        },
+      },
+    };
     const mockFilters = {
       bool: {
         must: [],
@@ -71,15 +90,28 @@ describe('getThresholdSignalQueryFields', () => {
       },
     };
 
-    expect(getThresholdSignalQueryFields(mockFilters)).toEqual({
-      'event.module': 'traefik',
+    expect(getThresholdSignalQueryFields(mockHit, mockFilters)).toEqual({
       'event.dataset': 'traefik.access',
+      'event.module': 'traefik',
       'traefik.access.entryPointName': 'web-secure',
       'url.domain': 'kibana.siem.estc.dev',
     });
   });
 
   it('should return proper fields object for nested match filters', () => {
+    const mockHit = {
+      ...sampleDocNoSortIdNoVersion(),
+      _source: {
+        '@timestamp': '2020-11-03T02:31:47.431Z',
+        event: {
+          dataset: 'traefik.access',
+          module: 'traefik',
+        },
+        url: {
+          domain: 'kibana.siem.estc.dev',
+        },
+      },
+    };
     const filters = {
       bool: {
         must: [],
@@ -104,7 +136,7 @@ describe('getThresholdSignalQueryFields', () => {
                     should: [
                       {
                         match: {
-                          'event.dataset': 'traefik.access',
+                          'event.dataset': 'traefik.*',
                         },
                       },
                     ],
@@ -120,13 +152,23 @@ describe('getThresholdSignalQueryFields', () => {
       },
     };
 
-    expect(getThresholdSignalQueryFields(filters)).toEqual({
-      'event.module': 'traefik',
+    expect(getThresholdSignalQueryFields(mockHit, filters)).toEqual({
       'event.dataset': 'traefik.access',
+      'event.module': 'traefik',
     });
   });
 
   it('should return proper object for simple match filters', () => {
+    const mockHit = {
+      ...sampleDocNoSortIdNoVersion(),
+      _source: {
+        '@timestamp': '2020-11-03T02:31:47.431Z',
+        event: {
+          dataset: 'traefik.access',
+          module: 'traefik',
+        },
+      },
+    };
     const filters = {
       bool: {
         must: [],
@@ -154,13 +196,23 @@ describe('getThresholdSignalQueryFields', () => {
       },
     };
 
-    expect(getThresholdSignalQueryFields(filters)).toEqual({
-      'event.module': 'traefik',
+    expect(getThresholdSignalQueryFields(mockHit, filters)).toEqual({
       'event.dataset': 'traefik.access',
+      'event.module': 'traefik',
     });
   });
 
   it('should return proper object for simple match_phrase filters', () => {
+    const mockHit = {
+      ...sampleDocNoSortIdNoVersion(),
+      _source: {
+        '@timestamp': '2020-11-03T02:31:47.431Z',
+        event: {
+          dataset: 'traefik.access',
+          module: 'traefik',
+        },
+      },
+    };
     const filters = {
       bool: {
         must: [],
@@ -188,13 +240,22 @@ describe('getThresholdSignalQueryFields', () => {
       },
     };
 
-    expect(getThresholdSignalQueryFields(filters)).toEqual({
+    expect(getThresholdSignalQueryFields(mockHit, filters)).toEqual({
       'event.module': 'traefik',
       'event.dataset': 'traefik.access',
     });
   });
 
   it('should return proper object for exists filters', () => {
+    const mockHit = {
+      ...sampleDocNoSortIdNoVersion(),
+      _source: {
+        '@timestamp': '2020-11-03T02:31:47.431Z',
+        event: {
+          module: 'traefik',
+        },
+      },
+    };
     const filters = {
       bool: {
         should: [
@@ -226,6 +287,46 @@ describe('getThresholdSignalQueryFields', () => {
         minimum_should_match: 1,
       },
     };
-    expect(getThresholdSignalQueryFields(filters)).toEqual({});
+    expect(getThresholdSignalQueryFields(mockHit, filters)).toEqual({});
+  });
+
+  it('should NOT add invalid characters from CIDR such as the "/" proper object for simple match_phrase filters', () => {
+    const mockHit = {
+      ...sampleDocNoSortIdNoVersion(),
+      _source: {
+        '@timestamp': '2020-11-03T02:31:47.431Z',
+        destination: {
+          ip: '192.168.0.16',
+        },
+        event: {
+          module: 'traefik',
+        },
+      },
+    };
+    const filters = {
+      bool: {
+        must: [],
+        filter: [
+          {
+            bool: {
+              should: [
+                {
+                  match: {
+                    'destination.ip': '192.168.0.0/16',
+                  },
+                },
+              ],
+              minimum_should_match: 1,
+            },
+          },
+        ],
+        should: [],
+        must_not: [],
+      },
+    };
+
+    expect(getThresholdSignalQueryFields(mockHit, filters)).toEqual({
+      'destination.ip': '192.168.0.16',
+    });
   });
 });

x-pack/plugins/security_solution/server/lib/detection_engine/signals/bulk_create_threshold_signals.ts

@@ -8,13 +8,16 @@ import uuidv5 from 'uuid/v5';
 import { reduce, get, isEmpty } from 'lodash/fp';
 import set from 'set-value';
 
-import { Threshold } from '../../../../common/detection_engine/schemas/common/schemas';
+import {
+  Threshold,
+  TimestampOverrideOrUndefined,
+} from '../../../../common/detection_engine/schemas/common/schemas';
 import { Logger } from '../../../../../../../src/core/server';
 import { AlertServices } from '../../../../../alerts/server';
 import { RuleAlertAction } from '../../../../common/detection_engine/types';
 import { RuleTypeParams, RefreshTypes } from '../types';
 import { singleBulkCreate, SingleBulkCreateResponse } from './single_bulk_create';
-import { SignalSearchResponse } from './types';
+import { SignalSearchResponse, SignalSourceHit, ThresholdAggregationBucket } from './types';
 import { BuildRuleMessage } from './rule_messages';
 
 // used to generate constant Threshold Signals ID when run with the same params
@@ -30,6 +33,7 @@ interface BulkCreateThresholdSignalsParams {
   id: string;
   filter: unknown;
   signalsIndex: string;
+  timestampOverride: TimestampOverrideOrUndefined;
   name: string;
   createdAt: string;
   createdBy: string;
@@ -51,11 +55,25 @@ interface FilterObject {
   };
 }
 
-const getNestedQueryFilters = (filtersObj: FilterObject): Record<string, string> => {
+const injectFirstMatch = (
+  hit: SignalSourceHit,
+  match: object | Record<string, string>
+): Record<string, string> | undefined => {
+  if (match != null) {
+    for (const key of Object.keys(match)) {
+      return { [key]: get(key, hit._source) } as Record<string, string>;
+    }
+  }
+};
+
+const getNestedQueryFilters = (
+  hit: SignalSourceHit,
+  filtersObj: FilterObject
+): Record<string, string> => {
   if (Array.isArray(filtersObj.bool?.filter)) {
     return reduce(
       (acc, filterItem) => {
-        const nestedFilter = getNestedQueryFilters(filterItem);
+        const nestedFilter = getNestedQueryFilters(hit, filterItem);
 
         if (nestedFilter) {
           return { ...acc, ...nestedFilter };
@@ -70,27 +88,32 @@ const getNestedQueryFilters = (filtersObj: FilterObject): Record<string, string>
     return (
       (filtersObj.bool?.should &&
         filtersObj.bool?.should[0] &&
-        (filtersObj.bool.should[0].match || filtersObj.bool.should[0].match_phrase)) ??
+        (injectFirstMatch(hit, filtersObj.bool.should[0].match) ||
+          injectFirstMatch(hit, filtersObj.bool.should[0].match_phrase))) ??
       {}
     );
   }
 };
 
-export const getThresholdSignalQueryFields = (filter: unknown) => {
+export const getThresholdSignalQueryFields = (hit: SignalSourceHit, filter: unknown) => {
   const filters = get('bool.filter', filter);
 
   return reduce(
     (acc, item) => {
       if (item.match_phrase) {
-        return { ...acc, ...item.match_phrase };
+        return { ...acc, ...injectFirstMatch(hit, item.match_phrase) };
       }
 
       if (item.bool?.should && (item.bool.should[0].match || item.bool.should[0].match_phrase)) {
-        return { ...acc, ...(item.bool.should[0].match || item.bool.should[0].match_phrase) };
+        return {
+          ...acc,
+          ...(injectFirstMatch(hit, item.bool.should[0].match) ||
+            injectFirstMatch(hit, item.bool.should[0].match_phrase)),
+        };
       }
 
       if (item.bool?.filter) {
-        return { ...acc, ...getNestedQueryFilters(item) };
+        return { ...acc, ...getNestedQueryFilters(hit, item) };
       }
 
       return acc;
@@ -104,9 +127,11 @@ const getTransformedHits = (
   results: SignalSearchResponse,
   inputIndex: string,
   startedAt: Date,
+  logger: Logger,
   threshold: Threshold,
   ruleId: string,
-  signalQueryFields: Record<string, string>
+  filter: unknown,
+  timestampOverride: TimestampOverrideOrUndefined
 ) => {
   if (isEmpty(threshold.field)) {
     const totalResults =
@@ -116,10 +141,16 @@ const getTransformedHits = (
       return [];
     }
 
+    const hit = results.hits.hits[0];
+    if (hit == null) {
+      logger.warn(`No hits returned, but totalResults >= threshold.value (${threshold.value})`);
+      return [];
+    }
+
     const source = {
-      '@timestamp': new Date().toISOString(),
+      '@timestamp': get(timestampOverride ?? '@timestamp', hit._source),
       threshold_count: totalResults,
-      ...signalQueryFields,
+      ...getThresholdSignalQueryFields(hit, filter),
     };
 
     return [
@@ -135,24 +166,30 @@ const getTransformedHits = (
     return [];
   }
 
-  return results.aggregations.threshold.buckets.map(
-    // eslint-disable-next-line @typescript-eslint/naming-convention
-    ({ key, doc_count }: { key: string; doc_count: number }) => {
-      const source = {
-        '@timestamp': new Date().toISOString(),
-        threshold_count: doc_count,
-        ...signalQueryFields,
-      };
+  return results.aggregations.threshold.buckets
+    .map(
+      ({ key, doc_count: docCount, top_threshold_hits: topHits }: ThresholdAggregationBucket) => {
+        const hit = topHits.hits.hits[0];
+        if (hit == null) {
+          return null;
+        }
 
-      set(source, threshold.field, key);
+        const source = {
+          '@timestamp': get(timestampOverride ?? '@timestamp', hit._source),
+          threshold_count: docCount,
+          ...getThresholdSignalQueryFields(hit, filter),
+        };
 
-      return {
-        _index: inputIndex,
-        _id: uuidv5(`${ruleId}${startedAt}${threshold.field}${key}`, NAMESPACE_ID),
-        _source: source,
-      };
-    }
-  );
+        set(source, threshold.field, key);
+
+        return {
+          _index: inputIndex,
+          _id: uuidv5(`${ruleId}${startedAt}${threshold.field}${key}`, NAMESPACE_ID),
+          _source: source,
+        };
+      }
+    )
+    .filter((bucket: ThresholdAggregationBucket) => bucket != null);
 };
 
 export const transformThresholdResultsToEcs = (
@@ -160,17 +197,20 @@ export const transformThresholdResultsToEcs = (
   inputIndex: string,
   startedAt: Date,
   filter: unknown,
+  logger: Logger,
   threshold: Threshold,
-  ruleId: string
+  ruleId: string,
+  timestampOverride: TimestampOverrideOrUndefined
 ): SignalSearchResponse => {
-  const signalQueryFields = getThresholdSignalQueryFields(filter);
   const transformedHits = getTransformedHits(
     results,
     inputIndex,
     startedAt,
+    logger,
     threshold,
     ruleId,
-    signalQueryFields
+    filter,
+    timestampOverride
   );
   const thresholdResults = {
     ...results,
@@ -194,8 +234,10 @@ export const bulkCreateThresholdSignals = async (
     params.inputIndexPattern.join(','),
     params.startedAt,
     params.filter,
+    params.logger,
     params.ruleParams.threshold!,
-    params.ruleParams.ruleId
+    params.ruleParams.ruleId,
+    params.timestampOverride
   );
   const buildRuleMessage = params.buildRuleMessage;
 

x-pack/plugins/security_solution/server/lib/detection_engine/signals/find_threshold_signals.ts

@@ -52,6 +52,21 @@ export const findThresholdSignals = async ({
               field: threshold.field,
               min_doc_count: threshold.value,
             },
+            aggs: {
+              // Get the most recent hit per bucket
+              top_threshold_hits: {
+                top_hits: {
+                  sort: [
+                    {
+                      [timestampOverride ?? '@timestamp']: {
+                        order: 'desc',
+                      },
+                    },
+                  ],
+                  size: 1,
+                },
+              },
+            },
           },
         }
       : {};
@@ -66,7 +81,8 @@ export const findThresholdSignals = async ({
     services,
     logger,
     filter,
-    pageSize: 0,
+    pageSize: 1,
+    sortOrder: 'desc',
     buildRuleMessage,
   });
 };

x-pack/plugins/security_solution/server/lib/detection_engine/signals/signal_rule_alert_type.ts

@@ -328,6 +328,7 @@ export const signalRulesAlertType = ({
             id: alertId,
             inputIndexPattern: inputIndex,
             signalsIndex: outputIndex,
+            timestampOverride,
             startedAt,
             name,
             createdBy,

x-pack/plugins/security_solution/server/lib/detection_engine/signals/single_search_after.ts

@@ -11,7 +11,10 @@ import { SignalSearchResponse } from './types';
 import { BuildRuleMessage } from './rule_messages';
 import { buildEventsSearchQuery } from './build_events_query';
 import { createErrorsFromShard, makeFloatString } from './utils';
-import { TimestampOverrideOrUndefined } from '../../../../common/detection_engine/schemas/common/schemas';
+import {
+  SortOrderOrUndefined,
+  TimestampOverrideOrUndefined,
+} from '../../../../common/detection_engine/schemas/common/schemas';
 
 interface SingleSearchAfterParams {
   aggregations?: unknown;
@@ -22,6 +25,7 @@ interface SingleSearchAfterParams {
   services: AlertServices;
   logger: Logger;
   pageSize: number;
+  sortOrder?: SortOrderOrUndefined;
   filter: unknown;
   timestampOverride: TimestampOverrideOrUndefined;
   buildRuleMessage: BuildRuleMessage;
@@ -38,6 +42,7 @@ export const singleSearchAfter = async ({
   filter,
   logger,
   pageSize,
+  sortOrder,
   timestampOverride,
   buildRuleMessage,
 }: SingleSearchAfterParams): Promise<{
@@ -53,6 +58,7 @@ export const singleSearchAfter = async ({
       to,
       filter,
       size: pageSize,
+      sortOrder,
       searchAfterSortId,
       timestampOverride,
     });

x-pack/plugins/security_solution/server/lib/detection_engine/signals/threat_mapping/types.ts

@@ -28,6 +28,8 @@ import { TelemetryEventsSender } from '../../../telemetry/sender';
 import { BuildRuleMessage } from '../rule_messages';
 import { SearchAfterAndBulkCreateReturnType } from '../types';
 
+export type SortOrderOrUndefined = 'asc' | 'desc' | undefined;
+
 export interface CreateThreatSignalsOptions {
   threatMapping: ThreatMapping;
   query: string;
@@ -146,7 +148,7 @@ export interface GetThreatListOptions {
   perPage?: number;
   searchAfter: string[] | undefined;
   sortField: string | undefined;
-  sortOrder: 'asc' | 'desc' | undefined;
+  sortOrder: SortOrderOrUndefined;
   threatFilters: PartialFilter[];
   exceptionItems: ExceptionListItemSchema[];
   listClient: ListClient;
@@ -165,7 +167,7 @@ export interface ThreatListCountOptions {
 
 export interface GetSortWithTieBreakerOptions {
   sortField: string | undefined;
-  sortOrder: 'asc' | 'desc' | undefined;
+  sortOrder: SortOrderOrUndefined;
   index: string[];
   listItemIndex: string;
 }

x-pack/plugins/security_solution/server/lib/detection_engine/signals/types.ts

@@ -14,7 +14,7 @@ import {
   AlertExecutorOptions,
   AlertServices,
 } from '../../../../../alerts/server';
-import { SearchResponse } from '../../types';
+import { BaseSearchResponse, SearchResponse, TermAggregationBucket } from '../../types';
 import {
   EqlSearchResponse,
   BaseHit,
@@ -235,3 +235,7 @@ export interface SearchAfterAndBulkCreateReturnType {
   createdSignalsCount: number;
   errors: string[];
 }
+
+export interface ThresholdAggregationBucket extends TermAggregationBucket {
+  top_threshold_hits: BaseSearchResponse<SignalSource>;
+}

x-pack/plugins/security_solution/server/lib/types.ts

@@ -69,41 +69,48 @@ export type ShardError = Partial<{
   }>;
 }>;
 
-export interface SearchResponse<T> {
+export interface SearchHits<T> {
+  total: TotalValue | number;
+  max_score: number;
+  hits: Array<
+    BaseHit<T> & {
+      _type: string;
+      _score: number;
+      _version?: number;
+      _explanation?: Explanation;
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any
+      highlight?: any;
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any
+      inner_hits?: any;
+      matched_queries?: string[];
+      sort?: string[];
+    }
+  >;
+}
+
+export interface BaseSearchResponse<T> {
+  hits: SearchHits<T>;
+}
+
+export interface SearchResponse<T> extends BaseSearchResponse<T> {
   took: number;
   timed_out: boolean;
   _scroll_id?: string;
   _shards: ShardsResponse;
-  hits: {
-    total: TotalValue | number;
-    max_score: number;
-    hits: Array<
-      BaseHit<T> & {
-        _type: string;
-        _score: number;
-        _version?: number;
-        _explanation?: Explanation;
-        // eslint-disable-next-line @typescript-eslint/no-explicit-any
-        highlight?: any;
-        // eslint-disable-next-line @typescript-eslint/no-explicit-any
-        inner_hits?: any;
-        matched_queries?: string[];
-        sort?: string[];
-      }
-    >;
-  };
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
   aggregations?: any;
 }
 
 export type SearchHit = SearchResponse<object>['hits']['hits'][0];
 
+export interface TermAggregationBucket {
+  key: string;
+  doc_count: number;
+}
+
 export interface TermAggregation {
   [agg: string]: {
-    buckets: Array<{
-      key: string;
-      doc_count: number;
-    }>;
+    buckets: TermAggregationBucket[];
   };
 }