KQL Search

// This query can help you to detect slow password spray events.
// If you want to check all the activity (not the new activity) over a time period, make sure query_frequency and query_period parameters are the same value.
//
// Click "Save as function", in Parameters write in the fields:
// "timespan" "query_frequency" "14d"
// "timespan" "query_period"    "14d"
//
// If you name the function "SlowPasswordSpray", you can check the function with queries like the following:
//
// SlowPasswordSpray()
//
// SlowPasswordSpray(1h, 14d)
//
// SlowPasswordSpray(14d, 14d)
//
// let query_frequency = 14d;
// let query_period = 14d;
//let Function = (query_frequency:timespan = 14d, query_period:timespan = 14d){
let ipv4_prefix_mask = 23;
let ipv6_prefix_mask = 48;
let session_period_limit = 7d;
let group_by_time_period = 30m;
let excluded_success_resulttypes = dynamic(["700082"]);
let _ExpectedIPAddresses = toscalar(
    union _GetWatchlist("IP-CorporateCollaborators"), _GetWatchlist("IP-Vendors")
    | summarize make_list(IPAddress)
);
let _ExpectedLocations = toscalar(
    _GetWatchlist("Activity-ExpectedSignificantActivity")
    | where Activity == "CorporateGeolocation"
    | summarize make_list(Auxiliar)
);
let _ExpectedASNs = toscalar(
    _GetWatchlist("Activity-ExpectedSignificantActivity")
    | where Activity == "CommonUserASN"
    | summarize make_list(Auxiliar)
);
let _HomeTenantId = toscalar(
    _GetWatchlist("UUID-EntraIdTenantIds")
    | where Notes has "[HomeTenant]"
    | summarize make_list(TenantId)
);
let _PartialSuccessResultTypes = toscalar(
    _GetWatchlist("ResultType-SignInLogsErrorCodes")
    | where Notes has_any ("[Success]", "[Expired]") //and not(ResultType in (excluded_success_resulttypes))
    | summarize make_list(ResultType)
);
let _ExcludedResultTypes = toscalar(
    _GetWatchlist("ResultType-SignInLogsErrorCodes")
    | where not(Notes has_any ("[Success]", "[Failure]")) and Notes has "[Interrupt]"
    | summarize make_list(ResultType)
);
// Query authentication events
let _SigninEvents =
    union isfuzzy=true SigninLogs, AADNonInteractiveUserSignInLogs, ADFSSignInLogs
    | where TimeGenerated > ago(query_period)
;
let _FilteredSigninEvents =
    _SigninEvents
    | where IsInteractive and ResourceTenantId in (_HomeTenantId)
    | where not(IPAddress == "127.0.0.1" or (isnotempty(parse_ipv4(IPAddress)) and ipv4_is_private(IPAddress)) or (isnotempty(parse_ipv4(tostring(split(IPAddress, ":")[0]))) and ipv4_is_private(tostring(split(IPAddress, ":")[0]))))
    // Remove expected IP addresses
    //| where not(NetworkLocationDetails has '"networkType":"trustedNamedLocation"')
    | where not((isnotempty(parse_ipv4(IPAddress)) and ipv4_is_in_any_range(IPAddress, _ExpectedIPAddresses)) 
        or (isempty(parse_ipv4(IPAddress)) and isnotempty(parse_ipv6(IPAddress)) and ipv6_is_in_any_range(IPAddress, _ExpectedIPAddresses)))
    | where not(UserType == "Guest" or UserId == "00000000-0000-0000-0000-000000000000" or CorrelationId == "00000000-0000-0000-0000-000000000000")
    | where not(ResultType in (_ExcludedResultTypes))
    // Distinguish failed and (partially) successful authentications
    | extend PartialSuccessResultType = ResultType in (_PartialSuccessResultTypes)
