Threat Intel Feed Job Scheduler with unit and integ test

build.gradle

@@ -69,6 +69,7 @@ opensearchplugin {
     name 'opensearch-security-analytics'
     description 'OpenSearch Security Analytics plugin'
     classname 'org.opensearch.securityanalytics.SecurityAnalyticsPlugin'
+//    extendedPlugins = ['opensearch-job-scheduler']
 }
 
 javaRestTest {
@@ -158,11 +159,14 @@ dependencies {
     api "org.opensearch:common-utils:${common_utils_version}@jar"
     api "org.opensearch.client:opensearch-rest-client:${opensearch_version}"
     implementation "org.jetbrains.kotlin:kotlin-stdlib:${kotlin_version}"
+    implementation "org.opensearch:opensearch-job-scheduler-spi:${opensearch_build}"
+    implementation "org.apache.commons:commons-csv:1.10.0"
 
     // Needed for integ tests
     zipArchive group: 'org.opensearch.plugin', name:'alerting', version: "${opensearch_build}"
     zipArchive group: 'org.opensearch.plugin', name:'opensearch-notifications-core', version: "${opensearch_build}"
     zipArchive group: 'org.opensearch.plugin', name:'notifications', version: "${opensearch_build}"
+    zipArchive group: 'org.opensearch.plugin', name:'opensearch-job-scheduler', version: "${opensearch_build}"
 
     //spotless
     implementation('com.google.googlejavaformat:google-java-format:1.17.0') {
@@ -289,6 +293,16 @@ testClusters.integTest {
             }
         }
     }))
