KQL Search

// Rule    : Workload Identity - Service Principal Token Replay (Same Token, Multiple IPs/Countries)
// Severity: High
// Tactics : CredentialAccess, LateralMovement
// MITRE   : T1528 (Steal Application Access Token), T1550.001
// Freq    : PT1H   Period: PT1H
//==========================================================================================

// ---- Network Allowlist (exclude trusted IPs / CIDR / ranges) --------------
let _allow = materialize(union isfuzzy=true (print R="" | take 0), (_GetWatchlist('NetworkAllowlist') | project R = tostring(IPOrRange)) | where isnotempty(R));
let _allowCIDR  = toscalar(_allow | where not(R matches regex @'^\d+\.\d+\.\d+\.\d+-\d+\.\d+\.\d+\.\d+$') | extend R = iff(R has '/', R, strcat(R, '/32')) | summarize make_list(R));
let _allowRange = toscalar(_allow | where R matches regex @'^\d+\.\d+\.\d+\.\d+-\d+\.\d+\.\d+\.\d+$' | summarize make_list(R));
let _ExcludeAllowlistedIPs = (T:(IPAddress:string)) {
    T
    | extend IPAddress = tostring(IPAddress)
    | where array_length(_allowCIDR) == 0 or isnull(ipv4_is_in_any_range(IPAddress, _allowCIDR)) or not(ipv4_is_in_any_range(IPAddress, _allowCIDR))
    | mv-apply _r = _allowRange to typeof(string) on (
        extend _lo = tostring(split(_r,'-')[0]), _hi = tostring(split(_r,'-')[1])
        | extend _inRange = ipv4_compare(IPAddress, _lo) >= 0 and ipv4_compare(IPAddress, _hi) <= 0
        | summarize _anyInRange = max(toint(_inRange)))
    | where isnull(_anyInRange) or _anyInRange == 0
    | project-away _anyInRange
};
// ---------------------------------------------------------------------------
let PrivateRanges = dynamic(["10.", "192.168.", "172.16.", "172.17.", "172.18.",
    "172.19.", "172.20.", "172.21.", "172.22.", "172.23.", "172.24.", "172.25.",
    "172.26.", "172.27.", "172.28.", "172.29.", "172.30.", "172.31.",
    "127.", "169.254.", "168.63."]);
(AADServicePrincipalSignInLogs | invoke _ExcludeAllowlistedIPs())
| where TimeGenerated > ago(1h)
| where ResultType == "0"
| where isnotempty(UniqueTokenIdentifier)
| where isnotempty(IPAddress)
| where not(IPAddress has_any (PrivateRanges))
| extend GeoInfo  = geo_info_from_ip_address(IPAddress)
| extend Country  = tostring(GeoInfo.country_iso_code)
| summarize
    UniqueIPs       = dcount(IPAddress),
    UniqueCountries = dcount(Country),
    IPList          = make_set(IPAddress, 10),
    Countries       = make_set(Country, 10),
    UseCount        = count(),
    Resources       = make_set(ResourceDisplayName, 10),
    FirstSeen       = min(TimeGenerated),
    LastSeen        = max(TimeGenerated)
    by UniqueTokenIdentifier, ServicePrincipalName, ServicePrincipalId, AppId
// Token replayed from 3+ different IPs or 2+ countries within the window
| where UniqueIPs >= 3 or UniqueCountries >= 2
| extend
    MinuteSpan    = datetime_diff("minute", LastSeen, FirstSeen),
    RiskLevel     = case(
        UniqueCountries >= 3,                 "Critical",
        UniqueCountries >= 2 and UniqueIPs >= 3, "High",
        "Medium")
| where MinuteSpan <= 60