METRON-675: Make Threat Triage rules able to be assigned names and comments

[cestella]

There may be many, many threat triage rules. To help organize these, we should make them slightly more complex than a simple key/value as we have it now. We should add optional name and optional comment fields.

This essentially makes the risk level rules slightly more complex. The format goes from:

"riskLevelRules" : {
  "stellar expression" : numeric score
}

to:

"riskLevelRules" : [
  {
     "name" : "optional name",
     "comment" : "optional comment",
     "rule" : "stellar expression",
     "score" : numeric score
  }
]

This is NOT backwards compatible, but I think it's more explicit and a bit more clear.

Testing plan to come in a follow-on comment.

[cestella]

Testing Instructions beyond the normal smoke test (i.e. letting data
flow through to the indices and checking them).

Preliminaries

It is helpful to install the elasticsearch head plugin:

• /usr/share/elasticsearch/bin/plugin install mobz/elasticsearch-head

Also, set an environment variable to indicate METRON_HOME:

• export METRON_HOME=/usr/metron/0.3.0

Adjust configs to Bro

We will adjust the bro topology to have a couple of threat triage rules:

• Edit $METRON_HOME/config/zookeeper/enrichment/bro.json as follows:

{
  "enrichment" : {
    "fieldMap": {
      "geo": ["ip_dst_addr", "ip_src_addr"],
      "host": ["host"]
    }
  },
  "threatIntel": {
    "fieldMap": {
      "hbaseThreatIntel": ["ip_src_addr", "ip_dst_addr"],
      "stellar" : {
          "config" : {
             "is_alert" : "ip_dst_port == 80 or ip_dst_port == 5353"
                     }
                  }
    },
    "fieldToTypeMap": {
      "ip_src_addr" : ["malicious_ip"],
      "ip_dst_addr" : ["malicious_ip"]
    },
    "triageConfig" : {
       "riskLevelRules" : [
              {
                 "name" : "web",
                 "comment" : "Bump risk if web connection",
                 "rule" : "ip_dst_port == 80",
                 "score" : 10
              },
              {
                 "name" : "dns",
                 "comment" : "Bump risk if dns connection",
                 "rule" : "ip_dst_port == 5353",
                 "score" : 20
              }
                          ],
       "aggregator" : "MAX"
                     }
  }
}

This should create 2 rules:

• set the triage level to 20 if the destination port is a DNS port
• set the triage level to 10 if the destination port is a web port

Ensure via the elasticsearch head plugin that the following is true:

• All bro messages with is_alert == true and ip_dst_port == 5353 have a threat:triage:level of 20
• All bro messages with is_alert == true and threat:triage:level == 20 have a ip_dst_port of 20
• All bro messages with is_alert == true and ip_dst_port == 80 have a threat:triage:level of 10
• All bro messages with is_alert == true and threat:triage:level == 10 have a ip_dst_port of 10

Test Case: Stellar Management Functions

• Upload the management functions via scp metron-platform/metron-management/target/metron-management-0.3.0.jar root@node1:/usr/metron/0.3.0/lib
• Create a file with the following contents named ~/script.stellar