;
let _GetSprayStages = (start_time:datetime, end_time:datetime){
    // Query events where distinct accounts were observed from the same address range or CorrelationId
    let _SprayEvents = materialize(
        union
        (
            _FilteredSigninEvents
            | where TimeGenerated between(start_time .. end_time) and isnotempty(IPAddress)
            // Compute the first time an account had activity from a specific IP address
            | summarize hint.shufflekey=IPAddress
                StartTime = min(TimeGenerated),
                Location = take_any(Location)
                by UserId, IPAddress, PartialSuccessResultType
            | extend IPRange = iff(isnotempty(parse_ipv4(IPAddress)), parse_ipv6_mask(IPAddress, 128 - (32 - ipv4_prefix_mask)), parse_ipv6_mask(IPAddress, ipv6_prefix_mask))
            // Group the first times by address range
            | summarize hint.shufflekey=IPRange
                minTimeGeneratedList = make_list(tostring(pack_array(StartTime, UserId))),
                IPAddresses = make_set(IPAddress, tolong(min_of(pow(2, (32 - ipv4_prefix_mask)), 10000))),
                DistinctAccountCount = dcount(UserId),
                Location = take_any(Location)
                by IPRange, PartialSuccessResultType
        ),
        (
            _FilteredSigninEvents
            | where TimeGenerated between(start_time .. end_time) and isnotempty(CorrelationId)
            // Compute the first time an account had activity from a specific IP address
            | summarize hint.shufflekey=CorrelationId
                StartTime = min(TimeGenerated),
                Location = take_any(Location)
                by UserId, CorrelationId, IPAddress, PartialSuccessResultType
            // Group the first times by address range
            | summarize hint.shufflekey=CorrelationId
                minTimeGeneratedList = make_list(tostring(pack_array(StartTime, UserId))),
                IPAddresses = make_set(IPAddress, 10000),
                DistinctAccountCount = dcount(UserId),
                Location = take_any(Location)
                by CorrelationId, PartialSuccessResultType
        )
        // Remove ranges with failures and less than 2 accounts or 2 events
        | where not(not(PartialSuccessResultType) and (DistinctAccountCount < 2 or array_length(minTimeGeneratedList) < 2))
        // Compute time periods between different accounts
        | mv-apply minTimeGeneratedElement = minTimeGeneratedList on (
            extend minTimeGeneratedElement = todynamic(dynamic_to_json(minTimeGeneratedElement))
            | extend
                TimeGenerated = todatetime(minTimeGeneratedElement[0]),
                UserId = tostring(minTimeGeneratedElement[1])
            | sort by TimeGenerated asc
            // Remove cases where the previous event was the same account (assumedly from another IP address)
            | where not(isnotempty(prev(UserId)) and UserId == prev(UserId))
            // Compute how much time passed until the next event
            | extend NextEventTimeDiff = next(TimeGenerated) - TimeGenerated
            | project-away minTimeGeneratedElement, TimeGenerated, UserId
            // Keep events where the next event happened within x time
            | where isnotempty(NextEventTimeDiff) and NextEventTimeDiff between (time(0s)..session_period_limit)
            // Count how many events happened with a specific frequency (e.g. from 0 to 30m, 30m to 1h, 1h to 1h30m)
            | summarize
                AdditionalEvents = count(),
                take_any(PartialSuccessResultType),
                Location = take_any(Location)
                by Frequency = bin(NextEventTimeDiff, group_by_time_period)
            // Remove frequencies of successful authentications that have lesser activity that 2 additional events per 5 hours
            // (thus if long periods between events, take only cases with many events)
            | where not(PartialSuccessResultType and AdditionalEvents < 2*(Frequency / 5h))
            // Remove frequencies of failed authentications that have lesser activity that 2 additional events per 5 hours from expected locations
            // (thus if long periods between events, take only cases with many events)
            | where not(not(PartialSuccessResultType) and AdditionalEvents < 2*(Frequency / 5h) and Location in (_ExpectedLocations))
            | summarize
                SprayCount = 1 + sum(AdditionalEvents),
                SprayFrequencies = make_bag(pack(tostring(Frequency), AdditionalEvents)),
                FrequenciesList = make_list(AdditionalEvents),
                Location = take_any(Location)
        )
        | project-away minTimeGeneratedList
        // Compute the spray consistency based on Shannon entropy
        | extend FrequenciesList = array_concat(FrequenciesList, pack_array(1)) // Add noise
        | mv-apply Frequency = FrequenciesList to typeof(int) on (
            extend AuxProb = Frequency / toreal(SprayCount)
            | summarize SprayConsistency = -sum(AuxProb*log2(AuxProb))
        )
        | project-away FrequenciesList
        // Underestimate spray consistency of grouped events with less than 5 events
        | extend SprayConsistency = SprayConsistency + iff(SprayCount <= 5, (5-SprayCount) / toreal(5), 0.0)
        // Fix spray values where there wasn't spray activity
        | extend
            SprayCount = iff(SprayCount == 1 and array_length(bag_keys(SprayFrequencies)) == 0, int(null), SprayCount),
            SprayConsistency = iff(SprayCount == 1 and array_length(bag_keys(SprayFrequencies)) == 0, real(null), SprayConsistency)
        // Remove ranges with failures but without spray activity
        | where not(not(PartialSuccessResultType) and isempty(SprayCount))
        // Pack the useful information
        | extend Properties = pack(
            "PartialSuccessResultType", PartialSuccessResultType,
            "DistinctAccountCount", DistinctAccountCount,
            "DistinctAddressCount", array_length(IPAddresses),
            "SprayCount", SprayCount,
            "SprayConsistency", SprayConsistency,
            "SprayFrequencies", SprayFrequencies,
            "IPAddresses", IPAddresses
            )
        | project PartialSuccessResultType, IPRange, CorrelationId, IPAddresses, Location, Properties, Source = coalesce(IPRange, CorrelationId)
    );
    // Join together failed and successful activity from the same "source"
    let _JoinedEvents =
        _SprayEvents
        | where PartialSuccessResultType
        | project-rename Success_IPAddresses = IPAddresses
        | lookup kind=inner (
            _SprayEvents
            | where not(PartialSuccessResultType)
            | project-rename Failure_IPAddresses = IPAddresses
        ) on Source
        // Join successful and failure events ONLY if they share the same CorrelationId OR set of addresses
        // Please, create another detection that will check successful authentications from password spray ranges
        | where isnotempty(CorrelationId) or not(array_length(set_intersect(Success_IPAddresses, Failure_IPAddresses)) == 0 and array_length(Success_IPAddresses) < (array_length(Failure_IPAddresses) * 3))
        | project Source, Joined = true
    ;
    _SprayEvents
    | lookup kind=leftouter _JoinedEvents on Source
    | extend Joined = coalesce(Joined, PartialSuccessResultType)
    | summarize
        take_any(IPRange, CorrelationId),
        IPAddresses = make_set(IPAddresses, 10000),
        Activity = make_bag(pack(iff(PartialSuccessResultType, "SuccessEvents", "FailureEvents"), Properties)),
        Location = take_any(Location)
        by Source, Joined
    | project-away Joined
    // Tag the potential stage of an attack
    | extend Stage = case(
        isnotempty(Activity["FailureEvents"]) and isnotempty(Activity["SuccessEvents"]), "Password Spray + Initial Access",
        isnotempty(Activity["FailureEvents"]) and isempty(Activity["SuccessEvents"]), "Password Spray",
        isempty(Activity["FailureEvents"]) and isnotempty(Activity["SuccessEvents"]), "Initial Access",
        ""
        )
    // Remove ranges without failures nor spray activity
    | where not(
        Stage == "Initial Access"
        and isempty(Activity["SuccessEvents"]["SprayCount"])
        )
    // Remove ranges, with failures and successes, with low activity
    | where not(
        Stage == "Password Spray + Initial Access"
        and toint(Activity["SuccessEvents"]["SprayCount"]) < 3
        and toint(Activity["FailureEvents"]["SprayCount"]) < 3
        )
    // Remove ranges where the failures don't cover extensively the successes
    | extend Auxiliar_FailureOverSuccessCoverage =
        100 *
        (toreal(Activity["FailureEvents"]["SprayCount"]) / toreal(Activity["SuccessEvents"]["DistinctAccountCount"])) *
        (toreal(Activity["FailureEvents"]["SprayCount"]) / toreal(Activity["FailureEvents"]["DistinctAccountCount"])) *
        (toreal(Activity["FailureEvents"]["DistinctAddressCount"]) / toreal(Activity["SuccessEvents"]["DistinctAddressCount"])) *
        iff(isnotempty(Activity["SuccessEvents"]["SprayConsistency"]), (toreal(Activity["SuccessEvents"]["SprayConsistency"]) / toreal(Activity["FailureEvents"]["SprayConsistency"])), 1.0)
    | where not(
        Stage == "Password Spray + Initial Access"
        and isnotempty(Auxiliar_FailureOverSuccessCoverage) and Auxiliar_FailureOverSuccessCoverage < 1
        )
    | where not(
        Stage == "Password Spray + Initial Access"
        and isnotempty(Auxiliar_FailureOverSuccessCoverage) and Auxiliar_FailureOverSuccessCoverage < 10 and Location in (_ExpectedLocations)
        )
    | project-away Auxiliar_FailureOverSuccessCoverage, Location
};
// Get the events relevant to query_frequency
let _SprayStages = materialize(
    _GetSprayStages(ago(query_period), now())
    | lookup kind=leftouter (
        _GetSprayStages(ago(query_period), ago(query_frequency))
        | project Source, Stage, Recurrent = true
    ) on Source, Stage
    | extend Recurrent = coalesce(Recurrent, false)
);
// Get the list of spraying IP addresses
let _SprayingAddresses = toscalar(
    _SprayStages
    | where isnotempty(IPRange)
    | summarize make_set(IPAddresses)
);
// Get the list of spraying CorrelationIds
let _SprayingCorrelationIds = toscalar(
    _SprayStages
    | where isnotempty(CorrelationId)
    | summarize make_set(CorrelationId)
);
// Add information about start time, location, ASNs, result types, devices...
_SprayStages
| lookup kind=leftouter (
    union
    (
        _SigninEvents
        | where IPAddress in (_SprayingAddresses)
        | extend Auxiliar = IPAddress
    ),
    (
        _SigninEvents
        | where CorrelationId in (_SprayingCorrelationIds)
        | extend Auxiliar = CorrelationId
    )
    //| where not(isempty(DeviceDetail_string) and isempty(DeviceDetail_dynamic) and isempty(UserAgent))
    | project
        Auxiliar,
        TimeGenerated,
        CorrelationId,
        IPAddress,
        Location,
        AutonomousSystemNumber,
        UserDisplayName,
        UserId,
        ResultType,
        UserAgent,
        DeviceDetail = iff(isnotempty(DeviceDetail_string), DeviceDetail_string, tostring(DeviceDetail_dynamic)),
        SampleBy = strcat(IPAddress, "|", UserAgent)
    // Take 1 sample event per "key"
    | summarize hint.shufflekey=SampleBy
        take_any(Location),
        take_any(*),
        StartTime = min(TimeGenerated),
        EndTime = max(TimeGenerated),
        UserDisplayNames = make_set_if(UserDisplayName, not(UserDisplayName == UserId), 100)
        by SampleBy, Auxiliar
    | project-away SampleBy, UserId, *1
    | extend
        IPRange = iff(Auxiliar == IPAddress, iff(isnotempty(parse_ipv4(IPAddress)), parse_ipv6_mask(IPAddress, 128 - (32 - ipv4_prefix_mask)), parse_ipv6_mask(IPAddress, ipv6_prefix_mask)), ""),
        CorrelationId = iff(Auxiliar == CorrelationId, CorrelationId, "")
    | extend
        Source = coalesce(IPRange, CorrelationId),
        DeviceDetail = todynamic(DeviceDetail),
        ParsedUserAgent = parse_user_agent(UserAgent, dynamic(["os", "browser"]))
    | extend
        DeviceId = tostring(DeviceDetail["deviceId"]),
        DeviceName = tostring(DeviceDetail["displayName"]),
        DeviceIsManaged = tostring(DeviceDetail["isManaged"]),
        DeviceTrustType = tostring(DeviceDetail["trustType"]),
        DeviceDetailOS = tostring(DeviceDetail["operatingSystem"]),
        UserAgentOS = tostring(ParsedUserAgent["OperatingSystem"]["Family"]),
        Browser = tostring(ParsedUserAgent["Browser"]["Family"])
    | extend
        OperatingSystem = case(
            isempty(DeviceDetailOS), UserAgentOS,
            isempty(UserAgent), extract(@"^([A-Za-z]+)", 1, DeviceDetailOS),
            UserAgentOS == "Other", extract(@"^([A-Za-z]+)", 1, DeviceDetailOS),
            UserAgentOS
        ),
        Browser = case(
            Browser == "Other", iff(UserAgent != "-", extract(@"^([^\/\s]+)", 1, UserAgent), ""),
            Browser
        )
    | extend
        OperatingSystem = case(
            OperatingSystem has "ios", "iOS",
            OperatingSystem has_any ("mac", "macos"), "macOS",
            OperatingSystem == "Ubuntu", "Linux",
            Browser == "Samsung Internet", "Android",
            Browser == "MacOutlook", "macOS",
            OperatingSystem
        )
    | project Source, StartTime, EndTime, UserDisplayNames, ResultType, Location, AutonomousSystemNumber, DeviceId, DeviceName, DeviceIsManaged, DeviceTrustType, Browser, OperatingSystem
    | as hint.materialized=true _SampleEvents
    | summarize
        StartTime = min(StartTime),
        EndTime = max(EndTime),
        UserDisplayNamesSample = make_set(UserDisplayNames, 100),
        ResultTypesSample = make_set(ResultType, 100),
        Locations = make_set_if(Location, isnotempty(Location), 100),
        AutonomousSystemNumbers = make_set_if(AutonomousSystemNumber, isnotempty(AutonomousSystemNumber), 100),
        //DeviceIdsSample = make_set_if(DeviceId, isnotempty(DeviceId), 100),
        DeviceNamesSample = make_set_if(DeviceName, isnotempty(DeviceName), 100),
        DeviceIsManagedSample = make_set_if(DeviceIsManaged, isnotempty(DeviceIsManaged), 100),
        DeviceTrustTypesSample = make_set_if(DeviceTrustType, isnotempty(DeviceTrustType), 100),
        Browsers = make_set_if(Browser, isnotempty(Browser), 100)
        by Source
    | lookup kind=leftouter (
        _SampleEvents
        | summarize OSCount = count() by OperatingSystem, Source
        | summarize
            OperatingSystems = make_bag(pack(OperatingSystem, OSCount)),
            OperatingSystemsCount = sum(OSCount)
            by Source
        ) on Source
    | extend
        Location = case(
            array_length(Locations) == 1, tostring(Locations[0]),
            ""
        ),
        ASN = case(
            array_length(AutonomousSystemNumbers) == 1, tostring(AutonomousSystemNumbers[0]),
            ""
        )
) on Source
// Remove ranges without failures that might be mobile ISP ranges
| where not(
    Stage == "Initial Access"
    and Location in (_ExpectedLocations)
    and ASN in (_ExpectedASNs)
    and (toint(Activity["SuccessEvents"]["DistinctAddressCount"]) > 10 or toint(Activity["SuccessEvents"]["DistinctAccountCount"]) > 10)
    and (toint(Activity["SuccessEvents"]["DistinctAddressCount"]) / toreal(Activity["SuccessEvents"]["DistinctAccountCount"])) >= 0.95
    and (toint(OperatingSystems["iOS"]) + toint(OperatingSystems["Android"])) >= 0.95*OperatingSystemsCount
    )
