In the previous section, we walked through how to load threat intel data into Metron and then apply those threat intels in realtime as telemetry events are streamed through the platform.
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 thethreat.triage.level
. Let's dig a bit deeper into this and provide an example.
Metron Query Language Explained
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 Explained
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 configurationheld 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.
Step 1: Setup and Pre-requisites
- You should have completed the instructions in Adding a new Telemetry Data Source
- Make sure the following variables are configured based on your environment:
KAFKA_HOST = host where a Kafka broker is installed
ZOOKEEPER_HOST = host where a Zookeeper server is installed
PROBE_HOST = Host where your sensor, probes are installed. If don't have any sensors installed, pick the host where a storm supervisor is running
SQUID_HOST = Host where you want to install SQUID. If you don't care, just install on the PROBE_HOST
NIFI_HOST = The host where you will install NIFI. You want this this to be same host that you installed Squid.
HOST_WITH_ENRICHMENT_TAG = This is the host in your inventory hosts file that you put under the group "enrichment"
SEARCH_HOST = This is the host where you have elastic or solr running. This is the host in your inventory hosts file that you put under the group "search". Pick one of the search hosts
SEARCH_HOST_PORT = The port of the search host where indexing is configured. (e.g: 9300)
METRON_UI_HOST = This is the host where your metron ui web application is running. This is the host in your inventory hosts file that you put under the group "web".
METRON_VERSION = The release of the metron binaries you are working with (e.g: 0.2.0BETA-RC2)
Step 2: Create the Threat Triage Rule Configuration
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, let's triage the threat alerts for the squid
sensor data higher under the following conditions:
- Rule 1: If the threat intel enrichment type
zeusList
as defined in part 4 is alerted, then we want to consider that an alert of score of 5 - Rule 2: If the
url
is neither a.com
nor a.net
, then we want to consider that alert a score of 10
For each message we will assign the maximum score across all conditions as the triage score. This translates into the following configuration:
Step 3: Upload the threat triage configuration to Zookeeper
In order to apply this triage configuration, we must modify the configuration for the squid
sensor in the enrichment topology. To do this, we should modify /usr/metron/0.1BETA/config/zookeeper/sensors/squid.json
on node1
However, since the configuration in zookeeper may have be out of sync with the configuration on disk, we must make sure they are in sync by executing the following command:
/usr/metron/$METRON_RELEASE/bin/zk_load_configs.sh -m PULL -z $ZOOKEEPER_HOST:2181 -f -o /usr/metron/$METRON_RELEASE/config/zookeeper
We should ensure that the configuration for squid exists by checking out
TODO: the directory sensors is wrong. It shoudl be changed to enrichments. Also change field url to domain_without_subdomains
cat /usr/metron/$METRON_RELEASE/config/zookeeper/sensors/squid.json
Now we can edit the configuration. In /usr/metron/$METRON_RELEASE/config/zookeeper/sensors/squid.json
edit the section titled riskLevelRules
and add the two rules above to the map:
"exists(threatintels.hbaseThreatIntel.url.zeusList)" : 5
"not(ENDS_WITH(url, '.com') or ENDS_WITH(url, '.net'))" : 10
Also, ensure that the aggregator
field indicates MAX
After modifying the configuration, we can push the configuration back to zookeeper and have the enrichment topology pick it up with live data via
/usr/metron/$METRON_RELEASE/bin/zk_load_configs.sh -m PUSH -z $ZOOKEEPER_HOST:2181 -i /usr/metron/$METRON_RELEASE/config/zookeeper
Now, if we reload the data from the part 4 via
tail /var/log/squid/access.log | /usr/hdp/current/kafka-broker/bin/kafka-console-producer.sh --broker-list $KAFKA_HOST:6667 --topic squid
Now, if we check the squid
index using the elasticsearch head plugin, we can see the threats triage as we would expect:
Step 4: View Triaged/Scored Alerts
View Non Threat Data
For URL's from cnn.com, we see no threat alert, so no triage level is set. Notice the lack of a threat.triage.level
field.
Threat Data from alamman.com has a triage level of 5
Because alamman.com is a malicious host from the zeusList
threat intel feed but is a .com
address, it's assigned threat.triage.level
of 5.
Threat Data from atmape.ru has a triage level of 10
Because atmape.ru is both a malicious host from the zeusList
threat intel feed as well as a non .com
and non .net
address, it's assigned threat.triage.level
of 10.