In part 4, you learned how we can attach threat intelligence indicators to the messages that are passing through the enrichment Storm topology. The problem, however, is that not all threat intelligence indicators are made equal. Some require immediate response, whereas others can be dealt with or investigated as time and availability permits. What we need is the ability to triage and rank threats by severity.
Now that we know what we should do, the next question is how to accomplish it; in other words, we must define what exactly we mean when we say "severity." The capability as implemented in Metron is accomplished by providing the ability to associate possibly complex conditions to numeric scores. Then, for each message, the set of conditions are evaluated and the set of numbers for matching conditions are aggregated via a configurable aggregation function. This aggregated score is added to the message in the threat.triage.level
. Let's dig a bit deeper into this and provide an example.
Metron Query Language
The heart of the problem is how one defines a "condition." In Metron, we provide a custom domain specific language for defining conditions.
The query language supports the following:
- Referencing fields in the enriched JSON
- Simple boolean operations: and, not, or
- Determining whether a field exists (via
exists
) - The ability to have parenthesis to make order of operations explicit
- A fixed set of functions which take strings and return boolean. Currently:
IN_SUBNET(ip, cidr1, cidr2, ...)
IS_EMPTY(str)
STARTS_WITH(str, prefix)
ENDS_WITH(str, suffix)
REGEXP_MATCH(str, pattern)
- A fixed set of string to string transformation functions. Currently:
TO_LOWER
TO_UPPER
TRIM
Consider, for example, the following JSON message:
{...
"src_ip_addr" : "192.168.0.1","is_local" : true
...
}
Consider the query:
IN_SUBNET( src_ip_addr, '192.168.0.0/24') or src_ip_addr in [ '10.0.0.1', '10.0.0.2' ] or exists(is_local)
This evaluates to true precisely when one of the following is true for a message:
- The value of the
src_ip_addr
field is in the192.168.0.0/24
subnet - The value of the
src_ip_addr
field is10.0.0.1
or10.0.0.2
- The field
is_local
exists
Threat Triage Configuration
Now that we have the ability to define conditions, for each sensor we need to associate these conditions to scores. Since this is a per-sensor configuration, this fits nicely within the sensor enrichment configuration held in zookeeper. This configuration fits well within the threatIntel
section of the configuration like so:
{...,"threatIntel" : {..., "triageConfig" : {"riskLevelRules" : {"condition1" : level1, "condition2" : level2...},"aggregator" : "MAX"}}}
riskLevelRules
correspond to the set of condition to numeric level mappings that define the threat triage for this particular sensor. aggregator
is an aggregation function that takes all non-zero scores representing the matching queries from riskLevelRules
and aggregates them into a single score. The current supported aggregation functions are
MAX
: The max of all of the associated values for matching queriesMIN
: The min of all of the associated values for matching queriesMEAN
: The mean of all of the associated values for matching queriesPOSITIVE_MEAN
: The mean of the positive associated values for the matching queries.
Example
So, where we left off in part 4 was a working threat intelligence enrichment. Now, let's see if we can triage those threats for the squid data flowing through. In particular,