// Remove ranges without failures that might be common user ISP ranges
| where not(
    Stage == "Initial Access"
    and Location in (_ExpectedLocations)
    and ASN in (_ExpectedASNs)
    and (toint(Activity["SuccessEvents"]["DistinctAccountCount"]) - toint(Activity["SuccessEvents"]["DistinctAddressCount"])) <= 2
    and (toint(Activity["SuccessEvents"]["SprayCount"]) - toint(Activity["SuccessEvents"]["DistinctAddressCount"])) <= 2
    and (toint(OperatingSystems["iOS"]) >= 2 or toint(OperatingSystems["Android"]) >= 2)
    )
| where not(
    Stage == "Initial Access"
    and Location in (_ExpectedLocations)
    and ASN in (_ExpectedASNs)
    and (toint(Activity["SuccessEvents"]["SprayCount"]) <= 2
        or (toint(Activity["SuccessEvents"]["SprayCount"]) / toreal(Activity["SuccessEvents"]["SprayConsistency"]) / toreal(Activity["SuccessEvents"]["DistinctAccountCount"])) < 0.2)
    and (toint(OperatingSystems["iOS"]) >= 2 or toint(OperatingSystems["Android"]) >= 2)
    )
// Remove ranges without failures from expected locations with few accounts per address
| where not(
    Stage == "Initial Access"
    and Location in (_ExpectedLocations)
    and (toint(Activity["SuccessEvents"]["DistinctAccountCount"]) - toint(Activity["SuccessEvents"]["DistinctAddressCount"])) <= 1
    and (toint(Activity["SuccessEvents"]["SprayCount"]) - toint(Activity["SuccessEvents"]["DistinctAddressCount"])) <= 1
    )
