|
Aggregation & Grouping
stats / summarize
|
| stats count, dc(dest_port) as d_ports by src_ip
|
SELECT sourceIP, COUNT(*), COUNT(DISTINCT destinationPort) FROM events GROUP BY sourceIP
|
| summarize Total=count(), DPorts=dcount(DestinationPort) by SourceIP
|
| groupBy([LocalAddressIP4], function=[count(), count(RemotePort, distinct=true, as=d_ports)])
|
|
Regex Extraction
rex / parse / extract
|
| rex field=_raw "user=(?<user>[^ ]+)"
|
SELECT UTF8(payload) FROM events WHERE UTF8(payload) MATCHES 'user=(\w+)'
|
| parse-where RawData with * "user=" User:string " "*
|
| regex("user=(?<UserName>[^ ]+)", field=@rawstring)
|
|
Calculated Fields
eval / extend / case
|
| eval delta_ms = end_time - start_time, severity = if(score > 80, "CRIT", "INFO")
|
SELECT (endtime - starttime) as delta_ms, CASE WHEN score > 80 THEN 'CRIT' ELSE 'INFO' END FROM events
|
| extend delta_ms = datetime_diff('millisecond', EndTime, StartTime), severity = iff(score > 80, "CRIT", "INFO")
|
| delta_ms := end_time - start_time | case { score > 80 | severity := "CRIT" ; * | severity := "INFO" }
|
|
Lookup / Threat Match
lookup / match / in_set
|
| lookup ti_intel ip as src_ip OUTPUT actor, malware
|
SELECT * FROM events WHERE REFERENCESETCONTAINS('Malicious_IPs', sourceIP)
|
| lookup ThreatIntel on $left.SourceIP == $right.ThreatIP
|
| match(file="ti_intel.csv", column=ip, field=LocalAddressIP4, include=[actor, malware])
|
|
Event Correlation
join / selfJoinFilter
|
| join type=inner host [search eventtype=failed_logon | stats count by host]
|
/* Limited AQL support: requires expensive subqueries or Custom Rules Engine */
|
SigninLogs | join kind=inner (SecurityAlert) on $left.UserId == $right.UserId
|
| selfJoinFilter(field=aid, where=[{#event_simpleName=UserLogonFailed}, {#event_simpleName=ProcessRollup2}])
|
|
CIDR Subnet Filter
cidrmatch / in_subnet
|
| where cidrmatch("10.0.0.0/8", src_ip)
|
WHERE INCIDR('10.0.0.0/8', sourceIP)
|
| where ipv4_is_in_range(SourceIP, "10.0.0.0/8")
|
| cidr(LocalAddressIP4, subnet="10.0.0.0/8")
|
|
Time Aggregation
timechart / bucket / bin
|
| timechart span=1h count by event_type
|
SELECT DATEFORMAT(starttime, 'YYYY-MM-dd HH:00'), COUNT(*) GROUP BY 1
|
| summarize count() by bin(TimeGenerated, 1h), EventType
|
| bucket(span=1h, field=@timestamp, function=count())
|