Integrate threat intel feeds

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 {
@@ -142,12 +143,6 @@ repositories {
 sourceSets.main.java.srcDirs = ['src/main/generated','src/main/java']
 configurations {
     zipArchive
-
-    all {
-        resolutionStrategy {
-            force "com.google.guava:guava:32.0.1-jre"
-        }
-    }
 }
 
 dependencies {
@@ -158,17 +153,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}"
+    compileOnly "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}"
-
-    //spotless
-    implementation('com.google.googlejavaformat:google-java-format:1.17.0') {
-        exclude group: 'com.google.guava'
-    }
-    implementation 'com.google.guava:guava:32.0.1-jre'
+    zipArchive group: 'org.opensearch.plugin', name:'opensearch-job-scheduler', version: "${opensearch_build}"
 }
 
 // RPM & Debian build
@@ -289,6 +281,22 @@ testClusters.integTest {
             }
         }
     }))
+    plugin(provider({
+        new RegularFile() {
+            @Override
+            File getAsFile() {
+                return configurations.zipArchive.asFileTree.matching {
+                    include '**/opensearch-job-scheduler*'
+                }.singleFile
+            }
+        }
+    }))
+    nodes.each { node ->
+        def plugins = node.plugins
+        def firstPlugin = plugins.get(0)
+        plugins.remove(0)
+        plugins.add(firstPlugin)
+    }
 }
 
 run {

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

@@ -8,16 +8,14 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 import java.util.function.Supplier;
+import java.util.Optional;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.opensearch.cluster.routing.Preference;
+import org.opensearch.cluster.metadata.IndexMetadata;
 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 +36,21 @@
 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.indices.SystemIndexDescriptor;
+import org.opensearch.jobscheduler.spi.JobSchedulerExtension;
+import org.opensearch.jobscheduler.spi.ScheduledJobParser;
+import org.opensearch.jobscheduler.spi.ScheduledJobRunner;
 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.plugins.SystemIndexPlugin;
 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 +61,18 @@
 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.PutTIFJobAction;
+import org.opensearch.securityanalytics.threatIntel.action.TransportPutTIFJobAction;
+import org.opensearch.securityanalytics.threatIntel.common.TIFLockService;
+import org.opensearch.securityanalytics.threatIntel.feedMetadata.BuiltInTIFMetadataLoader;
+import org.opensearch.securityanalytics.threatIntel.jobscheduler.TIFJobParameter;
+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;
@@ -75,7 +87,9 @@
 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, JobSchedulerExtension {
 
     private static final Logger log = LogManager.getLogger(SecurityAnalyticsPlugin.class);
 
@@ -91,6 +105,8 @@
     public static final String CORRELATION_RULES_BASE_URI = PLUGINS_BASE_URI + "/correlation/rules";
 
     public static final String CUSTOM_LOG_TYPE_URI = PLUGINS_BASE_URI + "/logtype";
+    public static final String JOB_INDEX_NAME = ".opensearch-sap--job";
+    public static final Map<String, Object> TIF_JOB_INDEX_SETTING = Map.of(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1, IndexMetadata.SETTING_AUTO_EXPAND_REPLICAS, "0-all", IndexMetadata.SETTING_INDEX_HIDDEN, true);
 
     private CorrelationRuleIndices correlationRuleIndices;
 
@@ -113,8 +129,12 @@
     private BuiltinLogTypeLoader builtinLogTypeLoader;
 
     private LogTypeService logTypeService;
+    @Override
+    public Collection<SystemIndexDescriptor> getSystemIndexDescriptors(Settings settings){
+        return Collections.singletonList(new SystemIndexDescriptor(THREAT_INTEL_DATA_INDEX_NAME_PREFIX, "System index used for threat intel data"));
+    }
+
 
-    private Client client;
 
     @Override
     public Collection<Object> createComponents(Client client,
@@ -128,7 +148,9 @@
                                                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 +160,18 @@
         mapperService = new MapperService(client, clusterService, indexNameExpressionResolver, indexTemplateManager, logTypeService);
         ruleIndices = new RuleIndices(logTypeService, client, clusterService, threadPool);
         correlationRuleIndices = new CorrelationRuleIndices(client, clusterService);
-        this.client = client;
+        ThreatIntelFeedDataService threatIntelFeedDataService = new ThreatIntelFeedDataService(clusterService, client, indexNameExpressionResolver, xContentRegistry);
+        DetectorThreatIntelService detectorThreatIntelService = new DetectorThreatIntelService(threatIntelFeedDataService, client, xContentRegistry);
+        TIFJobParameterService tifJobParameterService = new TIFJobParameterService(client, clusterService);
+        TIFJobUpdateService tifJobUpdateService = new TIFJobUpdateService(clusterService, tifJobParameterService, threatIntelFeedDataService, builtInTIFMetadataLoader);
+        TIFLockService threatIntelLockService = new TIFLockService(clusterService, client);
+
+        TIFJobRunner.getJobRunnerInstance().initialize(clusterService, tifJobUpdateService, tifJobParameterService, threatIntelLockService, threadPool, detectorThreatIntelService);
 
         return List.of(
                 detectorIndices, correlationIndices, correlationRuleIndices, ruleTopicIndices, customLogTypeIndices, ruleIndices,
-                mapperService, indexTemplateManager, builtinLogTypeLoader
-        );
+                mapperService, indexTemplateManager, builtinLogTypeLoader, builtInTIFMetadataLoader, threatIntelFeedDataService, detectorThreatIntelService,
+                tifJobUpdateService, tifJobParameterService, threatIntelLockService);
     }
 
     @Override
@@ -187,13 +215,34 @@
         );
     }
 