| mv-apply UserDisplayName = UserDisplayNamesSample to typeof(string) on (
    reduce by UserDisplayName with characters=" "
    | top 1 by Count desc
    | extend UserDisplayNamesSample = case(
        Pattern != "others", pack("UserDisplayNamePattern", Pattern, "Count", Count),
        dynamic(null)
        )
    | project-away Pattern, Count, Representative
    )
// Optional tags based on the addresses
// | extend SuspiciousAddresses = _SuspiciousAddresses
// | mv-apply SuspiciousIP = SuspiciousAddresses to typeof(string) on (
//     extend Match = ipv6_compare(SuspiciousIP, IPRange, iff(isnotempty(parse_ipv4(SuspiciousIP)), 128 - (32 - ipv4_prefix_mask), ipv6_prefix_mask))
//     | summarize Match = make_set(Match)
//     | extend MaliciousRange = Match has "0"
// )
// | project-away Match, SuspiciousAddresses
// | extend KnownAddresses = _ExpectedIPAddresses
// | mv-apply KnownIP = KnownAddresses to typeof(string) on (
//     extend Match = ipv6_compare(KnownIP, IPRange, iff(isnotempty(parse_ipv4(SuspiciousIP)), 128 - (32 - ipv4_prefix_mask), ipv6_prefix_mask))
//     | summarize Match = make_set(Match)
//     | extend KnownRange = Match has "0"
// )
// | project-away Match, KnownAddresses
// Unpack useful information
// | extend FailureActivity = Activity["FailureEvents"]
// | evaluate bag_unpack(FailureActivity, OutputColumnPrefix="Failure_", ignoredProperties=dynamic(["PartialSuccessResultType", "IPAddresses"]))
// | extend SuccessActivity = Activity["SuccessEvents"]
// | evaluate bag_unpack(SuccessActivity, OutputColumnPrefix="Success_", ignoredProperties=dynamic(["PartialSuccessResultType", "IPAddresses"]))
// Format address range to IPv4 if possible
| extend HexCodes = split(extract(@"^(?i:0+\:0+\:0+\:0+\:0+\:ffff\:([a-f0-9]+\:[a-f0-9]+))$", 1, IPRange), ":")
| extend
    IPRange = case(
        array_length(HexCodes) == 2, format_ipv4_mask(tolong(strcat("0x", tostring(HexCodes[0])))*65536 + tolong(strcat("0x", tostring(HexCodes[1]))), ipv4_prefix_mask),
        IPRange
    ),
    AddressScope = case(
        array_length(HexCodes) == 2, ipv4_prefix_mask,
        ipv6_prefix_mask
    )
