KQL Search

// 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 ("[SourceIPAddress]")
    | 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 =
        AzureDiagnostics
        | where TimeGenerated between (table_start .. table_end)
        | where ResourceProvider == "MICROSOFT.KEYVAULT"
        // Filter events that may contain indicators
        | where isnotempty(CallerIPAddress)
        //| where not(_IndicatorsPrefilterLength < 10000 and not(CallerIPAddress has_any (_IndicatorsPrefilter))) // "has_any" limit 10000
        | extend IPAddress = trim_start("::ffff:", CallerIPAddress)
        | 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
        | extend
            RequestUri = column_ifexists("requestUri_s", ""),
            Id = column_ifexists("id_s", ""),
            ClientInfo = column_ifexists("clientInfo_s", ""),
            HttpStatusCode = column_ifexists("httpStatusCode_d", ""),
            // Caller Object ID
            identity_claim_oid_g = column_ifexists("identity_claim_oid_g", ""),
            identity_claim_http_schemas_microsoft_com_identity_claims_objectidentifier_g  = column_ifexists("identity_claim_http_schemas_microsoft_com_identity_claims_objectidentifier_g", ""),
            // Caller UPN
            identity_claim_upn_s = column_ifexists("identity_claim_upn_s", ""),
            identity_claim_http_schemas_xmlsoap_org_ws_2005_05_identity_claims_upn_s = column_ifexists("identity_claim_http_schemas_xmlsoap_org_ws_2005_05_identity_claims_upn_s", "")
        | extend
            CallerObjectId = iff(isempty(identity_claim_oid_g), identity_claim_http_schemas_microsoft_com_identity_claims_objectidentifier_g, identity_claim_oid_g),
            CallerObjectUPN = iff(isempty(identity_claim_upn_s), identity_claim_http_schemas_xmlsoap_org_ws_2005_05_identity_claims_upn_s, identity_claim_upn_s)
        | project-rename AzureDiagnostics_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, CallerObjectId, ResourceId
    | project
        AzureDiagnostics_TimeGenerated,
        Description,
        ActivityGroupNames,
        IndicatorId,
        ThreatType,
        DomainName,
        Url,
        ExpirationDateTime,
        ConfidenceScore,
        SourceSystem,
        Tags,
        AdditionalInformation,
        TI_IPAddress,
        NetworkIP,
        NetworkSourceIP,
        NetworkDestinationIP,
        EmailSourceIpAddress,
        Category,
        ResourceId,
        ResourceType,
        OperationName,
        CallerObjectUPN,
        CallerObjectId,
        CallerIPAddress,
        ClientInfo,
        RequestUri,
        HttpStatusCode,
        Id,
        CorrelationId
};
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(AzureDiagnostics_TimeGenerated, *) by IndicatorId, CallerObjectId, ResourceId
| extend
    timestamp = AzureDiagnostics_TimeGenerated,
    AccountCustomEntity = CallerObjectUPN,
    IPCustomEntity = TI_IPAddress