+    plugin(provider({
+        new RegularFile() {
+            @Override
+            File getAsFile() {
+                return configurations.zipArchive.asFileTree.matching {
+                    include '**/opensearch-job-scheduler*'
+                }.singleFile
+            }
+        }
+    }))
 }
 
 run {

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

@@ -4,20 +4,13 @@
  */
 package org.opensearch.securityanalytics;
 
-import java.util.Collection;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
+import java.util.*;
 import java.util.function.Supplier;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.opensearch.cluster.routing.Preference;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.action.ActionRequest;
 import org.opensearch.core.action.ActionResponse;
-import org.opensearch.action.search.SearchRequest;
-import org.opensearch.action.search.SearchResponse;
 import org.opensearch.client.Client;
 import org.opensearch.cluster.metadata.IndexNameExpressionResolver;
 import org.opensearch.cluster.node.DiscoveryNode;
@@ -38,18 +31,12 @@
 import org.opensearch.index.codec.CodecServiceFactory;
 import org.opensearch.index.engine.EngineFactory;
 import org.opensearch.index.mapper.Mapper;
-import org.opensearch.index.query.QueryBuilders;
-import org.opensearch.plugins.ActionPlugin;
-import org.opensearch.plugins.ClusterPlugin;
-import org.opensearch.plugins.EnginePlugin;
-import org.opensearch.plugins.MapperPlugin;
-import org.opensearch.plugins.Plugin;
-import org.opensearch.plugins.SearchPlugin;
+import org.opensearch.indices.SystemIndexDescriptor;
+import org.opensearch.plugins.*;
 import org.opensearch.repositories.RepositoriesService;
 import org.opensearch.rest.RestController;
 import org.opensearch.rest.RestHandler;
 import org.opensearch.script.ScriptService;
-import org.opensearch.search.builder.SearchSourceBuilder;
 import org.opensearch.securityanalytics.action.*;
 import org.opensearch.securityanalytics.correlation.index.codec.CorrelationCodecService;
 import org.opensearch.securityanalytics.correlation.index.mapper.CorrelationVectorFieldMapper;
@@ -60,7 +47,16 @@
 import org.opensearch.securityanalytics.mapper.IndexTemplateManager;
 import org.opensearch.securityanalytics.mapper.MapperService;
 import org.opensearch.securityanalytics.model.CustomLogType;
+import org.opensearch.securityanalytics.model.ThreatIntelFeedData;
 import org.opensearch.securityanalytics.resthandler.*;
+import org.opensearch.securityanalytics.threatIntel.DetectorThreatIntelService;
+import org.opensearch.securityanalytics.threatIntel.ThreatIntelFeedDataService;
+import org.opensearch.securityanalytics.threatIntel.action.*;
+import org.opensearch.securityanalytics.threatIntel.common.TIFLockService;
+import org.opensearch.securityanalytics.threatIntel.feedMetadata.BuiltInTIFMetadataLoader;
+import org.opensearch.securityanalytics.threatIntel.jobscheduler.TIFJobParameterService;
+import org.opensearch.securityanalytics.threatIntel.jobscheduler.TIFJobRunner;
+import org.opensearch.securityanalytics.threatIntel.jobscheduler.TIFJobUpdateService;
 import org.opensearch.securityanalytics.transport.*;
 import org.opensearch.securityanalytics.model.Rule;
 import org.opensearch.securityanalytics.model.Detector;
@@ -72,10 +68,13 @@
 import org.opensearch.securityanalytics.util.DetectorIndices;
 import org.opensearch.securityanalytics.util.RuleIndices;
 import org.opensearch.securityanalytics.util.RuleTopicIndices;
+import org.opensearch.threadpool.ExecutorBuilder;
 import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.watcher.ResourceWatcherService;
 
-public class SecurityAnalyticsPlugin extends Plugin implements ActionPlugin, MapperPlugin, SearchPlugin, EnginePlugin, ClusterPlugin {
+import static org.opensearch.securityanalytics.threatIntel.jobscheduler.TIFJobParameter.THREAT_INTEL_DATA_INDEX_NAME_PREFIX;
+
+public class SecurityAnalyticsPlugin extends Plugin implements ActionPlugin, MapperPlugin, SearchPlugin, EnginePlugin, ClusterPlugin, SystemIndexPlugin {
 
     private static final Logger log = LogManager.getLogger(SecurityAnalyticsPlugin.class);
 
@@ -116,6 +115,11 @@ public class SecurityAnalyticsPlugin extends Plugin implements ActionPlugin, Map
 
     private Client client;
 
+    @Override
+    public Collection<SystemIndexDescriptor> getSystemIndexDescriptors(Settings settings){
+        return List.of(new SystemIndexDescriptor(THREAT_INTEL_DATA_INDEX_NAME_PREFIX, "System index used for threat intel data"));
+    }
+
     @Override
     public Collection<Object> createComponents(Client client,
                                                ClusterService clusterService,
@@ -128,7 +132,9 @@ public Collection<Object> createComponents(Client client,
                                                NamedWriteableRegistry namedWriteableRegistry,
                                                IndexNameExpressionResolver indexNameExpressionResolver,
                                                Supplier<RepositoriesService> repositoriesServiceSupplier) {
+
         builtinLogTypeLoader = new BuiltinLogTypeLoader();
+        BuiltInTIFMetadataLoader builtInTIFMetadataLoader = new BuiltInTIFMetadataLoader();
         logTypeService = new LogTypeService(client, clusterService, xContentRegistry, builtinLogTypeLoader);
         detectorIndices = new DetectorIndices(client.admin(), clusterService, threadPool);
         ruleTopicIndices = new RuleTopicIndices(client, clusterService, logTypeService);
@@ -138,12 +144,20 @@ 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, client, indexNameExpressionResolver, xContentRegistry);
+        DetectorThreatIntelService detectorThreatIntelService = new DetectorThreatIntelService(threatIntelFeedDataService);
+        TIFJobParameterService tifJobParameterService = new TIFJobParameterService(client, clusterService);
+        TIFJobUpdateService tifJobUpdateService = new TIFJobUpdateService(clusterService, tifJobParameterService, threatIntelFeedDataService, builtInTIFMetadataLoader);
+        TIFLockService threatIntelLockService = new TIFLockService(clusterService, client);
+
         this.client = client;
 
+        TIFJobRunner.getJobRunnerInstance().initialize(clusterService,tifJobUpdateService, tifJobParameterService, threatIntelLockService, threadPool);
+
         return List.of(
                 detectorIndices, correlationIndices, correlationRuleIndices, ruleTopicIndices, customLogTypeIndices, ruleIndices,
-                mapperService, indexTemplateManager, builtinLogTypeLoader
-        );
+                mapperService, indexTemplateManager, builtinLogTypeLoader, builtInTIFMetadataLoader, threatIntelFeedDataService, detectorThreatIntelService,
+                tifJobUpdateService, tifJobParameterService, threatIntelLockService);
     }
 
     @Override
@@ -193,7 +207,8 @@ public List<NamedXContentRegistry.Entry> getNamedXContent() {
                 Detector.XCONTENT_REGISTRY,
                 DetectorInput.XCONTENT_REGISTRY,
                 Rule.XCONTENT_REGISTRY,
-                CustomLogType.XCONTENT_REGISTRY
+                CustomLogType.XCONTENT_REGISTRY,
+                ThreatIntelFeedData.XCONTENT_REGISTRY
         );
     }
 
@@ -243,7 +258,9 @@ public List<Setting<?>> getSettings() {
                 SecurityAnalyticsSettings.IS_CORRELATION_INDEX_SETTING,
                 SecurityAnalyticsSettings.CORRELATION_TIME_WINDOW,
                 SecurityAnalyticsSettings.DEFAULT_MAPPING_SCHEMA,
-                SecurityAnalyticsSettings.ENABLE_WORKFLOW_USAGE
+                SecurityAnalyticsSettings.ENABLE_WORKFLOW_USAGE,
+                SecurityAnalyticsSettings.BATCH_SIZE,
+                SecurityAnalyticsSettings.THREAT_INTEL_TIMEOUT
         );
     }
 
