Adds support for uploading threat intelligence in Custom Format JSON

build.gradle

@@ -173,6 +173,9 @@ dependencies {
     compileOnly "org.opensearch:opensearch-job-scheduler-spi:${opensearch_build}"
     compileOnly "org.opensearch.alerting:alerting-spi:${alerting_spi_build}"
     implementation "org.apache.commons:commons-csv:1.10.0"
+    implementation 'com.jayway.jsonpath:json-path:2.9.0'
+    implementation 'net.minidev:json-smart:2.5.0'
+    implementation 'net.minidev:accessors-smart:2.5.0'
     compileOnly "com.google.guava:guava:32.1.3-jre"
 
     // TODO uncomment once SA commons is published to maven central

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

@@ -12,14 +12,12 @@
 import org.opensearch.core.xcontent.XContentBuilder;
 import org.opensearch.core.xcontent.XContentParser;
 import org.opensearch.core.xcontent.XContentParserUtils;
-import org.opensearch.securityanalytics.commons.model.IOCType;
 import org.opensearch.securityanalytics.commons.model.STIX2;
 
 import java.io.IOException;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Locale;
 
 /**
  * A data transfer object for STIX2IOC containing additional details.
@@ -58,7 +56,7 @@ public static DetailedSTIX2IOCDto parse(XContentParser xcp, String id, Long vers
         }
 
         String name = null;
-        IOCType type = null;
+        String type = null;
         String value = null;
         String severity = null;
         Instant created = null;
@@ -89,7 +87,7 @@ public static DetailedSTIX2IOCDto parse(XContentParser xcp, String id, Long vers
                     name = xcp.text();
                     break;
                 case STIX2.TYPE_FIELD:
-                    type = new IOCType(xcp.text().toLowerCase(Locale.ROOT));
+                    type = xcp.text();
                     break;
                 case STIX2.VALUE_FIELD:
                     value = xcp.text();

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

@@ -11,22 +11,17 @@
 import org.opensearch.core.common.io.stream.StreamInput;
 import org.opensearch.core.common.io.stream.StreamOutput;
 import org.opensearch.core.common.io.stream.Writeable;
-import org.opensearch.core.rest.RestStatus;
 import org.opensearch.core.xcontent.ToXContentObject;
 import org.opensearch.core.xcontent.XContentBuilder;
 import org.opensearch.core.xcontent.XContentParser;
 import org.opensearch.core.xcontent.XContentParserUtils;
-import org.opensearch.securityanalytics.commons.model.IOCType;
 import org.opensearch.securityanalytics.commons.model.STIX2;
-import org.opensearch.securityanalytics.util.SecurityAnalyticsException;
 import org.opensearch.securityanalytics.util.XContentUtils;
 
 import java.io.IOException;
 import java.time.Instant;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.List;
-import java.util.Locale;
 import java.util.UUID;
 
 public class STIX2IOC extends STIX2 implements Writeable, ToXContentObject {
@@ -46,7 +41,7 @@ public STIX2IOC() {
     public STIX2IOC(
             String id,
             String name,
-            IOCType type,
+            String type,
             String value,
             String severity,
             Instant created,
@@ -86,7 +81,7 @@ public STIX2IOC(StreamInput sin) throws IOException {
         this(
                 sin.readString(), // id
                 sin.readString(), // name
-                new IOCType(sin.readString()), // type
+                sin.readString(), // type
                 sin.readString(), // value
                 sin.readString(), // severity
                 sin.readInstant(), // created
@@ -186,7 +181,7 @@ public static STIX2IOC parse(XContentParser xcp, String id, Long version) throws
         }
 
         String name = null;
-        IOCType type = null;
+        String type = null;
         String value = null;
         String severity = null;
         Instant created = null;
@@ -204,26 +199,27 @@ public static STIX2IOC parse(XContentParser xcp, String id, Long version) throws
 
             switch (fieldName) {
                 case NAME_FIELD:
+                    if (xcp.currentToken() == XContentParser.Token.VALUE_NULL) {
+                        break;
+                    }
                     name = xcp.text();
                     break;
                 case TYPE_FIELD:
-                    String typeString = xcp.text();
-                    try {
-                        type = new IOCType(typeString);
-                    } catch (Exception e) {
-                        String error = String.format(
-                                "Couldn't parse IOC type '%s' while deserializing STIX2IOC with ID '%s': ",
-                                typeString,
-                                id
-                        );
-                        logger.error(error, e);
-                        throw new SecurityAnalyticsException(error, RestStatus.BAD_REQUEST, e);
+                    if (xcp.currentToken() == XContentParser.Token.VALUE_NULL) {
+                        break;
                     }
+                    type = xcp.text();
                     break;
                 case VALUE_FIELD:
+                    if (xcp.currentToken() == XContentParser.Token.VALUE_NULL) {
+                        break;
+                    }
                     value = xcp.text();
                     break;
                 case SEVERITY_FIELD:
+                    if (xcp.currentToken() == XContentParser.Token.VALUE_NULL) {
+                        break;
+                    }
                     severity = xcp.text();
                     break;
                 case CREATED_FIELD:
@@ -255,6 +251,9 @@ public static STIX2IOC parse(XContentParser xcp, String id, Long version) throws
                     }
                     break;
                 case DESCRIPTION_FIELD:
+                    if (xcp.currentToken() == XContentParser.Token.VALUE_NULL) {
+                        break;
+                    }
                     description = xcp.text();
                     break;
                 case LABELS_FIELD:
@@ -267,12 +266,21 @@ public static STIX2IOC parse(XContentParser xcp, String id, Long version) throws
                     }
                     break;
                 case SPEC_VERSION_FIELD:
+                    if (xcp.currentToken() == XContentParser.Token.VALUE_NULL) {
+                        break;
+                    }
                     specVersion = xcp.text();
                     break;
                 case FEED_ID_FIELD:
+                    if (xcp.currentToken() == XContentParser.Token.VALUE_NULL) {
+                        break;
+                    }
                     feedId = xcp.text();
                     break;
                 case FEED_NAME_FIELD:
+                    if (xcp.currentToken() == XContentParser.Token.VALUE_NULL) {
+                        break;
+                    }
                     feedName = xcp.text();
                     break;
                 default:
@@ -305,9 +313,6 @@ public static STIX2IOC parse(XContentParser xcp, String id, Long version) throws
     public void validate() throws IllegalArgumentException {
         if (super.getType() == null) {
             throw new IllegalArgumentException(String.format("[%s] is required.", TYPE_FIELD));
-        } else if (!IOCType.supportedType(super.getType().toString())) {
-            logger.debug("Unsupported IOCType: {}", super.getType().toString());
-            throw new IllegalArgumentException(String.format("[%s] is not supported.", TYPE_FIELD));
         }
 
         if (super.getValue() == null || super.getValue().isEmpty()) {

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

@@ -10,14 +10,11 @@
 import org.opensearch.core.common.io.stream.StreamInput;
 import org.opensearch.core.common.io.stream.StreamOutput;
 import org.opensearch.core.common.io.stream.Writeable;
-import org.opensearch.core.rest.RestStatus;
 import org.opensearch.core.xcontent.ToXContentObject;
 import org.opensearch.core.xcontent.XContentBuilder;
 import org.opensearch.core.xcontent.XContentParser;
 import org.opensearch.core.xcontent.XContentParserUtils;
-import org.opensearch.securityanalytics.commons.model.IOCType;
 import org.opensearch.securityanalytics.commons.model.STIX2;
-import org.opensearch.securityanalytics.util.SecurityAnalyticsException;
 
 import java.io.IOException;
 import java.time.Instant;
@@ -32,7 +29,7 @@ public class STIX2IOCDto implements Writeable, ToXContentObject {
 
     private String id;
     private String name;
-    private IOCType type;
+    private String type;
     private String value;
     private String severity;
     private Instant created;
@@ -50,7 +47,7 @@ public STIX2IOCDto() {}
     public STIX2IOCDto(
             String id,
             String name,
-            IOCType type,
+            String type,
             String value,
             String severity,
             Instant created,
@@ -149,7 +146,7 @@ public static STIX2IOCDto parse(XContentParser xcp, String id, Long version) thr
         }
 
         String name = null;
-        IOCType type = null;
+        String type = null;
         String value = null;
         String severity = null;
         Instant created = null;
@@ -177,27 +174,24 @@ public static STIX2IOCDto parse(XContentParser xcp, String id, Long version) thr
                     }
                     break;
                 case STIX2.NAME_FIELD:
-                    name = xcp.text();
+                    if (xcp.currentToken() != XContentParser.Token.VALUE_NULL) {
+                        name = xcp.text();
+                    }
                     break;
                 case STIX2.TYPE_FIELD:
-                    String typeString = xcp.text();
-                    try {
-                        type = new IOCType(typeString);
-                    } catch (Exception e) {
-                        String error = String.format(
-                                "Couldn't parse IOC type '%s' while deserializing STIX2IOCDto with ID '%s': ",
-                                typeString,
-                                id
-                        );
-                        logger.error(error, e);
-                        throw new SecurityAnalyticsException(error, RestStatus.BAD_REQUEST, e);
+                    if (xcp.currentToken() != XContentParser.Token.VALUE_NULL) {
+                        type = xcp.text();
                     }
                     break;
                 case STIX2.VALUE_FIELD:
-                    value = xcp.text();
+                    if (xcp.currentToken() != XContentParser.Token.VALUE_NULL) {
+                        value = xcp.text();
+                    }
                     break;
                 case STIX2.SEVERITY_FIELD:
-                    severity = xcp.text();
+                    if (xcp.currentToken() != XContentParser.Token.VALUE_NULL) {
+                        severity = xcp.text();
+                    }
                     break;
                 case STIX2.CREATED_FIELD:
                     if (xcp.currentToken() == XContentParser.Token.VALUE_NULL) {
@@ -228,7 +222,9 @@ public static STIX2IOCDto parse(XContentParser xcp, String id, Long version) thr
                     }
                     break;
                 case STIX2.DESCRIPTION_FIELD:
-                    description = xcp.text();
+                    if (xcp.currentToken() != XContentParser.Token.VALUE_NULL) {
+                        description = xcp.text();
+                    }
                     break;
                 case STIX2.LABELS_FIELD:
                     XContentParserUtils.ensureExpectedToken(XContentParser.Token.START_ARRAY, xcp.currentToken(), xcp);
@@ -240,13 +236,19 @@ public static STIX2IOCDto parse(XContentParser xcp, String id, Long version) thr
                     }
                     break;
                 case STIX2.SPEC_VERSION_FIELD:
-                    specVersion = xcp.text();
+                    if (xcp.currentToken() != XContentParser.Token.VALUE_NULL) {
+                        specVersion = xcp.text();
+                    }
                     break;
                 case STIX2IOC.FEED_ID_FIELD:
-                    feedId = xcp.text();
+                    if (xcp.currentToken() != XContentParser.Token.VALUE_NULL) {
+                        feedId = xcp.text();
+                    }
                     break;
                 case STIX2IOC.FEED_NAME_FIELD:
-                    feedName = xcp.text();
+                    if (xcp.currentToken() != XContentParser.Token.VALUE_NULL) {
+                        feedName = xcp.text();
+                    }
                     break;
                 default:
                     xcp.skipChildren();
@@ -286,11 +288,11 @@ public void setName(String name) {
         this.name = name;
     }
 
-    public IOCType getType() {
+    public String getType() {
         return type;
     }
 
-    public void setType(IOCType type) {
+    public void setType(String type) {
         this.type = type;
     }
 

src/main/java/org/opensearch/securityanalytics/services/JsonPathAwareInputCodec.java

@@ -0,0 +1,42 @@
+package org.opensearch.securityanalytics.services;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.opensearch.securityanalytics.commons.connector.codec.InputCodec;
+import org.opensearch.securityanalytics.model.STIX2IOC;
+import org.opensearch.securityanalytics.threatIntel.model.JsonPathIocSchema;
+import org.opensearch.securityanalytics.threatIntel.model.SATIFSourceConfig;
+import org.opensearch.securityanalytics.threatIntel.service.JsonPathIocSchemaThreatIntelHandler;
+
+import java.io.InputStream;
+import java.util.List;
+import java.util.function.Consumer;
+
+/**
+ * An implementation of InputCodec used to parse input stream using JsonPath notations from {@link JsonPathIocSchema} and build a list of {@link STIX2IOC} objects
+ */
+public class JsonPathAwareInputCodec implements InputCodec<STIX2IOC> {
+    private static final Logger logger = LogManager.getLogger(JsonPathAwareInputCodec.class);
+    private final SATIFSourceConfig satifSourceConfig;
+
+    public JsonPathAwareInputCodec(SATIFSourceConfig satifSourceConfig) {
+        this.satifSourceConfig = satifSourceConfig;
+    }
+
+    @Override
+    public void parse(final InputStream inputStream, final Consumer<STIX2IOC> consumer) {
+        try {
+            List<STIX2IOC> stix2IOCS = JsonPathIocSchemaThreatIntelHandler.parseCustomSchema(
+                    (JsonPathIocSchema) satifSourceConfig.getIocSchema(), inputStream, satifSourceConfig.getName(), satifSourceConfig.getId());
+            stix2IOCS.forEach(ioc -> {
+                try {
+                    consumer.accept(ioc);
+                } catch (Exception e) {
+                    logger.error(String.format("Error while indexing STIX2Ioc - type [%s], value [%s]"), e);
+                }
+            });
+        } catch (Exception e) {
+            logger.error(String.format("Error while downloading and indexing STIX2Ioc"), e);
+        }
+    }
+}

src/main/java/org/opensearch/securityanalytics/services/STIX2IOCConnectorFactory.java

@@ -18,6 +18,7 @@
 import org.opensearch.securityanalytics.commons.model.FeedConfiguration;
 import org.opensearch.securityanalytics.commons.model.FeedLocation;
 import org.opensearch.securityanalytics.commons.model.STIX2;
+import org.opensearch.securityanalytics.threatIntel.model.SATIFSourceConfig;
 import software.amazon.awssdk.services.s3.S3Client;
 
 import java.util.List;
@@ -35,16 +36,34 @@ public STIX2IOCConnectorFactory(final InputCodecFactory inputCodecFactory, final
     protected Connector<STIX2> doCreate(FeedConfiguration feedConfiguration) {
         final FeedLocation feedLocation = FeedLocation.fromFeedConfiguration(feedConfiguration);
         logger.debug("FeedLocation: {}", feedLocation);
-        switch(feedLocation) {
-            case S3: return createS3Connector(feedConfiguration);
-            default: throw new IllegalArgumentException("Unsupported feedLocation: " + feedLocation);
+        switch (feedLocation) {
+            case S3:
+                return createS3Connector(feedConfiguration, null);
+            default:
+                throw new IllegalArgumentException("Unsupported feedLocation: " + feedLocation);
         }
     }
 
-    private S3Connector<STIX2> createS3Connector(final FeedConfiguration feedConfiguration) {
+    protected Connector<STIX2> doCreate(FeedConfiguration feedConfiguration, SATIFSourceConfig satifSourceConfig) {
+        final FeedLocation feedLocation = FeedLocation.fromFeedConfiguration(feedConfiguration);
+        logger.debug("FeedLocation: {}", feedLocation);
+        switch (feedLocation) {
+            case S3:
+                return createS3Connector(feedConfiguration, satifSourceConfig);
+            default:
+                throw new IllegalArgumentException("Unsupported feedLocation: " + feedLocation);
+        }
+    }
+
+    private S3Connector<STIX2> createS3Connector(final FeedConfiguration feedConfiguration, SATIFSourceConfig satifSourceConfig) {
         final S3ConnectorConfig s3ConnectorConfig = feedConfiguration.getS3ConnectorConfig();
         final S3Client s3Client = s3ClientFactory.create(s3ConnectorConfig.getRoleArn(), s3ConnectorConfig.getRegion());
-        final InputCodec inputCodec = inputCodecFactory.create(feedConfiguration.getIocSchema().getModelClass(), feedConfiguration.getInputCodecSchema());
+        final InputCodec inputCodec;
+        if (satifSourceConfig.getIocSchema() != null) {
+            inputCodec = new JsonPathAwareInputCodec(satifSourceConfig);
+        } else {
+            inputCodec = inputCodecFactory.create(feedConfiguration.getIocSchema().getModelClass(), feedConfiguration.getInputCodecSchema());
+        }
         return new S3Connector<>(s3ConnectorConfig, s3Client, inputCodec);
     }