// Prefill entities
| mv-apply with_itemindex = Index_aux IPAddress = IPAddresses to typeof(string) on (
    extend
        Range = iff(isnotempty(parse_ipv4(IPAddress)), parse_ipv6_mask(IPAddress, 128 - (32 - ipv4_prefix_mask)), parse_ipv6_mask(IPAddress, ipv6_prefix_mask))
    | extend
        RangeHexCodes = split(extract(@"^(?i:0+\:0+\:0+\:0+\:0+\:ffff\:([a-f0-9]+\:[a-f0-9]+))$", 1, Range), ":")
    | extend
        Range = case(
            array_length(RangeHexCodes) == 2, format_ipv4_mask(tolong(strcat("0x", tostring(RangeHexCodes[0])))*65536 + tolong(strcat("0x", tostring(RangeHexCodes[1]))), ipv4_prefix_mask),
            Range
            ),
        RangeScope = case(
            array_length(RangeHexCodes) == 2, ipv4_prefix_mask,
            ipv6_prefix_mask
        )
    | extend
        Range = tostring(split(Range, "/", 0))
    | extend
        AddressRange = pack("$id", tostring(Index_aux + 3), "Address", todynamic(Range)[0], "AddressScope", tostring(RangeScope), "Type", "ip"),
        Address = pack("$id", tostring(Index_aux + 3), "Address", tostring(IPAddress), "Type", "ip")
    | project-away Range, RangeHexCodes, RangeScope
    | summarize CorrelationIdEntities = make_list(pack_array(Address))
    )