@@ -274,7 +291,11 @@ public List<Setting<?>> getSettings() {
                 new ActionPlugin.ActionHandler<>(SearchCorrelationRuleAction.INSTANCE, TransportSearchCorrelationRuleAction.class),
                 new ActionHandler<>(IndexCustomLogTypeAction.INSTANCE, TransportIndexCustomLogTypeAction.class),
                 new ActionHandler<>(SearchCustomLogTypeAction.INSTANCE, TransportSearchCustomLogTypeAction.class),
-                new ActionHandler<>(DeleteCustomLogTypeAction.INSTANCE, TransportDeleteCustomLogTypeAction.class)
+                new ActionHandler<>(DeleteCustomLogTypeAction.INSTANCE, TransportDeleteCustomLogTypeAction.class),
+
+                new ActionHandler<>(PutTIFJobAction.INSTANCE, TransportPutTIFJobAction.class),
+                new ActionHandler<>(DeleteTIFJobAction.INSTANCE, TransportDeleteTIFJobAction.class)
+
         );
     }
 
@@ -292,5 +313,5 @@ public void onFailure(Exception e) {
                 log.warn("Failed to initialize LogType config index and builtin log types");
             }
         });
-      }
+    }
 }

src/main/java/org/opensearch/securityanalytics/action/GetDetectorResponse.java

@@ -68,6 +68,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
                 .field(Detector.INPUTS_FIELD, detector.getInputs())
                 .field(Detector.LAST_UPDATE_TIME_FIELD, detector.getLastUpdateTime())
                 .field(Detector.ENABLED_TIME_FIELD, detector.getEnabledTime())
+                .field(Detector.THREAT_INTEL_ENABLED_FIELD, detector.getThreatIntelEnabled())
                 .endObject();
         return builder.endObject();
     }

src/main/java/org/opensearch/securityanalytics/action/IndexDetectorResponse.java

@@ -64,6 +64,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
             .field(Detector.TRIGGERS_FIELD, detector.getTriggers())
             .field(Detector.LAST_UPDATE_TIME_FIELD, detector.getLastUpdateTime())
             .field(Detector.ENABLED_TIME_FIELD, detector.getEnabledTime())
+            .field(Detector.THREAT_INTEL_ENABLED_FIELD, detector.getThreatIntelEnabled())
             .endObject();
         return builder.endObject();
     }

src/main/java/org/opensearch/securityanalytics/config/monitors/opensearch_security.policy

@@ -0,0 +1,11 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+grant {
+    permission java.lang.management.ManagementPermission "reputation.alienvault.com:443" "connect,resolve";
+};

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

@@ -25,14 +25,11 @@
 import java.io.IOException;
 import java.time.Instant;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import java.util.Locale;
 import java.util.Objects;
 
