Detect Enumeration Activity Using Unique Identifiers and Session Aggregation
Description
"This Kusto (KQL) hunting query detects blob-enumeration or file-spraying behaviour in Azure Storage by: - Aggregating requests into time-bound sessions with row_window_session(). - Defining a "user" as the combination of AccountName | UserAgentHeader, which is tolerant of rapid IP rotation. - Raising an alert when, within any single session: - The actor touches at least 10 distinct objects, and - At least 50 % of those requests return an HTTP 4xx/5xx status, and - Each object is accessed exactly once ("touch-once" pattern typical of enumeration). All tuning knobs (look-back window, session gap, thresholds, authentication type) are exposed at the top of the query. Be sure to validate the query against both benign and malicious samples in your own environment before relying on it for production detection. Reference: https://techcommunity.microsoft.com/blog/microsoftdefendercloudblog/protect-your-storage-resources-against-blob-hunting/3735238"
Query · kql
// ==========================================================
// Detect Enumeration Activity Using Unique User Identifiers
// ==========================================================
// ---------- Tuning parameters ----------
let maxTimeBetweenRequests = 30s; // gap that keeps events in the same session
let maxWindowTime = 12h; // hard cap on session duration
let lookback = 30d; // history window
let authTypes = dynamic(['Anonymous']); // adjust if needed
let minDistinctObjects = 10; // minimum objects per session
let minFailureRatio = 0.5; // minimum failure rate (0.5 = 50 %)
// ---------------------------------------
// ---------- Core query ----------
StorageBlobLogs
| where TimeGenerated > ago(lookback)
| where AuthenticationType has_any(authTypes)
| where Category == 'StorageRead'
| where Uri !endswith 'favicon.ico'
// Extract full blob path
| extend FilePathElems = array_slice(split(split(Uri,'?')[0], '/'), 3, -1)
| extend FullPath = strcat('/', strcat_array(FilePathElems, '/'))
// Normalise caller IP
| extend CallerIp = tostring(split(CallerIpAddress, ':')[0])
| where not(ipv4_is_private(CallerIp))
// Build identity key resilient to IP hopping
| extend IdentityKey = strcat(AccountName, '|', UserAgentHeader)
| project TimeGenerated, AccountName, FullPath,
CallerIp, StatusCode, UserAgentHeader, IdentityKey
| order by TimeGenerated asc
| serialize
// Build activity sessions
| extend SessionId = row_window_session(
TimeGenerated,
maxWindowTime,
maxTimeBetweenRequests,
IdentityKey != prev(IdentityKey))
// Status code as integer for numeric comparisons
| extend StatusCodeInt = toint(StatusCode)
| summarize
DistinctObjCount = dcount(FullPath),
RequestCount = count(),
FailCount = countif(StatusCodeInt >= 400),
CallerIPs = make_set(CallerIp, 64),
SessionStart = min(TimeGenerated),
SessionEnd = max(TimeGenerated),
UserAgent = any(UserAgentHeader)
by SessionId, IdentityKey, AccountName
| extend FailureRatio = todouble(FailCount) / RequestCount,
DurationMins = datetime_diff('minute', SessionEnd, SessionStart)
// ----------- Detection logic ----------
| where DistinctObjCount >= minDistinctObjects
and FailureRatio >= minFailureRatio
and DistinctObjCount == RequestCount // touch-once rule
// ----------- Output ----------
| project SessionStart, SessionEnd, DurationMins,
AccountName, IdentityKey, DistinctObjCount,
RequestCount, FailureRatio,
CallerIPCount = array_length(CallerIPs),
CallerIPs, UserAgent
| order by DistinctObjCount desc