| mv-apply with_itemindex = Index_aux IPAddress = IPAddresses to typeof(string) on (
    extend Entities = pack("$id", tostring(Index_aux + 3), "Address", tostring(IPAddress), "Type", "ip")
    | summarize IPRangeEntities = make_list(Entities)
    )
| extend IPRangeEntities = array_concat(pack_array(pack("$id", tostring(2), "Address", todynamic(tostring(split(IPRange, "/", 0)))[0], "AddressScope", tostring(AddressScope), "Type", "ip")), IPRangeEntities)
| extend Entities = tostring(array_sort_asc(case(
    isnotempty(IPRange), IPRangeEntities,
    CorrelationIdEntities
    )))
| extend
    Entities = case(
        Stage == "Password Spray", Entities,
        Stage == "Password Spray + Initial Access", "",
        Stage == "Initial Access", "",
        Entities
    ),
    AlertName = case(
        Stage == "Password Spray" and Recurrent, "Slow password spray attack - Recurrent address range",
        Stage == "Password Spray" and not(Recurrent), "Slow password spray attack",
        Stage == "Password Spray + Initial Access", "Slow password spray attack - Potential compromised account",
        Stage == "Initial Access", "Authentication of several accounts from unexpected source",
        "Slow password spray attack"
    ),
    AlertSeverity = case(
        Stage == "Password Spray", "Informational",
        Stage == "Password Spray + Initial Access" and Location in (_ExpectedLocations), "Medium",
        Stage == "Initial Access" and Location in (_ExpectedLocations), "Low",
        "High"
    ),
    BenignAlert = case(
        Stage == "Initial Access" and Recurrent, true,
        Stage == "Password Spray + Initial Access" and Recurrent, true,
        false
    )
