[Backport 2.x] #737 add field based rules support in correlation engine

src/main/java/org/opensearch/securityanalytics/SecurityAnalyticsPlugin.java

@@ -295,6 +295,7 @@ public List<Setting<?>> getSettings() {
                 SecurityAnalyticsSettings.CORRELATION_HISTORY_RETENTION_PERIOD,
                 SecurityAnalyticsSettings.IS_CORRELATION_INDEX_SETTING,
                 SecurityAnalyticsSettings.CORRELATION_TIME_WINDOW,
+                SecurityAnalyticsSettings.ENABLE_AUTO_CORRELATIONS,
                 SecurityAnalyticsSettings.DEFAULT_MAPPING_SCHEMA,
                 SecurityAnalyticsSettings.ENABLE_WORKFLOW_USAGE,
                 SecurityAnalyticsSettings.TIF_UPDATE_INTERVAL,

src/main/java/org/opensearch/securityanalytics/model/CorrelationQuery.java

@@ -22,40 +22,53 @@
     private static final String QUERY = "query";
     private static final String CATEGORY = "category";
 
+    private static final String FIELD = "field";
+
     private String index;
 
     private String query;
 
     private String category;
 
-    public CorrelationQuery(String index, String query, String category) {
+    private String field;
+
+    public CorrelationQuery(String index, String query, String category, String field) {
         this.index = index;
         this.query = query;
         this.category = category;
+        this.field = field;
     }
 
     public CorrelationQuery(StreamInput sin) throws IOException {
-        this(sin.readString(), sin.readString(), sin.readString());
+        this(sin.readString(), sin.readOptionalString(), sin.readString(), sin.readOptionalString());
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         out.writeString(index);
-        out.writeString(query);
+        out.writeOptionalString(query);
         out.writeString(category);
+        out.writeOptionalString(field);
     }
 
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         builder.startObject();
-        builder.field(INDEX, index).field(QUERY, query).field(CATEGORY, category);
+        builder.field(INDEX, index).field(CATEGORY, category);
+        if (query != null) {
+            builder.field(QUERY, query);
+        }
+        if (field != null) {
+            builder.field(FIELD, field);
+        }
         return builder.endObject();
     }
 
     public static CorrelationQuery parse(XContentParser xcp) throws IOException {
         String index = null;
         String query = null;
         String category = null;
+        String field = null;
 
         XContentParserUtils.ensureExpectedToken(XContentParser.Token.START_OBJECT, xcp.currentToken(), xcp);
         while (xcp.nextToken() != XContentParser.Token.END_OBJECT) {
@@ -72,11 +85,14 @@
                 case CATEGORY:
                     category = xcp.text();
                     break;
+                case FIELD:
+                    field = xcp.text();
+                    break;
                 default:
                     xcp.skipChildren();
             }
         }
-        return new CorrelationQuery(index, query, category);
+        return new CorrelationQuery(index, query, category, field);
     }
 
     public static CorrelationQuery readFrom(StreamInput sin) throws IOException {
@@ -94,4 +110,8 @@
     public String getCategory() {
         return category;
     }
+
+    public String getField() {
+        return field;
+    }
 }

src/main/java/org/opensearch/securityanalytics/model/CorrelationRule.java

@@ -28,6 +28,7 @@
     public static final String NO_ID = "";
     public static final Long NO_VERSION = 1L;
     private static final String CORRELATION_QUERIES = "correlate";
+    private static final String CORRELATION_TIME_WINDOW = "time_window";
 
     private String id;
 
@@ -37,15 +38,18 @@
 
     private List<CorrelationQuery> correlationQueries;
 
-    public CorrelationRule(String id, Long version, String name, List<CorrelationQuery> correlationQueries) {
+    private Long corrTimeWindow;
+
+    public CorrelationRule(String id, Long version, String name, List<CorrelationQuery> correlationQueries, Long corrTimeWindow) {
         this.id = id != null ? id : NO_ID;
         this.version = version != null ? version : NO_VERSION;
         this.name = name;
         this.correlationQueries = correlationQueries;
+        this.corrTimeWindow = corrTimeWindow != null? corrTimeWindow: 300000L;
     }
 
     public CorrelationRule(StreamInput sin) throws IOException {
-        this(sin.readString(), sin.readLong(), sin.readString(), sin.readList(CorrelationQuery::readFrom));
+        this(sin.readString(), sin.readLong(), sin.readString(), sin.readList(CorrelationQuery::readFrom), sin.readLong());
     }
 
     @Override
@@ -57,6 +61,7 @@
         CorrelationQuery[] correlationQueries = new CorrelationQuery[] {};
         correlationQueries = this.correlationQueries.toArray(correlationQueries);
         builder.field(CORRELATION_QUERIES, correlationQueries);
+        builder.field(CORRELATION_TIME_WINDOW, corrTimeWindow);
         return builder.endObject();
     }
 
