add field based rules support in correlation engine (#737) (#812)

Signed-off-by: Joanne Wang <[email protected]>
opensearch-project · Feb 6, 2024 · 1cf3743 · 1cf3743
1 parent 7f68458
commit 1cf3743
Show file tree

Hide file tree

Showing 11 changed files with 1,196 additions and 59 deletions.
diff --git a/src/main/java/org/opensearch/securityanalytics/SecurityAnalyticsPlugin.java b/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,

diff --git a/src/main/java/org/opensearch/securityanalytics/correlation/JoinEngine.java b/src/main/java/org/opensearch/securityanalytics/correlation/JoinEngine.java
diff --git a/src/main/java/org/opensearch/securityanalytics/model/CorrelationQuery.java b/src/main/java/org/opensearch/securityanalytics/model/CorrelationQuery.java
@@ -22,40 +22,53 @@ public class CorrelationQuery implements Writeable, ToXContentObject {
     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 @@ public static CorrelationQuery parse(XContentParser xcp) throws IOException {
                 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 getQuery() {
     public String getCategory() {
         return category;
     }
+
+    public String getField() {
+        return field;
+    }
 }
diff --git a/src/main/java/org/opensearch/securityanalytics/model/CorrelationRule.java b/src/main/java/org/opensearch/securityanalytics/model/CorrelationRule.java
@@ -28,6 +28,7 @@ public class CorrelationRule implements Writeable, ToXContentObject {
     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 @@ public class CorrelationRule implements Writeable, ToXContentObject {
 
     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 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         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 @@ public void writeTo(StreamOutput out) throws IOException {
         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 @@ public static CorrelationRule parse(XContentParser xcp, String id, Long version)
 
         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 @@ public static CorrelationRule parse(XContentParser xcp, String id, Long version)
                         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 @@ public List<CorrelationQuery> getCorrelationQueries() {
         return correlationQueries;
     }
 
+    public Long getCorrTimeWindow() {
+        return corrTimeWindow;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;

diff --git a/src/main/java/org/opensearch/securityanalytics/settings/SecurityAnalyticsSettings.java b/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",

diff --git a/...main/java/org/opensearch/securityanalytics/transport/TransportCorrelateFindingAction.java b/...main/java/org/opensearch/securityanalytics/transport/TransportCorrelateFindingAction.java
@@ -96,6 +96,8 @@ public class TransportCorrelateFindingAction extends HandledTransportAction<Acti
 
     private volatile long setupTimestamp;
 
+    private volatile boolean enableAutoCorrelation;
+
     @Inject
     public TransportCorrelateFindingAction(TransportService transportService,
                                            Client client,
@@ -118,8 +120,10 @@ public TransportCorrelateFindingAction(TransportService transportService,
 
         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 @@ public class AsyncCorrelateFindingAction {
 
             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);
         }
 

diff --git a/src/main/resources/mappings/finding_mapping.json b/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"
         }
       }
     },