+    @Override
+    public String getJobType() {
+        return "opensearch_sap_job";
+    }
+
+    @Override
+    public String getJobIndex() {
+        return JOB_INDEX_NAME;
+    }
+
+    @Override
+    public ScheduledJobRunner getJobRunner() {
+        return TIFJobRunner.getJobRunnerInstance();
+    }
+
+    @Override
+    public ScheduledJobParser getJobParser() {
+        return (parser, id, jobDocVersion) -> TIFJobParameter.PARSER.parse(parser, null);
+    }
+
     @Override
     public List<NamedXContentRegistry.Entry> getNamedXContent() {
         return List.of(
                 Detector.XCONTENT_REGISTRY,
                 DetectorInput.XCONTENT_REGISTRY,
                 Rule.XCONTENT_REGISTRY,
-                CustomLogType.XCONTENT_REGISTRY
+                CustomLogType.XCONTENT_REGISTRY,
+                ThreatIntelFeedData.XCONTENT_REGISTRY
         );
     }
 
@@ -243,7 +292,10 @@
                 SecurityAnalyticsSettings.IS_CORRELATION_INDEX_SETTING,
                 SecurityAnalyticsSettings.CORRELATION_TIME_WINDOW,
                 SecurityAnalyticsSettings.DEFAULT_MAPPING_SCHEMA,
-                SecurityAnalyticsSettings.ENABLE_WORKFLOW_USAGE
+                SecurityAnalyticsSettings.ENABLE_WORKFLOW_USAGE,
+                SecurityAnalyticsSettings.TIF_UPDATE_INTERVAL,
+                SecurityAnalyticsSettings.BATCH_SIZE,
+                SecurityAnalyticsSettings.THREAT_INTEL_TIMEOUT
         );
     }
 
@@ -274,7 +326,8 @@
                 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)
         );
     }
 
@@ -292,5 +345,5 @@
                 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 @@
                 .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/GetMappingsViewResponse.java