# First we get the squid enrichment config from zookeeper.
# If it is not there, which it is not by default, a suitable default
# config will be specified.
squid_enrichment_config := CONFIG_GET('ENRICHMENT', 'squid')
# We should not have any threat triage rules
THREAT_TRIAGE_PRINT(squid_enrichment_config)
# Just for illustration, we can create a threat alert if the country of the domain registered
# is non-US, then we can make an alert.  To do that, we need to create an is_alert field on the message.
#
# I know that maps get folded into the message, so that whois_info enrichment is going to create a few fields:
#  * domain mapped to whois_info.domain
#  * registrar mapped to whois_info.registrar
#  * home_country mapped to whois_info.home_country
#  * owner mapped to whois_info.owner
whois_info.home_country := 'US'
# Now with this, we can create a rule or two to triage these alerts.
# This means associating a rule as described by a stellar expression that returns true or false with a score
# Also associated with this ruleset is an aggregation function, the default of which is MAX.
# Now we can make a couple rules:
#  * If the message is an alert and from a non-us whois source, we can set the level to 10
#  * If the message is an alert and non-local, we can set the level to 20
#  * If the message is an alert and both non-local and non-us, then we can set the level to 50
# If multiple rules hit, then we should take the max (default behavior)
non_us := whois_info.home_country != 'US'
is_local := IN_SUBNET( if IS_IP(ip_src_addr) then ip_src_addr else NULL, '192.168.0.0/21')
is_both := whois_info.home_country != 'US' && IN_SUBNET( if IS_IP(ip_src_addr) then ip_src_addr else NULL, '192.168.0.0/21')
rules := [ { 'name' : 'is non-us', 'rule' : SHELL_GET_EXPRESSION('non_us'), 'score' : 10 } , { 'name' : 'is local', 'rule' : SHELL_GET_EXPRESSION('is_local'), 'score' : 20 } , { 'name' : 'both non-us and local', 'comment' : 'union of both rules.',  'rule' : SHELL_GET_EXPRESSION('is_both'), 'score' : 50 } ]
# Now that we have our rules staged, we can add them to our config.
squid_enrichment_config_new := THREAT_TRIAGE_ADD( squid_enrichment_config_new, rules )
# Pretty Print the rules
THREAT_TRIAGE_PRINT(squid_enrichment_config_new)
# Now just print the raw config to make sure it jives
squid_enrichment_config_new
# Now that we have admired it, we can remove the rules
squid_enrichment_config_new := THREAT_TRIAGE_REMOVE( squid_enrichment_config_new, [ SHELL_GET_EXPRESSION('non_us') , SHELL_GET_EXPRESSION('is_local') , SHELL_GET_EXPRESSION('is_both') ] )
THREAT_TRIAGE_PRINT(squid_enrichment_config_new)

• Execute the script via cat script.stellar | /usr/metron/0.3.0/bin/stellar -z node1 -na You should see the following output:

Stellar, Go!
Please note that functions are loading lazily in the background and will be unavailable until loaded fully.
{es.clustername=metron, es.ip=node1, es.port=9300, es.date.format=yyyy.MM.dd.HH}
[Stellar]>>> # First we get the squid enrichment config from zookeeper.
[Stellar]>>> # If it is not there, which it is not by default, a suitable default
[Stellar]>>> # config will be specified.
[Stellar]>>> squid_enrichment_config := CONFIG_GET('ENRICHMENT', 'squid')
Functions loaded, you may refer to functions now...
[Stellar]>>> # We should not have any threat triage rules
[Stellar]>>> THREAT_TRIAGE_PRINT(squid_enrichment_config)
╔══════╤═════════╤═════════════╤═══════╗
║ Name │ Comment │ Triage Rule │ Score ║
╠══════╧═════════╧═════════════╧═══════╣
║ (empty)                              ║
╚══════════════════════════════════════╝

