Multiple IP Entity VM Connection

Query

// This query assumes a feed of threat indicators is ingested/synchronized periodically, and each synchronization ingests new indicators and only old indicators that have been modified.
// Active threat indicators in Sentinel are renovated as ThreatIntelligenceIndicator events every ~12 days.
let query_frequency = 1h;
let query_period = 14d;
let query_wait = 0h;
let table_query_lookback = 14d;
let _TIBenignProperty =
    _GetWatchlist('ID-TIBenignProperty')
    | where Notes has_any ("[DestinationIPAddress]")
    | project IndicatorId, BenignProperty
;
let _TIExcludedSources = toscalar(
    _GetWatchlist('Activity-ExpectedSignificantActivity')
    | where Activity == "ThreatIndicatorSource"
    | summarize make_list(Auxiliar)
    );
let _TITableMatch = (table_start: datetime, table_end: datetime, only_new_ti: boolean, ti_start: datetime = datetime(null)) {
    // Scheduled Analytics rules have a query period limit of 14d
    let _Indicators =// materialize(
        ThreatIntelligenceIndicator
        | where TimeGenerated > ago(query_period)
        // Take the earliest TimeGenerated and the latest column info
        | summarize hint.strategy=shuffle
            minTimeGenerated = min(TimeGenerated),
            arg_max(TimeGenerated, Active, Description, ActivityGroupNames, IndicatorId, ThreatType, DomainName, Url, ExpirationDateTime, ConfidenceScore, SourceSystem, Tags, AdditionalInformation, ExternalIndicatorId, NetworkIP, NetworkSourceIP, NetworkDestinationIP, EmailSourceIpAddress)
            by IndicatorId
        // Remove inactive or expired indicators
        | where not(not(Active) or ExpirationDateTime < now())
        // Pick indicators that contain the desired entity type
        | mv-expand IPAddress = pack_array(NetworkIP, NetworkSourceIP, NetworkDestinationIP, EmailSourceIpAddress) to typeof(string)
        | where isnotempty(IPAddress)
        | extend TI_IPAddress = IPAddress
        // Remove indicators from specific sources
        | where not(AdditionalInformation has_any (_TIExcludedSources) or Description has_any (_TIExcludedSources))
        // Remove excluded indicators with benign properties
        | join kind=leftanti _TIBenignProperty on IndicatorId, $left.IPAddress == $right.BenignProperty
        // Deduplicate indicators by IPAddress column, equivalent to using join kind=innerunique afterwards
        | summarize hint.strategy=shuffle
            minTimeGenerated = min(minTimeGenerated),
            take_any(*)
            by IPAddress
        // If we want only new indicators, remove indicators received previously
        | where not(only_new_ti and minTimeGenerated < ti_start)
    //)
    ;
    //let _IndicatorsLength = toscalar(_Indicators | summarize count());
    //let _IndicatorsPrefilter = toscalar(
    //    _Indicators
    //    | extend AuxiliarField = tostring(extract(@"([0-9A-Za-f]+)[\.\:]", 1, IPAddress))
    //    | summarize make_set_if(AuxiliarField, isnotempty(AuxiliarField), 10000)
    //);
    //let _IndicatorsPrefilterLength = array_length(_IndicatorsPrefilter);
    let _TableEvents =
        VMConnection
        | where ingestion_time() between (table_start .. table_end)
        // Filter events that may contain indicators
        | where not(isempty(RemoteIp))
        //| where not(_IndicatorsPrefilterLength < 10000 and not(RemoteIP has_any (_IndicatorsPrefilter))) // "has_any" limit 10000
        | extend IPAddress = trim_start("::ffff:", RemoteIp)
        | where not(isnotempty(parse_ipv4(IPAddress)) and (ipv4_is_private(IPAddress) or ipv4_is_in_any_range(IPAddress, "0.0.0.0/8", "127.0.0.0/8")))
        //| where not(_IndicatorsLength < 1000000 and not(IPAddress in (toscalar(_Indicators | summarize make_list(TI_IPAddress))))) // "in" limit 1.000.000
        | project-rename VMConnection_TimeGenerated = TimeGenerated
    ;
    _Indicators
    | join kind=inner hint.strategy=shuffle _TableEvents on IPAddress
    // Take only a single event by key columns
    //| summarize hint.strategy=shuffle take_any(*) by IPAddress, Computer
    | project
        VMConnection_TimeGenerated,
        Description,
        ActivityGroupNames,
        IndicatorId,
        ThreatType,
        DomainName,
        Url,
        ExpirationDateTime,
        ConfidenceScore,
        SourceSystem,
        Tags,
        AdditionalInformation,
        TI_IPAddress,
        NetworkIP,
        NetworkSourceIP,
        NetworkDestinationIP,
        EmailSourceIpAddress,
        _ResourceId,
        Computer,
        SourceIp,
        DestinationIp,
        Protocol,
        DestinationPort,
        Direction,
        RemoteIp,
        ProcessName,
        RemoteDnsQuestions,
        RemoteDnsCanonicalNames,
        ConnectionId
};
union// isfuzzy=true
    // Match      current table events                                all indicators available
    _TITableMatch(ago(query_frequency + query_wait), ago(query_wait), false),
    // Match      past table events                                                          new indicators since last query execution
    _TITableMatch(ago(table_query_lookback + query_wait), ago(query_frequency + query_wait), true, ago(query_frequency))
| summarize arg_max(VMConnection_TimeGenerated, *) by IndicatorId, Computer
| extend
    timestamp = VMConnection_TimeGenerated,
    HostCustomEntity = Computer,
    IPCustomEntity = TI_IPAddress

Explanation

This query is designed to identify and correlate threat indicators with network events in a security monitoring system, such as Microsoft Sentinel. Here's a simplified breakdown of what the query does:

Setup and Configuration:
- The query defines several parameters, such as the frequency of the query (query_frequency), the period over which data is analyzed (query_period), and a waiting time (query_wait).
- It also sets up a lookback period for querying the table (table_query_lookback).
Data Preparation:
- It retrieves a list of benign properties and excluded sources from specific watchlists to filter out non-threatening indicators.
- It prepares a set of threat indicators from the ThreatIntelligenceIndicator table, filtering out inactive or expired indicators and those from excluded sources.
Event Matching:
- The query looks for network connection events in the VMConnection table that match the threat indicators. It filters out private IP addresses and other non-relevant data.
- It performs two types of matches:
  - Current table events with all available indicators.
  - Past table events with new indicators since the last query execution.
Result Compilation:
- The query combines the results of these matches and summarizes them, ensuring that only the most recent event for each indicator and computer combination is kept.
- It extends the results with additional information, such as the timestamp of the event and custom entities for further analysis.

In essence, this query is designed to continuously monitor network events for any matches with known threat indicators, helping security analysts identify potential threats in their environment.

Details

Jose Sebastián Canós

Released: December 13, 2023

Tables

ThreatIntelligenceIndicatorVMConnection

Keywords

ThreatIntelligenceIndicatorVMConnection

Operators

lethas_anyprojecttoscalarwheresummarizemake_listdatetimematerializeagominarg_maxnotmv-expandisnotemptyextendjoinkindleftantion$left==take_anybybetweenisemptytrim_startparse_ipv4ipv4_is_privateipv4_is_in_any_rangeproject-renameinunionisfuzzy

Actions

GitHub

KQL Search