Advanced KQL for Threat Hunting: Window Functions — Part 1

Photo by R Mo on Unsplash

Window functions are one of the powerful methods for data analysis. While they are primarily used in finance and business analytics, they can also be used in threat hunting and DFIR and solve complicated use cases. In this post, I will briefly explain two KQL(Kusto Query Language)window functions, prev() and next(), and how to use them for threat hunting. I’ll use the cloud account takeover scenario I’ve previously posted as an example to demonstrate. If you haven’t read it yet, take a quick look:

Detecting Azure AD Account Takeover Attacks

What is a window function?

From PostgreSQL documentation:

A window function performs a calculation across a set of table rows that are somehow related to the current row. This is comparable to the type of calculation that can be done with an aggregate function. However, window functions do not cause rows to become grouped into a single output row like non-window aggregate calls would. Instead, the rows retain their separate identities. Behind the scenes, the window function is able to access more than just the current row of the query result.

To use window functions, we need to sort/order the data. Otherwise, it doesn’t make sense to access the previous or next row.

prev() function

The prev() function returns a value of a column from a row that is a specified number of rows(offset) before the current row. If there is no row at the specified offset, it returns null or a default value defined in the function. Let’s see an example:

next() function

Similarly, the next() function returns a value of a column from a row that is a specified number of rows(offset) after the current row.

Now we see what they do; let’s see how we can use them in threat hunting.

Below is the anatomy of an account takeover attack(from the previous blog):

0: the time when the attack is executed
t: timespan

1. A victim is performing regular daily tasks on the computer (0-t)
2. In some way, the victim gets tricked into signing in somewhere, and the credentials get stolen (0)
3. The victim continues working on the computer(0+t)
4. The attacker signs in with the stolen credentials and starts performing malicious activities(0+t)

And the detection logic:

If a user signs in from an IP address that has not been observed in the last X days AND the sign-in happens in close proximity to the user’s latest sign-in time, generate an alert.

One thing to note about the sign-in activity is that there must be an App used in the attack. Therefore, we’ll also use the AppIdinformation in the detection logic as well this time.

Since the query needs to analyze lots of accounts and apps simultaneously, we need to use the window function carefully.

1. Generate a list of known IPs for each user

let query_period = 1d;
let look_back = 14d;
let knownIPs = 
    union SigninLogs, AADNonInteractiveUserSignInLogs
    | where TimeGenerated between (ago(look_back) .. ago(query_period))
    | summarize ObservedIPsOfUPN = make_set(IPAddress) by UserPrincipalName
    ;

2. Combining SigninLogs and AADNonInteractiveUserSignInLogs:

2.a. Identify when a user first seen
Here, we need to put an if condition to analyse each user separately.

union SigninLogs, AADNonInteractiveUserSignInLogs
| where TimeGenerated > ago(query_period)
| sort by  UserPrincipalName asc, TimeGenerated asc // sorting only by User and Timegenerated
| extend PrevUserPrincipleName = iff(prev(UserPrincipalName) != UserPrincipalName, 'FirstTimeSeen', prev(UserPrincipalName))
| extend TimeDiffInSeconds = iff(prev(UserPrincipalName) == UserPrincipalName, datetime_diff('second', TimeGenerated, prev(TimeGenerated)), -1)

2.b. Identify when an application was first seen for each user

// here we are sorting the data again and adding AppId 
| sort by UserPrincipalName asc, AppId asc, TimeGenerated asc
| extend PrevAppId = case(PrevUserPrincipleName == 'FirstTimeSeen', 'FirstTimeSeen', PrevUserPrincipleName == UserPrincipalName and prev(AppId) != AppId, 'FirstTimeSeen', prev(AppId))
| extend PrevIPAddress = case(PrevUserPrincipleName == 'FirstTimeSeen', 'FirstTimeSeen',
                              PrevUserPrincipleName == UserPrincipalName and PrevAppId == 'FirstTimeSeen', 'FirstTimeSeen',
                              prev(IPAddress))

2.c. Enrich data with ObservedIPsOfUPN(comes from the knownIPs query)

