Adding custom enrichment.

[Laurens Vets]

Hi list,

I want to enrich AWS Cloudtrail events with an extra field "is_us" 
("yes" or "no") which shows whether the source ip address in my events 
is from our network or not.

I created the file my_subnets.csv with the following content:
1.2.3.0/24;AS1230;Company1
1.2.4.0/24;AS1240;Company2

These are some example events from Cloudtrail:

{"eventVersion":"1.05","userIdentity":{"type":"AssumedRole","principalId":"<PRINCIPALID>:secmonkey","arn":"arn:aws:sts::<ACCOUNT>:assumed-role/SecurityMonkey/secmonkey","accountId":"<ACCOUNT>","accessKeyId":"<ACCESSKEY>","sessionContext":{"attributes":{"mfaAuthenticated":"false","creationDate":"2017-07-01T15:54:02Z"},"sessionIssuer":{"type":"Role","principalId":"<PRINCIPALID>","arn":"arn:aws:iam::<ACCOUNT>:role/SecurityMonkey","accountId":"<ACCOUNT>","userName":"SecurityMonkey"}}},"eventTime":"2017-07-01T15:54:03Z","eventSource":"ec2.amazonaws.com","eventName":"DescribeInstances","awsRegion":"ca-central-1","sourceIPAddress":"<SOURCEIP>","userAgent":"Boto3/1.4.4 
Python/2.7.12 Linux/4.4.0-1016-aws 
Botocore/1.5.42","requestParameters":{"instancesSet":{},"filterSet":{}},"responseElements":null,"requestID":"<REQUESTID>","eventID":"<EVENTID>","eventType":"AwsApiCall","recipientAccountId":"<ACCOUNT>"}

{"eventVersion":"1.04","userIdentity":{"type":"AssumedRole","principalId":"<PRINCIPALID>:secmonkey","arn":"arn:aws:sts::<ACCOUNT>:assumed-role/SecurityMonkey/secmonkey","accountId":"<ACCOUNT>","accessKeyId":"<ACCESSKEY>","sessionContext":{"attributes":{"mfaAuthenticated":"false","creationDate":"2017-07-01T15:55:39Z"},"sessionIssuer":{"type":"Role","principalId":"<PRINCIPALID>","arn":"arn:aws:iam::<ACCOUNT>:role/SecurityMonkey","accountId":"<ACCOUNT>","userName":"SecurityMonkey"}},"invokedBy":"ec2-frontend-api.amazonaws.com"},"eventTime":"2017-07-01T15:55:40Z","eventSource":"ec2.amazonaws.com","eventName":"DescribeFlowLogs","awsRegion":"ca-central-1","sourceIPAddress":"<SOURCEIP>","userAgent":"ec2-frontend-api.amazonaws.com","requestParameters":null,"responseElements":{"flowLogSet":[]},"requestID":"<REQUESTID>","eventID":"<EVENTID>","eventType":"AwsApiCall","recipientAccountId":"<ACCOUNT>"}

-> These events are correctly mapped with jsonMap and appear in Kibana.

I created the enrichment source extractor_config.json:

{
  "config" : {
    "columns" : {
      "subnet" : 0,
      "AS" : 1,
      "owner" : 2
    },
    "indicator_column" : "subnet",
    "type" : "COMPANY",
    "separator" : ";"
  },
  "extractor" : "CSV"
}

I created the Zookeeper enrichment config enrichment_config.json:

{
     "zkQuorum" : "$ZK:2181"
    ,"sensorToFieldList" : {
          "cttest1" : {
             "type" : "ENRICHMENT"
            ,"fieldToEnrichmentTypes" : {
                 "OurSubnets" : [ "COMPANY" ]
              }
          }
    }
}

I do the following:

${METRON_HOME}/bin/flatfile_loader.sh -n enrichment_config.json -i 
my_subnets.csv -t enrichment -c t -e extractor_config.json
-> OK

echo "scan 'enrichment'" | hbase shell
-> OK

${METRON_HOME}/bin/zk_load_configs.sh -m DUMP -z $ZOOKEEPER
-> Results:

Results:

PARSER Config: cttest1
{
        "parserClassName": "org.apache.metron.parsers.json.JSONMapParser",
        "filterClassName": null,
        "sensorTopic": "cttest1",
        "writerClassName": null,
        "errorWriterClassName": null,
        "invalidWriterClassName": null,
        "parserConfig": {
                "mapStrategy": "UNFOLD"
        },
        "fieldTransformations": []
}

INDEXING Config: cttest1
{
        "hdfs": {
                "batchSize": 1,
                "enabled": true,
                "index": "cttest1"
        },
        "elasticsearch": {
                "batchSize": 1,
                "enabled": true,
                "index": "cttest1"
        },
        "solr": {
                "batchSize": 1,
                "enabled": true,
                "index": "cttest1"
        }
}

ENRICHMENT Config: cttest1
{
  "enrichment" : {
    "fieldMap" : {
      "hbaseEnrichment" : [ "MySubnets" ]
    },
    "fieldToTypeMap" : {
      "MySubnets" : [ "COMPANY" ]
    },
    "config" : { }
  },
  "threatIntel" : {
    "fieldMap" : { },
    "fieldToTypeMap" : { },
    "config" : { },
    "triageConfig" : {
      "riskLevelRules" : [ ],
      "aggregator" : "MAX",
      "aggregationConfig" : { }
    }
  },
  "configuration" : { }
}

Here is where I'm stuck. I can't seem to fetch the enrichments from 
hbase:

test := ENRICHMENT_GET('COMPANY', subnet, 'enrichment', 't')
-> empty
test := ENRICHMENT_GET('MySubnets', subnet, 'enrichment', 't')
-> empty as well

So I don't know how to continue... I know I have to use IN_SUBNET 
somehow as well.

Any idea how I should continue here?