RULE 03 SP Token Replay Multi IP

Query

// 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

Explanation

This KQL query is designed to detect suspicious activity involving the use of service principal tokens in Azure Active Directory (AAD). Here's a simplified breakdown of what the query does:

Network Allowlist: It first defines a list of trusted IP addresses or ranges that should be excluded from analysis. This is done by fetching a watchlist named 'NetworkAllowlist' and processing it to handle both CIDR notation and IP ranges.
Private IP Ranges: It defines a list of private IP address ranges that should be ignored in the analysis, as these are typically internal network addresses.
Data Filtering: The query then processes the AADServicePrincipalSignInLogs to:
- Exclude any sign-ins from allowlisted IPs.
- Focus on sign-ins that occurred in the last hour (TimeGenerated > ago(1h)).
- Only consider successful sign-ins (ResultType == "0").
- Ensure that the sign-in event has a unique token identifier and a non-private IP address.
Geographical Information: It retrieves geographical information based on the IP address to determine the country of origin for each sign-in.
Aggregation: The query aggregates data by the unique token identifier and service principal details, summarizing:
- The number of unique IPs and countries from which the token was used.
- A list of up to 10 IP addresses and countries involved.
- The total number of times the token was used.
- The resources accessed.
- The first and last time the token was seen.
Suspicious Activity Detection: It identifies tokens that have been used from:
- 3 or more different IP addresses, or
- 2 or more different countries within the last hour.
Risk Assessment: It assigns a risk level based on the number of unique countries and IPs:
- "Critical" if used from 3 or more countries.
- "High" if used from 2 countries and 3 or more IPs.
- "Medium" otherwise.
Time Constraint: It ensures that the suspicious activity (token replay) occurred within a 60-minute window.

Overall, this query aims to detect potential token replay attacks where the same service principal token is used from multiple locations, indicating possible credential theft or misuse.

Details

David Alonso

Released: April 21, 2026

Tables

AADServicePrincipalSignInLogs

Keywords

WorkloadIdentityServicePrincipalTokenReplayIPsCountriesCredentialAccessLateralMovementApplicationNetworkAllowlistCIDRRangesIPAddressIP

Operators

letmaterializeunionisfuzzyprinttakeprojecttostringwhereisnotemptytoscalarmatches regexextendiffstrcatsummarizemake_listnotarray_lengthisnullipv4_is_in_any_rangemv-applyto typeofsplitipv4_comparemaxtointproject-awaydynamicinvokeagohas_anygeo_info_from_ip_addressdcountmake_setcountminbydatetime_diffcase

MITRE Techniques

T1528 T1550.001

Actions

GitHub

KQL Search