| where not(BenignAlert)
// If there are more than 150 different IP ranges doing password spray, Sentinel will only generate 150 alerts (1 alert for each row) and 1 of those alerts will contain the data from several rows. If this "grouped" alert contains the data from more than 10 rows, some Entities data will be lost.
// So this query should try to group the rows and Entities by itself, instead of Sentinel. At least rows that won't generate an incident to be reviewed (e.g. Stage == "Password Spray") should be grouped.
| as hint.materialized=true _Events
| where not(Stage == "Password Spray")
| union (
    _Events
    | where Stage == "Password Spray"
    | summarize Entities = tostring(make_list(todynamic(Entities))) by Stage, AlertName, AlertSeverity, AuxiliarKey = case(isnotempty(CorrelationId), "CorrelationId", "IPRange")
    )
| sort by Stage desc, CorrelationId desc, toint(ASN) asc, toint(extract(@"^(\d+)", 1, IPRange)) asc, toint(extract(@"^\d+\.(\d+)", 1, IPRange)) asc, toint(extract(@"^\d+\.\d+\.(\d+)", 1, IPRange)) asc, toint(extract(@"^\d+\.\d+\.\d+\.(\d+)", 1, IPRange)) asc
| project-away HexCodes, AddressScope, OperatingSystemsCount, AutonomousSystemNumbers, Source, CorrelationIdEntities, IPRangeEntities, AuxiliarKey//, Locations
| project-reorder
    Stage,
    StartTime,
    EndTime,
    MaliciousRange*,
    KnownRange*,
    ResultTypesSample,
    Location,
    ASN,
    IPRange,
    CorrelationId,
    IPAddresses,
    Success_*,
    Failure_*,
    OperatingSystems,
    Browsers,
    //DeviceIds,
    DeviceNamesSample,
    DeviceIsManagedSample,
    DeviceTrustTypesSample,
    UserDisplayNamesSample,
    Activity,
    AlertName,
    AlertSeverity,
    Entities
//};
//Function(query_frequency, query_period)