@@ -4,37 +4,49 @@
  */
 package org.opensearch.securityanalytics.action;
 
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
+import org.opensearch.common.xcontent.XContentType;
 import org.opensearch.core.action.ActionResponse;
 import org.opensearch.core.common.Strings;
 import org.opensearch.core.common.io.stream.StreamInput;
 import org.opensearch.core.common.io.stream.StreamOutput;
-import org.opensearch.common.xcontent.XContentType;
+import org.opensearch.core.common.io.stream.Writeable;
 import org.opensearch.core.xcontent.ToXContentObject;
 import org.opensearch.core.xcontent.XContentBuilder;
 import org.opensearch.securityanalytics.mapper.MapperUtils;
+import org.opensearch.securityanalytics.model.LogType;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
 
 public class GetMappingsViewResponse extends ActionResponse implements ToXContentObject {
 
     public static final String UNMAPPED_INDEX_FIELDS = "unmapped_index_fields";
     public static final String UNMAPPED_FIELD_ALIASES = "unmapped_field_aliases";
+    public static final String THREAT_INTEL_FIELD_ALIASES = "threat_intel_field_aliases";
 
     private Map<String, Object> aliasMappings;
     List<String> unmappedIndexFields;
     List<String> unmappedFieldAliases;
 
+    /** This field sheds information on the list of field aliases that need to be mapped for a given IoC.
+     * For ex. one element for windows logtype would be
+     *{"ioc": "ip", "fields":  ["destination.ip","source.ip"]} where "ip" is the IoC and the required field aliases to be mapped for
+     * threat intel based detection are "destination.ip","source.ip".*/
+    private List<LogType.IocFields> threatIntelFieldAliases;
+
     public GetMappingsViewResponse(
             Map<String, Object> aliasMappings,
             List<String> unmappedIndexFields,
-            List<String> unmappedFieldAliases
+            List<String> unmappedFieldAliases,
+            List<LogType.IocFields> threatIntelFieldAliases
             ) {
         this.aliasMappings = aliasMappings;
         this.unmappedIndexFields = unmappedIndexFields;
         this.unmappedFieldAliases = unmappedFieldAliases;
+        this.threatIntelFieldAliases = threatIntelFieldAliases;
     }
 
     public GetMappingsViewResponse(StreamInput in) throws IOException {
@@ -56,6 +68,7 @@
                 unmappedFieldAliases.add(in.readString());
             }
         }
+        this.threatIntelFieldAliases = in.readList(LogType.IocFields::readFrom);
     }
 
     @Override
@@ -82,6 +95,12 @@
         } else {
             out.writeVInt(0);
         }
+        if(threatIntelFieldAliases!=null) {
+            out.writeBoolean(true);
+            out.writeCollection(threatIntelFieldAliases);
+        }  else {
+            out.writeBoolean(false);
+        }
     }
 
     @Override
@@ -96,6 +115,9 @@
         if (unmappedFieldAliases != null && unmappedFieldAliases.size() > 0) {
             builder.field(UNMAPPED_FIELD_ALIASES, unmappedFieldAliases);
         }
+        if(threatIntelFieldAliases != null && false == threatIntelFieldAliases.isEmpty()) {
+            builder.field(THREAT_INTEL_FIELD_ALIASES, threatIntelFieldAliases);
+        }
         return builder.endObject();
     }
 

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

@@ -64,6 +64,7 @@
             .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/logtype/LogTypeService.java

@@ -10,6 +10,7 @@
 import java.net.URISyntaxException;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -660,6 +661,13 @@
         return;
     }
 
+    public List<LogType.IocFields> getIocFieldsList(String logType) {
+        LogType logTypeByName = builtinLogTypeLoader.getLogTypeByName(logType);
+        if(logTypeByName == null)
+            return Collections.emptyList();
+        return logTypeByName.getIocFieldsList();
+    }
+
     public void getRuleFieldMappingsAllSchemas(String logType, ActionListener<List<LogType.Mapping>> listener) {
 
         if (builtinLogTypeLoader.logTypeExists(logType)) {