-import java.util.stream.Collectors;
-
 public class Detector implements Writeable, ToXContentObject {
 
     private static final Logger log = LogManager.getLogger(Detector.class);
@@ -51,6 +48,7 @@ public class Detector implements Writeable, ToXContentObject {
     public static final String TRIGGERS_FIELD = "triggers";
     public static final String LAST_UPDATE_TIME_FIELD = "last_update_time";
     public static final String ENABLED_TIME_FIELD = "enabled_time";
+    public static final String THREAT_INTEL_ENABLED_FIELD = "threat_intel_enabled";
     public static final String ALERTING_MONITOR_ID = "monitor_id";
 
     public static final String ALERTING_WORKFLOW_ID = "workflow_ids";
@@ -82,6 +80,8 @@ public class Detector implements Writeable, ToXContentObject {
 
     private String name;
 
+    private Boolean threatIntelEnabled;
+
     private Boolean enabled;
 
     private Schedule schedule;
@@ -122,7 +122,8 @@ public Detector(String id, Long version, String name, Boolean enabled, Schedule
                     Instant lastUpdateTime, Instant enabledTime, String logType,
                     User user, List<DetectorInput> inputs, List<DetectorTrigger> triggers, List<String> monitorIds,
                     String ruleIndex, String alertsIndex, String alertsHistoryIndex, String alertsHistoryIndexPattern,
-                    String findingsIndex, String findingsIndexPattern, Map<String, String> rulePerMonitor, List<String> workflowIds) {
+                    String findingsIndex, String findingsIndexPattern, Map<String, String> rulePerMonitor,
+                    List<String> workflowIds, Boolean threatIntelEnabled) {
         this.type = DETECTOR_TYPE;
 
         this.id = id != null ? id : NO_ID;
@@ -145,6 +146,7 @@ public Detector(String id, Long version, String name, Boolean enabled, Schedule
         this.ruleIdMonitorIdMap = rulePerMonitor;
         this.logType = logType;
         this.workflowIds = workflowIds != null ? workflowIds : null;
+        this.threatIntelEnabled = threatIntelEnabled != null && threatIntelEnabled;
 
         if (enabled) {
             Objects.requireNonNull(enabledTime);
@@ -172,7 +174,8 @@ public Detector(StreamInput sin) throws IOException {
                 sin.readString(),
                 sin.readString(),
                 sin.readMap(StreamInput::readString, StreamInput::readString),
-                sin.readStringList()
+                sin.readStringList(),
+                sin.readOptionalBoolean()
             );
     }
 
@@ -211,6 +214,7 @@ public void writeTo(StreamOutput out) throws IOException {
         if (workflowIds != null) {
             out.writeStringCollection(workflowIds);
         }
+        out.writeOptionalBoolean(threatIntelEnabled);
     }
 
     public XContentBuilder toXContentWithUser(XContentBuilder builder, Params params) throws IOException {
@@ -239,6 +243,7 @@ private XContentBuilder createXContentBuilder(XContentBuilder builder, ToXConten
             }
         }
 
+        builder.field(THREAT_INTEL_ENABLED_FIELD, threatIntelEnabled);
         builder.field(ENABLED_FIELD, enabled);
 
         if (enabledTime == null) {
@@ -280,7 +285,6 @@ private XContentBuilder createXContentBuilder(XContentBuilder builder, ToXConten
         builder.field(FINDINGS_INDEX, findingsIndex);
         builder.field(FINDINGS_INDEX_PATTERN, findingsIndexPattern);
 
-
         if (params.paramAsBoolean("with_type", false)) {
             builder.endObject();
         }
@@ -327,6 +331,7 @@ public static Detector parse(XContentParser xcp, String id, Long version) throws
         String alertsHistoryIndexPattern = null;
         String findingsIndex = null;
         String findingsIndexPattern = null;
+        Boolean enableThreatIntel = false;
 
         XContentParserUtils.ensureExpectedToken(XContentParser.Token.START_OBJECT, xcp.currentToken(), xcp);
         while (xcp.nextToken() != XContentParser.Token.END_OBJECT) {
@@ -350,6 +355,9 @@ public static Detector parse(XContentParser xcp, String id, Long version) throws
                 case ENABLED_FIELD:
                     enabled = xcp.booleanValue();
                     break;
+                case THREAT_INTEL_ENABLED_FIELD:
+                    enableThreatIntel = xcp.booleanValue();
+                    break;
                 case SCHEDULE_FIELD:
                     schedule = Schedule.parse(xcp);
                     break;
@@ -459,7 +467,8 @@ public static Detector parse(XContentParser xcp, String id, Long version) throws
                 findingsIndex,
                 findingsIndexPattern,
                 rulePerMonitor,
-                workflowIds
+                workflowIds,
+                enableThreatIntel
                 );
     }
 
@@ -600,6 +609,10 @@ public void setWorkflowIds(List<String> workflowIds) {
         this.workflowIds = workflowIds;
     }
 
+    public void setThreatIntelEnabled(boolean threatIntelEnabled) {
+        this.threatIntelEnabled = threatIntelEnabled;
+    }
+
     public List<String> getWorkflowIds() {
         return workflowIds;
     }
@@ -612,6 +625,10 @@ public boolean isWorkflowSupported() {
         return workflowIds != null && !workflowIds.isEmpty();
     }
 
+    public Boolean getThreatIntelEnabled() {
+        return threatIntelEnabled;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;