[CYB-60] Support enrichment storage configuration in all places where enrichments are used (#17)

Author: vpavlenko-cv

flink-cyber/flink-cyber-api/src/main/java/com/cloudera/cyber/enrichment/stix/parsing/ParsedThreatIntelligence.java

@@ -96,6 +96,12 @@
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-streaming-java</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-clients</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
     </dependencies>
 
 

flink-cyber/flink-cyber-api/src/main/java/com/cloudera/cyber/enrichment/stix/parsing/ThreatIntelligenceDetails.java

@@ -16,24 +16,23 @@
 import com.cloudera.cyber.commands.EnrichmentCommand;
 import com.cloudera.cyber.commands.EnrichmentCommandResponse;
 import com.cloudera.cyber.enrichemnt.stellar.StellarEnrichmentJob;
-import com.cloudera.cyber.enrichment.geocode.IpGeo;
 import com.cloudera.cyber.enrichment.cidr.IpRegionCidr;
+import com.cloudera.cyber.enrichment.geocode.IpGeo;
 import com.cloudera.cyber.enrichment.hbase.HbaseJob;
 import com.cloudera.cyber.enrichment.hbase.HbaseJobRawKafka;
+import com.cloudera.cyber.enrichment.hbase.config.EnrichmentsConfig;
 import com.cloudera.cyber.enrichment.lookup.LookupJob;
 import com.cloudera.cyber.enrichment.lookup.config.EnrichmentConfig;
 import com.cloudera.cyber.enrichment.lookup.config.EnrichmentKind;
 import com.cloudera.cyber.enrichment.rest.RestLookupJob;
 import com.cloudera.cyber.enrichment.threatq.ThreatQConfig;
-import com.cloudera.cyber.enrichment.threatq.ThreatQEntry;
 import com.cloudera.cyber.enrichment.threatq.ThreatQJob;
 import com.cloudera.cyber.flink.FlinkUtils;
 import com.cloudera.cyber.scoring.ScoredMessage;
 import com.cloudera.cyber.scoring.ScoringJob;
 import com.cloudera.cyber.scoring.ScoringRuleCommand;
 import com.cloudera.cyber.scoring.ScoringRuleCommandResult;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.flink.api.common.functions.MapFunction;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.utils.ParameterTool;
 import org.apache.flink.streaming.api.datastream.DataStream;
@@ -46,12 +45,10 @@
 import java.util.Arrays;
 import java.util.List;
 
-import static com.cloudera.cyber.enrichment.geocode.IpGeoJob.PARAM_ASN_DATABASE_PATH;
-import static com.cloudera.cyber.enrichment.geocode.IpGeoJob.PARAM_ASN_FIELDS;
-import static com.cloudera.cyber.enrichment.geocode.IpGeoJob.PARAM_GEO_DATABASE_PATH;
-import static com.cloudera.cyber.enrichment.geocode.IpGeoJob.PARAM_GEO_FIELDS;
 import static com.cloudera.cyber.enrichment.cidr.IpRegionCidrJob.PARAM_CIDR_CONFIG_PATH;
 import static com.cloudera.cyber.enrichment.cidr.IpRegionCidrJob.PARAM_CIDR_IP_FIELDS;
+import static com.cloudera.cyber.enrichment.geocode.IpGeoJob.*;
+import static com.cloudera.cyber.enrichment.hbase.HbaseJob.PARAMS_ENRICHMENT_CONFIG;
 
 @Slf4j
 public abstract class EnrichmentJob {
@@ -93,40 +90,47 @@ protected StreamExecutionEnvironment createPipeline(ParameterTool params) throws
                 Arrays.asList(params.getRequired(PARAM_CIDR_IP_FIELDS).split(",")),
                 params.getRequired(PARAM_CIDR_CONFIG_PATH)) : asnEnriched;
 
-        Tuple2<SingleOutputStreamOperator<Message>, DataStream<EnrichmentCommandResponse>> enriched = LookupJob.enrich(localEnrichments, cidrEnriched, enrichmentConfigs);
+        Tuple2<DataStream<Message>, DataStream<EnrichmentCommandResponse>> enriched = LookupJob.enrich(localEnrichments, cidrEnriched, enrichmentConfigs);
 
         DataStream<EnrichmentCommandResponse> enrichmentCommandResponses = enriched.f1;
 
-        // write the hbase enrichments to hbase
-        if (params.getBoolean(PARAMS_ENABLE_HBASE, true)) {
-            DataStream<EnrichmentCommandResponse> hbaseEnrichmentResponses = new HbaseJobRawKafka().writeEnrichments(env, params, hbaseEnrichments);
-            if (enrichmentCommandResponses != null) {
-                enrichmentCommandResponses = enriched.f1.union(hbaseEnrichmentResponses);
-            } else {
-                enrichmentCommandResponses = hbaseEnrichmentResponses;
-            }
+        boolean hbaseEnabled = params.getBoolean(PARAMS_ENABLE_HBASE, true);
+        boolean threatqEnabled = params.getBoolean(PARAMS_ENABLE_THREATQ, true);
+
+        EnrichmentsConfig enrichmentsStorageConfig = null;
+        if (hbaseEnabled || threatqEnabled) {
+            enrichmentsStorageConfig = EnrichmentsConfig.load(params.getRequired(PARAMS_ENRICHMENT_CONFIG));
         }
 
-        writeEnrichmentQueryResults(env, params, enrichmentCommandResponses);
 
-        DataStream<Message> hbased = params.getBoolean(PARAMS_ENABLE_HBASE, true) ?
-                HbaseJob.enrich(enriched.f0, enrichmentConfigs) : enriched.f0;
+        DataStream<Message> hbased = hbaseEnabled ?
+                HbaseJob.enrich(enriched.f0, enrichmentConfigs, enrichmentsStorageConfig) : enriched.f0;
 
         // rest based enrichments
         DataStream<Message> rested = params.getBoolean(PARAMS_ENABLE_REST, true) ?
                 RestLookupJob.enrich(hbased, params.getRequired(PARAMS_REST_CONFIG_FILE)) : hbased;
 
         // Run threatQ integrations
         DataStream<Message> tqed;
-        if (params.getBoolean(PARAMS_ENABLE_THREATQ, true)) {
+        DataStream<EnrichmentCommand> threatqEnrichments = null;
+        if (threatqEnabled) {
             List<ThreatQConfig> threatQconfigs = ThreatQJob.parseConfigs(Files.readAllBytes(Paths.get(params.getRequired(PARAMS_THREATQ_CONFIG_FILE))));
             log.info("ThreatQ Configs {}", threatQconfigs);
-            tqed = ThreatQJob.enrich(rested, threatQconfigs);
-            ThreatQJob.ingest(createThreatQSource(env, params), threatQconfigs);
+            tqed = ThreatQJob.enrich(rested, threatQconfigs, enrichmentsStorageConfig);
+            threatqEnrichments = createThreatQSource(env, params);
         } else {
             tqed = rested;
         }
 
+        DataStream<EnrichmentCommandResponse> hbaseTqEnrichResults = null;
+        if (hbaseEnabled || threatqEnabled) {
+            hbaseTqEnrichResults = writeHbaseThreatQEnrichmentsToHbaseAndRespond(params, env, hbaseEnrichments, threatqEnrichments, enrichmentsStorageConfig);
+        }
+
+
+        // write the local, hbase, and threatq responses to the output topic
+        writeEnrichmentQueryResults(env, params, unionStreams(enrichmentCommandResponses, hbaseTqEnrichResults));
+
         DataStream<Message> stellarStream;
         if (params.getBoolean(PARAMS_ENABLE_STELLAR, true)) {
             String configDir = params.getRequired(StellarEnrichmentJob.PARAMS_CONFIG_DIR);
@@ -139,21 +143,37 @@ protected StreamExecutionEnvironment createPipeline(ParameterTool params) throws
 
         // disabled by default - NOT IMPLEMENTED
         DataStream<Message> ruled = params.getBoolean(PARAMS_ENABLE_RULES, false) ?
-                doRules(stellarStream, params) : stellarStream;
+                doRules(stellarStream) : stellarStream;
 
         DataStream<ScoredMessage> scoring = doScoring(ruled, env, params);
 
         writeResults(env, params, scoring);
         return env;
     }
 
+    private DataStream<EnrichmentCommandResponse> writeHbaseThreatQEnrichmentsToHbaseAndRespond(ParameterTool params, StreamExecutionEnvironment env,
+                                                                                                DataStream<EnrichmentCommand> hbaseEnrichments, DataStream<EnrichmentCommand> threatqEnrichments,
+                                                                                                EnrichmentsConfig enrichmentsStorageConfig) {
+
+        // write the threatq and hbase enrichments to hbase
+        return new HbaseJobRawKafka().writeEnrichments(env, params, unionStreams(hbaseEnrichments, threatqEnrichments), enrichmentsStorageConfig);
+    }
+
+    private static<T> DataStream<T> unionStreams(DataStream<T> stream1, DataStream<T> stream2) {
+        if (stream1 != null && stream2 != null) {
+            return stream1.union(stream2);
+        } else if (stream1 != null) {
+            return stream1;
+        } else {
+            return stream2;
+        }
+    }
 
     /**
      * @param in     Messages incoming for rules processing
-     * @param params Global Job Parameters
      * @return incoming stream for now.  Not implemented.
      */
-    private DataStream<Message> doRules(DataStream<Message> in, ParameterTool params) {
+    private DataStream<Message> doRules(DataStream<Message> in) {
         return in;
     }
 
@@ -172,11 +192,7 @@ private DataStream<ScoredMessage> doScoring(DataStream<Message> in, StreamExecut
 
     protected abstract void writeEnrichmentQueryResults(StreamExecutionEnvironment env, ParameterTool params, DataStream<EnrichmentCommandResponse> sideOutput);
 
-    protected abstract DataStream<ThreatQEntry> createThreatQSource(StreamExecutionEnvironment env, ParameterTool params);
-
-    protected MapFunction<Message, Message> getLongTermLookupFunction() {
-        return null;
-    }
+    protected abstract DataStream<EnrichmentCommand> createThreatQSource(StreamExecutionEnvironment env, ParameterTool params);
 
     protected abstract DataStream<ScoringRuleCommand> createRulesSource(StreamExecutionEnvironment env, ParameterTool params);
 

flink-cyber/flink-enrichment/flink-enrichment-combined/pom.xml

@@ -15,7 +15,6 @@
 import com.cloudera.cyber.Message;
 import com.cloudera.cyber.commands.EnrichmentCommand;
 import com.cloudera.cyber.commands.EnrichmentCommandResponse;
-import com.cloudera.cyber.enrichment.threatq.ThreatQEntry;
 import com.cloudera.cyber.enrichment.threatq.ThreatQParserFlatMap;
 import com.cloudera.cyber.flink.FlinkUtils;
 import com.cloudera.cyber.flink.Utils;
@@ -44,7 +43,6 @@ public class EnrichmentJobKafka extends EnrichmentJob {
     private static final String PARAMS_GROUP_ID = "group.id";
     private static final String DEFAULT_GROUP_ID = "enrichment-combined";
     public static final String SCORING_RULES_GROUP_ID = "scoring-rules";
-    private static final String DEFAULT_TI_TABLE = "threatIntelligence";
     private static final String PARAMS_TOPIC_THREATQ_INPUT = "threatq.topic.input";
 
 
@@ -81,7 +79,7 @@ protected void writeEnrichmentQueryResults(StreamExecutionEnvironment env, Param
     }
 
     @Override
-    protected DataStream<ThreatQEntry> createThreatQSource(StreamExecutionEnvironment env, ParameterTool params) {
+    protected DataStream<EnrichmentCommand> createThreatQSource(StreamExecutionEnvironment env, ParameterTool params) {
         String topic = params.getRequired(PARAMS_TOPIC_THREATQ_INPUT);
         String groupId = "threatq-parser";
 

flink-cyber/flink-enrichment/flink-enrichment-combined/src/main/java/com/cloudera/cyber/enrichment/EnrichmentJob.java

@@ -127,6 +127,39 @@ The majestic_million enrichment is ingested in batch only.
 }
 ```
 
+### Reserved Enrichment Types
+There are two reserved enrichment names: threatq and first_seen.   Reserved enrichments define a mapping to the hbase table, column family and storage format.
+The keys and values are defined by the enrichment and profile jobs.  
+
+* The first_seen enrichment mapping is ignored.  Specify the hbase table and format using the profile properties file settings.  
+
+* If there is no specific mapping for threatq enrichment, the default table and format is used.
+
+* To store threatq enrichments in a different table and format, override the threatq enrichment.  In the example below the threatq enrichment is store in the threatq table 
+and cf column family with the HBASE_METRON format.
+```
+{
+  "storageConfigs": {
+    "default": {
+      "format": "HBASE_METRON",
+      "hbaseTableName": "enrich_default",
+      "columnFamily": "cf"
+    },
+    "threatq": {
+      "format": "HBASE_METRON",
+      "hbaseTableName": "threatq",
+      "columnFamily": "cf"
+    }
+  },
+  "enrichmentConfigs": {
+    "threatq" : {
+       "storage": "threatq",
+       "fieldMapping": {
+       }
+    }
+}
+```
+
 ## EnrichmentsConfig Json
 
 | Json Field       | Type                              | Description                | Required/Default | 

flink-cyber/flink-enrichment/flink-enrichment-combined/src/main/java/com/cloudera/cyber/enrichment/EnrichmentJobKafka.java

@@ -0,0 +1,10 @@
+package com.cloudera.cyber.enrichment.hbase;
+
+import com.cloudera.cyber.enrichment.hbase.config.EnrichmentStorageConfig;
+import com.cloudera.cyber.hbase.LookupKey;
+
+import java.io.Serializable;
+
+public interface EnrichmentLookupBuilder extends Serializable {
+    LookupKey build(EnrichmentStorageConfig storageConfig, String enrichmentType, String fieldValue);
+}