create doc level query from threat intel feed data index docs"

Signed-off-by: Surya Sashank Nistala <[email protected]>

build.gradle

@@ -155,7 +155,7 @@ dependencies {
     implementation group: 'org.apache.commons', name: 'commons-lang3', version: "${versions.commonslang}"
     implementation "org.antlr:antlr4-runtime:4.10.1"
     implementation "com.cronutils:cron-utils:9.1.6"
-    api "org.opensearch:common-utils:${common_utils_version}@jar"
+    api files("/Users/snistala/Documents/opensearch/common-utils/build/libs/common-utils-3.0.0.0-SNAPSHOT.jar")
     api "org.opensearch.client:opensearch-rest-client:${opensearch_version}"
     implementation "org.jetbrains.kotlin:kotlin-stdlib:${kotlin_version}"
     compileOnly "org.opensearch:opensearch-job-scheduler-spi:${opensearch_build}"

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

@@ -137,7 +137,7 @@ public Collection<Object> createComponents(Client client,
         mapperService = new MapperService(client, clusterService, indexNameExpressionResolver, indexTemplateManager, logTypeService);
         ruleIndices = new RuleIndices(logTypeService, client, clusterService, threadPool);
         correlationRuleIndices = new CorrelationRuleIndices(client, clusterService);
-        ThreatIntelFeedDataService threatIntelFeedDataService = new ThreatIntelFeedDataService(clusterService.state(), clusterService, client, indexNameExpressionResolver, xContentRegistry);
+        ThreatIntelFeedDataService threatIntelFeedDataService = new ThreatIntelFeedDataService(clusterService, client, indexNameExpressionResolver, xContentRegistry);
         DetectorThreatIntelService detectorThreatIntelService = new DetectorThreatIntelService(threatIntelFeedDataService);
         this.client = client;
 

src/main/java/org/opensearch/securityanalytics/findings/FindingsService.java

@@ -216,7 +216,7 @@ public FindingDto mapFindingWithDocsToFindingDto(FindingWithDocs findingWithDocs
         if (docLevelQueries.isEmpty()) { // this is finding generated by a bucket level monitor
             for (Map.Entry<String, String> entry : detector.getRuleIdMonitorIdMap().entrySet()) {
                 if(entry.getValue().equals(findingWithDocs.getFinding().getMonitorId())) {
-                    docLevelQueries = Collections.singletonList(new DocLevelQuery(entry.getKey(),"","",Collections.emptyList()));
+                    docLevelQueries = Collections.singletonList(new DocLevelQuery(entry.getKey(), "", Collections.emptyList(), "", Collections.emptyList()));
                 }
             }
         }

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

@@ -56,7 +56,7 @@ public static ThreatIntelFeedData parse(XContentParser xcp, String id, Long vers
         String iocValue = null;
         String feedId = null;
         Instant timestamp = null;
-
+        xcp.nextToken();
         XContentParserUtils.ensureExpectedToken(XContentParser.Token.START_OBJECT, xcp.currentToken(), xcp);
         while (xcp.nextToken() != XContentParser.Token.END_OBJECT) {
             String fieldName = xcp.currentName();
@@ -126,6 +126,7 @@ public ThreatIntelFeedData(StreamInput sin) throws IOException {
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         return createXContentBuilder(builder, params);
+
     }
 
     private XContentBuilder createXContentBuilder(XContentBuilder builder, ToXContent.Params params) throws IOException {

src/main/java/org/opensearch/securityanalytics/threatIntel/DetectorThreatIntelService.java

@@ -1,14 +1,19 @@
 package org.opensearch.securityanalytics.threatIntel;
 
 import org.opensearch.commons.alerting.model.DocLevelQuery;
+import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.rest.RestStatus;
+import org.opensearch.securityanalytics.SecurityAnalyticsPlugin;
 import org.opensearch.securityanalytics.model.Detector;
 import org.opensearch.securityanalytics.model.ThreatIntelFeedData;
 import org.opensearch.securityanalytics.util.SecurityAnalyticsException;
 
 import java.util.Collections;
 import java.util.List;
 import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 
@@ -20,42 +25,75 @@ public DetectorThreatIntelService(ThreatIntelFeedDataService threatIntelFeedData
         this.threatIntelFeedDataService = threatIntelFeedDataService;
     }
 
-    /** Convert the feed data IOCs into query string query format to create doc level queries. */
+    /**
+     * Convert the feed data IOCs into query string query format to create doc level queries.
+     */
     public DocLevelQuery createDocLevelQueryFromThreatIntelList(
             List<ThreatIntelFeedData> tifdList, String docLevelQueryId
-            ) {
+    ) {
         Set<String> iocs = tifdList.stream().map(ThreatIntelFeedData::getIocValue).collect(Collectors.toSet());
         String query = buildQueryStringQueryWithIocList(iocs);
         return new DocLevelQuery(
-                docLevelQueryId,tifdList.get(0).getFeedId(), query,
+                docLevelQueryId, tifdList.get(0).getFeedId(),
+                Collections.singletonList("*"),
+                query,
                 Collections.singletonList("threat_intel")
         );
     }
 
     private String buildQueryStringQueryWithIocList(Set<String> iocs) {
         StringBuilder sb = new StringBuilder();
-
-        for(String ioc : iocs) {
-            if(sb.length() != 0) {
-                sb.append(" ");
+        sb.append("(");
+        for (String ioc : iocs) {
+            if (sb.length() > 2) {
+                sb.append(" OR ");
             }
-            sb.append("(");
             sb.append(ioc);
-            sb.append(")");
+
         }
+        sb.append(")");
         return sb.toString();
     }
 
-    public DocLevelQuery createDocLevelQueryFromThreatIntel(Detector detector) {
-        // for testing validation only.
-        if(detector.getThreatIntelEnabled() ==false) {
-            throw new SecurityAnalyticsException(
-                    "trying to create threat intel feed queries when flag to use threat intel is disabled.",
-                    RestStatus.FORBIDDEN, new IllegalArgumentException());
+    public void createDocLevelQueryFromThreatIntel(Detector detector, ActionListener<DocLevelQuery> listener) {
+        try {
+            if (detector.getThreatIntelEnabled() == false) {
+                listener.onResponse(null);
+                return;
+
+            }
+            CountDownLatch latch = new CountDownLatch(1);
+            // TODO: plugin logic to run job for populating threat intel feed data
+            //TODO populateFeedData()
+            threatIntelFeedDataService.getThreatIntelFeedData(new ActionListener<>() {
+                @Override
+                public void onResponse(List<ThreatIntelFeedData> threatIntelFeedData) {
+                    if (threatIntelFeedData.isEmpty()) {
+                        listener.onResponse(null);
+                    } else {
+                        listener.onResponse(createDocLevelQueryFromThreatIntelList(
+                                threatIntelFeedData,
+                                detector.getName() + "_threat_intel" + UUID.randomUUID()
+                        ));
+                    }
+                    latch.countDown();
+                }
+
+                @Override
+                public void onFailure(Exception e) {
+                    listener.onFailure(e);
+                    latch.countDown();
+                }
+            });
 
+            latch.await(30, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+            listener.onFailure(e);
         }
-        // TODO: plugin logic to run job for populating threat intel feed data
-        /*threatIntelFeedDataService.getThreatIntelFeedData("ip_address", );*/
-        return null;
+
+    }
+
+    public void updateDetectorsWithLatestThreatIntelRules() {
+
     }
 }

src/main/java/org/opensearch/securityanalytics/threatIntel/ThreatIntelFeedDataService.java

@@ -75,7 +75,6 @@ public class ThreatIntelFeedDataService {
     private static final String TYPE = "type";
     private static final String DATA_FIELD_NAME = "_data";
 
-    private final ClusterState state;
     private final Client client;
     private final IndexNameExpressionResolver indexNameExpressionResolver;
 
@@ -96,35 +95,29 @@ public class ThreatIntelFeedDataService {
             true
     );
     private final ClusterService clusterService;
-    private final ClusterSettings clusterSettings;
 
     public ThreatIntelFeedDataService(
-            ClusterState state,
             ClusterService clusterService,
             Client client,
             IndexNameExpressionResolver indexNameExpressionResolver,
             NamedXContentRegistry xContentRegistry) {
-        this.state = state;
         this.client = client;
         this.indexNameExpressionResolver = indexNameExpressionResolver;
         this.xContentRegistry = xContentRegistry;
         this.clusterService = clusterService;
-        this.clusterSettings = clusterService.getClusterSettings();
     }
 
     private final NamedXContentRegistry xContentRegistry;
 
     public void getThreatIntelFeedData(
-            String iocType,
             ActionListener<List<ThreatIntelFeedData>> listener
     ) {
         String tifdIndex = IndexUtils.getNewIndexByCreationDate(
-                this.state,
+                this.clusterService.state(),
                 this.indexNameExpressionResolver,
                 ".opensearch-sap-threatintel*" //name?
         );
         SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
-        sourceBuilder.query(QueryBuilders.boolQuery().filter(QueryBuilders.termQuery("ioc_type", iocType)));
         SearchRequest searchRequest = new SearchRequest(tifdIndex);
         searchRequest.source().size(9999); //TODO: convert to scroll
         searchRequest.source(sourceBuilder);
@@ -174,12 +167,13 @@ public void createIndexIfNotExists(final String indexName) {
                 .mapping(getIndexMapping());
         StashedThreadContext.run(
                 client,
-                () -> client.admin().indices().create(createIndexRequest).actionGet(clusterSettings.get(ThreatIntelSettings.THREAT_INTEL_TIMEOUT))
+                () -> client.admin().indices().create(createIndexRequest).actionGet(this.clusterService.getClusterSettings().get(ThreatIntelSettings.THREAT_INTEL_TIMEOUT))
         );
     }
 
     private void freezeIndex(final String indexName) {
-        TimeValue timeout = clusterSettings.get(ThreatIntelSettings.THREAT_INTEL_TIMEOUT);
+        ClusterSettings clusterSettings = this.clusterService.getClusterSettings();
+        TimeValue timeout = this.clusterService.getClusterSettings().get(ThreatIntelSettings.THREAT_INTEL_TIMEOUT);
         StashedThreadContext.run(client, () -> {
             client.admin().indices().prepareForceMerge(indexName).setMaxNumSegments(1).execute().actionGet(timeout);
             client.admin().indices().prepareRefresh(indexName).execute().actionGet(timeout);
@@ -260,7 +254,7 @@ public void saveThreatIntelFeedData(
         if (indexName == null || fields == null || iterator == null || renewLock == null){
             throw new IllegalArgumentException("Fields cannot be null");
         }
-
+        ClusterSettings clusterSettings = this.clusterService.getClusterSettings();
         TimeValue timeout = clusterSettings.get(ThreatIntelSettings.THREAT_INTEL_TIMEOUT);
         Integer batchSize = clusterSettings.get(ThreatIntelSettings.BATCH_SIZE);
         final BulkRequest bulkRequest = new BulkRequest();
@@ -297,6 +291,7 @@ public void deleteThreatIntelDataIndex(final String index) {
     }
 
     public void deleteThreatIntelDataIndex(final List<String> indices) {
+        ClusterSettings clusterSettings = this.clusterService.getClusterSettings();
         if (indices == null || indices.isEmpty()) {
             return;
         }

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

@@ -116,6 +116,7 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
@@ -646,13 +647,28 @@ private IndexMonitorRequest createDocLevelMonitorRequest(List<Pair<String, Rule>
             tags.add(rule.getCategory());
             tags.addAll(rule.getTags().stream().map(Value::getValue).collect(Collectors.toList()));
 
-            DocLevelQuery docLevelQuery = new DocLevelQuery(id, name, actualQuery, tags);
+            DocLevelQuery docLevelQuery = new DocLevelQuery(id, name, Collections.emptyList(), actualQuery, tags);
             docLevelQueries.add(docLevelQuery);
         }
         try {
             if (detector.getThreatIntelEnabled()) {
-                DocLevelQuery docLevelQueryFromThreatIntel = detectorThreatIntelService.createDocLevelQueryFromThreatIntel(detector);
-                docLevelQueries.add(docLevelQueryFromThreatIntel);
+                CountDownLatch countDownLatch = new CountDownLatch(1);
+                detectorThreatIntelService.createDocLevelQueryFromThreatIntel(detector, new ActionListener<>() {
+                    @Override
+                    public void onResponse(DocLevelQuery dlq) {
+                        if (dlq != null)
+                            docLevelQueries.add(dlq);
+                        countDownLatch.countDown();
+                    }
+
+                    @Override
+                    public void onFailure(Exception e) {
+                        // not failing detector creation if any fatal exception occurs during doc level query creation from threat intel feed data
+                        log.error("Failed to convert threat intel feed to. Proceeding with detector creation", e);
+                        countDownLatch.countDown();
+                    }
+                });
+                countDownLatch.await();
             }
         } catch (Exception e) {
             // not failing detector creation if any fatal exception occurs during doc level query creation from threat intel feed data
@@ -705,6 +721,7 @@ private IndexMonitorRequest createDocLevelMonitorMatchAllRequest(
         DocLevelQuery docLevelQuery = new DocLevelQuery(
                 monitorName,
                 monitorName + "doc",
+                Collections.emptyList(),
                 actualQuery,
                 Collections.emptyList()
         );

src/test/java/org/opensearch/securityanalytics/SecurityAnalyticsRestTestCase.java

@@ -64,6 +64,7 @@
 import org.opensearch.securityanalytics.model.CustomLogType;
 import org.opensearch.securityanalytics.model.Detector;
 import org.opensearch.securityanalytics.model.Rule;
+import org.opensearch.securityanalytics.model.ThreatIntelFeedData;
 import org.opensearch.test.rest.OpenSearchRestTestCase;
 
 
@@ -682,6 +683,11 @@ protected String toJsonString(CorrelationRule rule) throws IOException {
         return IndexUtilsKt.string(shuffleXContent(rule.toXContent(builder, ToXContent.EMPTY_PARAMS)));
     }
 
+    protected String toJsonString(ThreatIntelFeedData tifd) throws IOException {
+        XContentBuilder builder = XContentFactory.jsonBuilder();
+        return IndexUtilsKt.string(shuffleXContent(tifd.toXContent(builder, ToXContent.EMPTY_PARAMS)));
+    }
+
     private String alertingScheduledJobMappings() {
         return "  \"_meta\" : {\n" +
                 "    \"schema_version\": 5\n" +