Lesson 5 of 5 · 30 min read

Worked Example: Failed Logons + Success in SPL, KQL, and EQL

The best way to truly understand query language differences is to write the same detection in each. This lesson works through a real detection scenario — brute force followed by account compromise — in SPL, KQL, and EQL.

Scenario: An attacker performs a password spray against corporate Entra ID/Active Directory. They try each account once per hour (to stay below lockout thresholds). After finding valid credentials for user jdoe, they authenticate successfully.


Defining the Detection

Detection goal: Find source IPs that:

  1. Had > 10 failed authentication attempts
  2. Against > 5 unique accounts
  3. Followed by at least one successful authentication

Within: 1 hour

Log sources: Windows Security Event 4624/4625 (AD), SigninLogs (Entra ID), or any authentication source


Version 1: SPL (Splunk)

// Step 1: Find spraying sources
index=windows (EventCode=4625 OR EventCode=4624) earliest=-1h
| eval Outcome=if(EventCode=4624, "success", "failure")
| stats 
    count(eval(Outcome="failure")) as Failures,
    dc(eval(Outcome="failure", TargetUserName, null())) as UniqueTargets,
    count(eval(Outcome="success")) as Successes,
    values(eval(Outcome="success", TargetUserName, null())) as CompromisedAccounts
  by IpAddress
| where Failures > 10 AND UniqueTargets > 5 AND Successes > 0
| eval SprayAndCompromise="YES"
| sort -Failures
| table IpAddress, Failures, UniqueTargets, Successes, CompromisedAccounts

Breakdown:

Alternative two-pass approach (more readable):

index=windows EventCode=4625 earliest=-1h
| stats count as Failures, dc(TargetUserName) as UniqueTargets by IpAddress
| where Failures > 10 AND UniqueTargets > 5
| join IpAddress [
    search index=windows EventCode=4624 earliest=-1h
    | stats count as Successes, values(TargetUserName) as Accounts by IpAddress
  ]
| where Successes > 0
| table IpAddress, Failures, UniqueTargets, Successes, Accounts

Version 2: KQL (Microsoft Sentinel)

// Method 1: Using let statements for clarity
let Failures = SecurityEvent
    | where TimeGenerated > ago(1h)
    | where EventID == 4625
    | where Account !contains "ANONYMOUS"
    | summarize
        FailCount = count(),
        UniqueAccounts = dcount(Account),
        Accounts = make_set(Account)
      by IpAddress;
let Successes = SecurityEvent
    | where TimeGenerated > ago(1h)
    | where EventID == 4624
    | where LogonType in (3, 10)  // Network or RemoteInteractive (not local)
    | summarize
        SuccessCount = count(),
        SuccessAccounts = make_set(Account)
      by IpAddress;
Failures
| where FailCount > 10 and UniqueAccounts > 5
| join kind=inner Successes on IpAddress
| project
    IpAddress,
    FailCount,
    UniqueAccounts,
    SuccessCount,
    SuccessAccounts,
    SprayedAccounts = Accounts
| order by FailCount desc

Breakdown:


Version 3: EQL (Elastic Security)

EQL is best for the sequence version of this detection — failure events THEN a success event, ordered:

sequence by source.ip with maxspan=1h
  [authentication where event.outcome == "failure"] with runs=10
  [authentication where event.outcome == "success"]

What with runs=10 means: The failure event must occur at least 10 times before the sequence advances to the success event. This ensures we’re looking at a spray, not a single typo followed by a correct password.

For the “unique accounts” part of the spray pattern, EQL alone can’t express it (EQL doesn’t aggregate inside sequence conditions). Combine with an ES|QL pre-filter:

// First: find spraying IPs via ES|QL (aggregation)
FROM logs-*
| WHERE event.category == "authentication" AND event.outcome == "failure"
| WHERE @timestamp > NOW() - 1 HOURS
| STATS failures=COUNT(*), unique_accounts=COUNT_DISTINCT(user.name) BY source.ip
| WHERE failures > 10 AND unique_accounts > 5
// → produces a list of suspect IPs

// Then: use EQL to find if those IPs had successes (in the detection rule alert condition)

In Elastic Security, this is typically implemented as two separate rules with an alert correlation step.


Comparison Table

FeatureSPLKQLEQL
Distinct countdc()dcount()Not native (use ES
Conditional countcount(eval(...))countif(...)with runs=N
Temporal sequenceComplex join on timeComplex join on timeNative sequence ... with maxspan
ReadabilityPipelineLet + pipelineEvent-first
Performance tuningIndex firstPre-filter with timemaxspan limits scope
Sigma Rule Converter— sigconverter.io Uncoder IO — Sigma Conversion— Uncoder