Multiple Email Entity Email Events

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 ("[SourceEmailAddress]", "[DestinationEmailAddress]")
    | 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, EmailSenderAddress)
            by IndicatorId
        // Remove inactive or expired indicators
        | where not(not(Active) or ExpirationDateTime < now())
        // Pick indicators that contain the desired entity type
        | where isnotempty(EmailSenderAddress)
        | extend EmailAddress = tolower(EmailSenderAddress)
        // 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.EmailAddress == $right.BenignProperty
        // Deduplicate indicators by EmailAddress column, equivalent to using join kind=innerunique afterwards
        | summarize hint.strategy=shuffle
            minTimeGenerated = min(minTimeGenerated),
            take_any(*)
            by EmailAddress
        // 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(split(EmailAddress, ".")[-1])
    //    | summarize make_set_if(AuxiliarField, isnotempty(AuxiliarField))
    //);
    //let _IndicatorsPrefilterLength = array_length(_IndicatorsPrefilter);
    let _TableEvents =
        EmailEvents
        | where ingestion_time() between (table_start .. table_end)
        // Filter events that may contain indicators
        | where not(EmailDirection in ("Intra-org"))
        | mv-expand EmailAddress = pack_array(SenderMailFromAddress, iff(SenderFromAddress != SenderMailFromAddress, SenderFromAddress, ""), RecipientEmailAddress) to typeof(string)
        | where isnotempty(EmailAddress)
        //| where not(_IndicatorsPrefilterLength < 10000 and not(EmailAddress has_any (_IndicatorsPrefilter))) // valid TLD ~1500 , "has_any" limit 10000
        | extend EmailAddress = tolower(EmailAddress)
        //| where not(_IndicatorsLength < 1000000 and not(EmailAddress in (toscalar(_Indicators | summarize make_list(EmailAddress))))) // "in" limit 1.000.000
        | project-rename EmailEvents_TimeGenerated = TimeGenerated
    ;
    _Indicators
    | join kind=inner hint.strategy=shuffle _TableEvents on EmailAddress
    // Take only a single event by key columns
    //| summarize hint.strategy=shuffle take_any(*) by EmailAddress, NetworkMessageId
    | project
        EmailEvents_TimeGenerated,
        Description, ActivityGroupNames, IndicatorId, ThreatType, DomainName, Url, ExpirationDateTime, ConfidenceScore, SourceSystem, Tags, AdditionalInformation, EmailSenderAddress,
        SenderFromAddress, SenderFromDomain, SenderMailFromAddress, SenderMailFromDomain, SenderDisplayName, SenderIPv4, SenderIPv6, AuthenticationDetails, RecipientEmailAddress, EmailDirection, Subject, EmailLanguage, UrlCount, AttachmentCount, AdditionalFields, OrgLevelPolicy, OrgLevelAction, UserLevelPolicy, UserLevelAction, EmailActionPolicy, EmailAction, DeliveryAction, DeliveryLocation, ThreatTypes, ConfidenceLevel, DetectionMethods, Connectors, NetworkMessageId, ReportId
};
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(EmailEvents_TimeGenerated, *) by IndicatorId, NetworkMessageId
| extend
    timestamp = EmailEvents_TimeGenerated,
    AccountCustomEntity = EmailSenderAddress

Explanation

This query is designed to identify and analyze email events that match threat indicators from a threat intelligence feed. Here's a simplified breakdown of what the query does:

Setup Parameters:
- It defines several parameters like query_frequency, query_period, query_wait, and table_query_lookback to control the time frames for the query execution and data lookback.
Load Watchlists:
- It retrieves two watchlists: one for benign properties (_TIBenignProperty) and another for excluded sources (_TIExcludedSources). These are used to filter out non-threatening or irrelevant indicators.
Filter Threat Indicators:
- It processes the ThreatIntelligenceIndicator table to get active and non-expired indicators that contain email addresses. It excludes indicators from specific sources and those with benign properties.
Process Email Events:
- It retrieves email events from the EmailEvents table within specified time frames, excluding intra-organizational emails. It extracts and normalizes email addresses for comparison.
Match Indicators with Email Events:
- It performs an inner join between the filtered threat indicators and email events based on email addresses to find matches.
Output:
- The query outputs a summarized list of matched email events and threat indicators, including details like the email's sender, recipient, and threat-related information.
Union of Results:
- It combines results from current email events matched with all available indicators and past email events matched with new indicators since the last query execution.
Final Output:
- The final output is summarized to show the most recent event for each unique combination of threat indicator and email event, along with additional details for further analysis.

In essence, this query helps identify potential threats by correlating email events with known threat indicators, while filtering out benign or irrelevant data.

Details

Jose Sebastián Canós

Released: December 13, 2023

Tables

ThreatIntelligenceIndicatorEmailEvents

Keywords

ThreatIntelligenceIndicatorEmailEvents

Operators

lethas_anyprojecttoscalarwheresummarizemake_listdatetimeagomaterializeminarg_maxisnotemptyextendtolowerjoinkindleftantitake_anybetweenmv-expandpack_arrayiffproject-renameunion

Actions

GitHub

KQL Search