[Stellar]>>> # Just for illustration, we can create a threat alert if the country of the domain registered
[Stellar]>>> # is non-US, then we can make an alert.  To do that, we need to create an is_alert field on the message.
[Stellar]>>> #
[Stellar]>>> # I know that maps get folded into the message, so that whois_info enrichment is going to create a few fields:
[Stellar]>>> #  * domain mapped to whois_info.domain
[Stellar]>>> #  * registrar mapped to whois_info.registrar
[Stellar]>>> #  * home_country mapped to whois_info.home_country
[Stellar]>>> #  * owner mapped to whois_info.owner
[Stellar]>>> whois_info.home_country := 'US'
[Stellar]>>> # Now with this, we can create a rule or two to triage these alerts.
[Stellar]>>> # This means associating a rule as described by a stellar expression that returns true or false with a score
[Stellar]>>> # Also associated with this ruleset is an aggregation function, the default of which is MAX.
[Stellar]>>> # Now we can make a couple rules:
[Stellar]>>> #  * If the message is an alert and from a non-us whois source, we can set the level to 10
[Stellar]>>> #  * If the message is an alert and non-local, we can set the level to 20
[Stellar]>>> #  * If the message is an alert and both non-local and non-us, then we can set the level to 50
[Stellar]>>> # If multiple rules hit, then we should take the max (default behavior)
[Stellar]>>> non_us := whois_info.home_country != 'US'
[Stellar]>>> is_local := IN_SUBNET( if IS_IP(ip_src_addr) then ip_src_addr else NULL, '192.168.0.0/21')
[Stellar]>>> is_both := whois_info.home_country != 'US' && IN_SUBNET( if IS_IP(ip_src_addr) then ip_src_addr else NULL, '192.168.0.0/21')
[Stellar]>>> rules := [ { 'name' : 'is non-us', 'rule' : SHELL_GET_EXPRESSION('non_us'), 'score' : 10 } , { 'name' : 'is local', 'rule' : SHELL_GET_EXPRESSION('is_local '), 'score' : 20 } , { 'name' : 'both non-us and local', 'comment' : 'union of both rules.',  'rule' : SHELL_GET_EXPRESSION('is_both'), 'score' : 50 } ]
[Stellar]>>> # Now that we have our rules staged, we can add them to our config.
[Stellar]>>> squid_enrichment_config_new := THREAT_TRIAGE_ADD( squid_enrichment_config_new, rules )
[Stellar]>>> # Pretty Print the rules
[Stellar]>>> THREAT_TRIAGE_PRINT(squid_enrichment_config_new)
╔═══════════════════════╤══════════════════════╤═══════════════════════════════════════════════════════════════════════════════════════════════════════════════════╤═══════╗
║ Name                  │ Comment              │ Triage Rule                                                                                                       │ Score ║
╠═══════════════════════╪══════════════════════╪═══════════════════════════════════════════════════════════════════════════════════════════════════════════════════╪═══════╣
║ is non-us             │                      │ whois_info.home_country != 'US'                                                                                   │ 10    ║
╟───────────────────────┼──────────────────────┼───────────────────────────────────────────────────────────────────────────────────────────────────────────────────┼───────╢
║ is local              │                      │ IN_SUBNET( if IS_IP(ip_src_addr) then ip_src_addr else NULL, '192.168.0.0/21')                                    │ 20    ║
╟───────────────────────┼──────────────────────┼───────────────────────────────────────────────────────────────────────────────────────────────────────────────────┼───────╢
║ both non-us and local │ union of both rules. │ whois_info.home_country != 'US' && IN_SUBNET( if IS_IP(ip_src_addr) then ip_src_addr else NULL, '192.168.0.0/21') │ 50    ║
╚═══════════════════════╧══════════════════════╧═══════════════════════════════════════════════════════════════════════════════════════════════════════════════════╧═══════╝


Aggregation: MAX
[Stellar]>>> # Now just print the raw config to make sure it jives
[Stellar]>>> squid_enrichment_config_new
{
  "enrichment" : {
    "fieldMap" : { },
    "fieldToTypeMap" : { },
    "config" : { }
  },
  "threatIntel" : {
    "fieldMap" : { },
    "fieldToTypeMap" : { },
    "config" : { },
    "triageConfig" : {
      "riskLevelRules" : [ {
        "name" : "is non-us",
        "rule" : "whois_info.home_country != 'US'",
        "score" : 10.0
      }, {
        "name" : "is local",
        "rule" : "IN_SUBNET( if IS_IP(ip_src_addr) then ip_src_addr else NULL, '192.168.0.0/21')",
        "score" : 20.0
      }, {
        "name" : "both non-us and local",
        "comment" : "union of both rules.",
        "rule" : "whois_info.home_country != 'US' && IN_SUBNET( if IS_IP(ip_src_addr) then ip_src_addr else NULL, '192.168.0.0/21')",
        "score" : 50.0
      } ],
      "aggregator" : "MAX",
      "aggregationConfig" : { }
    }
  },
  "configuration" : { }
}
[Stellar]>>> # Now that we have admired it, we can remove the rules
[Stellar]>>> squid_enrichment_config_new := THREAT_TRIAGE_REMOVE( squid_enrichment_config_new, [ SHELL_GET_EXPRESSION('non_us') , SHELL_GET_EXPRESSION('is_local') , SHE LL_GET_EXPRESSION('is_both') ] )
[Stellar]>>> THREAT_TRIAGE_PRINT(squid_enrichment_config_new)
╔══════╤═════════╤═════════════╤═══════╗
║ Name │ Comment │ Triage Rule │ Score ║
╠══════╧═════════╧═════════════╧═══════╣
║ (empty)                              ║
╚══════════════════════════════════════╝

[nickwallen]

+1 Works great. Spun everything up, followed your script, created my own triage rules and validated the scoring.

The 'RiskLevelRule' POJO certainly makes things a little cleaner.

As a random side note, will be really cool when the aggregation of the scores is just Stellar code, rather than MAX or SUM. This would allow us to plug-in a real model for scoring the alerts.

[cestella]

@nickwallen Agreed on the aggregator being stellar. I created https://issues.apache.org/jira/browse/METRON-683 for it