| lookup kind=leftouter knownIPs on UserPrincipalName

2.d. Assign a Risk_Score by comparing the PrevIPAddresswith ObservedIPsOfUPN

| extend Risk_Score = case(IPAddress == PrevIPAddress, 0,
                           PrevIPAddress == 'FirstTimeSeen' and ObservedIPsOfUPN has IPAddress, 1,
                           PrevIPAddress != 'FirstTimeSeen' and ObservedIPsOfUPN has IPAddress, 2,
                           PrevIPAddress == 'FirstTimeSeen' and ObservedIPsOfUPN !has IPAddress, 3,
                           PrevIPAddress != 'FirstTimeSeen' and ObservedIPsOfUPN !has IPAddress, 4,
                           99)

3. Filter the results based on the Risk_Score
You should be looking for results having Risk_Score greater than 2.

Below is the whole query:

let query_period = 1d;
let look_back = 14d;
let knownIPs = 
    union SigninLogs, AADNonInteractiveUserSignInLogs
    | where TimeGenerated between (ago(look_back) .. ago(query_period))
    | summarize ObservedIPsOfUPN = make_set(IPAddress) by UserPrincipalName
    ;
union SigninLogs, AADNonInteractiveUserSignInLogs
| where TimeGenerated > ago(query_period)
| sort by  UserPrincipalName asc, TimeGenerated asc
| extend PrevUserPrincipleName = iff(prev(UserPrincipalName) != UserPrincipalName, 'FirstTimeSeen', prev(UserPrincipalName))
| extend TimeDiffInSeconds = iff(prev(UserPrincipalName) == UserPrincipalName, datetime_diff('second', TimeGenerated, prev(TimeGenerated)), -1)
| sort by UserPrincipalName asc, AppId asc, TimeGenerated asc
| extend PrevAppId = case(PrevUserPrincipleName == 'FirstTimeSeen', 'FirstTimeSeen', PrevUserPrincipleName == UserPrincipalName and prev(AppId) != AppId, 'FirstTimeSeen', prev(AppId))
| extend PrevIPAddress = case(PrevUserPrincipleName == 'FirstTimeSeen', 'FirstTimeSeen',
                              PrevUserPrincipleName == UserPrincipalName and PrevAppId == 'FirstTimeSeen', 'FirstTimeSeen',
                              prev(IPAddress))
| lookup kind=leftouter knownIPs on UserPrincipalName
| extend Risk_Score = case(IPAddress == PrevIPAddress, 0,
                           PrevIPAddress == 'FirstTimeSeen' and ObservedIPsOfUPN has IPAddress, 1,
                           PrevIPAddress != 'FirstTimeSeen' and ObservedIPsOfUPN has IPAddress, 2,
                           PrevIPAddress == 'FirstTimeSeen' and ObservedIPsOfUPN !has IPAddress, 3,
                           PrevIPAddress != 'FirstTimeSeen' and ObservedIPsOfUPN !has IPAddress, 4,
                           99)
| where Risk_Score > 2
| project-reorder TimeGenerated, TimeDiffInSeconds, UserPrincipalName, PrevUserPrincipleName, AppDisplayName, AppId, PrevAppId, IPAddress, PrevIPAddress, Risk_Score, ObservedIPsOfUPN
| sort by UserPrincipalName asc, TimeGenerated asc

Thanks for reading this article! If you have any questions, leave a comment below. Want to master KQL for Threat Hunting, Detection Engineering, and DFIR in a hyper-realistic environment? Visit my academy for a free course!

The queries are also available on my GitHub repo.

Threat-Hunting-and-Detection/Potential Cloud Acount Takeover - Window.md at…

Mehmet is the founder of Blu Raven Academy. He brings over 15 years of experience in cybersecurity, with a unique blend of expertise in KQL, threat hunting, detection engineering, and data science to his courses to help others advance their skills. Recognized four times as a Microsoft Security MVP, he is renowned for adapting the RITA beacon analyzer to KQL, developing novel methods for detecting threats, and for his insightful presentations at key conferences like the SANS DFIR Summit.