@@ -69,6 +74,7 @@
         for (CorrelationQuery query : correlationQueries) {
             query.writeTo(out);
         }
+        out.writeLong(corrTimeWindow);
     }
 
     public static CorrelationRule parse(XContentParser xcp, String id, Long version) throws IOException {
@@ -81,6 +87,7 @@
 
         String name = null;
         List<CorrelationQuery> correlationQueries = new ArrayList<>();
+        Long corrTimeWindow = null;
 
         XContentParserUtils.ensureExpectedToken(XContentParser.Token.START_OBJECT, xcp.nextToken(), xcp);
         while (xcp.nextToken() != XContentParser.Token.END_OBJECT) {
@@ -98,11 +105,14 @@
                         correlationQueries.add(query);
                     }
                     break;
+                case CORRELATION_TIME_WINDOW:
+                    corrTimeWindow = xcp.longValue();
+                    break;
                 default:
                     xcp.skipChildren();
             }
         }
-        return new CorrelationRule(id, version, name, correlationQueries);
+        return new CorrelationRule(id, version, name, correlationQueries, corrTimeWindow);
     }
 
     public static CorrelationRule readFrom(StreamInput sin) throws IOException {
@@ -137,6 +147,10 @@
         return correlationQueries;
     }
 
+    public Long getCorrTimeWindow() {
+        return corrTimeWindow;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;

src/main/java/org/opensearch/securityanalytics/settings/SecurityAnalyticsSettings.java

@@ -138,6 +138,15 @@ public class SecurityAnalyticsSettings {
             Setting.Property.NodeScope, Setting.Property.Dynamic
     );
 
+    /**
+     * Setting which enables auto correlations
+     */
+    public static final Setting<Boolean> ENABLE_AUTO_CORRELATIONS = Setting.boolSetting(
+            "plugins.security_analytics.auto_correlations_enabled",
+            false,
+            Setting.Property.NodeScope, Setting.Property.Dynamic
+    );
+
     public static final Setting<String> DEFAULT_MAPPING_SCHEMA = Setting.simpleString(
             "plugins.security_analytics.mappings.default_schema",
             "ecs",

src/main/java/org/opensearch/securityanalytics/transport/TransportCorrelateFindingAction.java

@@ -96,6 +96,8 @@
 
     private volatile long setupTimestamp;
 
+    private volatile boolean enableAutoCorrelation;
+
     @Inject
     public TransportCorrelateFindingAction(TransportService transportService,
                                            Client client,
@@ -118,8 +120,10 @@
 
         this.indexTimeout = SecurityAnalyticsSettings.INDEX_TIMEOUT.get(this.settings);
         this.corrTimeWindow = SecurityAnalyticsSettings.CORRELATION_TIME_WINDOW.get(this.settings).getMillis();
+        this.enableAutoCorrelation = SecurityAnalyticsSettings.ENABLE_AUTO_CORRELATIONS.get(this.settings);
         this.clusterService.getClusterSettings().addSettingsUpdateConsumer(SecurityAnalyticsSettings.INDEX_TIMEOUT, it -> indexTimeout = it);
         this.clusterService.getClusterSettings().addSettingsUpdateConsumer(SecurityAnalyticsSettings.CORRELATION_TIME_WINDOW, it -> corrTimeWindow = it.getMillis());
+        this.clusterService.getClusterSettings().addSettingsUpdateConsumer(SecurityAnalyticsSettings.ENABLE_AUTO_CORRELATIONS, it -> enableAutoCorrelation = it);
         this.setupTimestamp = System.currentTimeMillis();
     }
 
@@ -220,7 +224,7 @@
 
             this.response =new AtomicReference<>();
 
-            this.joinEngine = new JoinEngine(client, request, xContentRegistry, corrTimeWindow, this, logTypeService);
+            this.joinEngine = new JoinEngine(client, request, xContentRegistry, corrTimeWindow, this, logTypeService, enableAutoCorrelation);
             this.vectorEmbeddingsEngine = new VectorEmbeddingsEngine(client, indexTimeout, corrTimeWindow, this);
         }
 

src/main/resources/mappings/finding_mapping.json

@@ -1,7 +1,7 @@
 {
   "dynamic": "strict",
   "_meta" : {
-    "schema_version": 3
+    "schema_version": 4
   },
   "properties": {
     "schema_version": {
@@ -46,6 +46,9 @@
               "type" : "keyword"
             }
           }
+        },
+        "fields": {
+          "type": "text"
         }
       }
     },