# KQL Telemetry, Detection & Automation 50 KQL recipes for Microsoft Sentinel and Defender XDR. Each one runs in the Sentinel Logs blade or the Defender XDR Advanced Hunting blade. The recipes assume the default ingest tables (`SigninLogs`, `SecurityEvent`, `OfficeActivity`, `DeviceEvents`, `DeviceProcessEvents`, etc.) — adjust table names if you've remapped them. Sections: - **Telemetry** (1–20) — read, parse, project, join, enrich. - **Detection** (21–35) — failed-login burst, impossible travel, LSASS access, encoded PowerShell, new admin, OAuth consent grant, beacon, DGA. - **Automation** (36–50) — analytic rules, watchlists, workbooks, functions, materialized views, cross-workspace, Logic App triggers. --- ## Table of contents | # | Recipe | |---|---| | 1 | [List recent SecurityEvent records](#1-list-recent-securityevent-records) | | 2 | [Filter SigninLogs to failures](#2-filter-signinlogs-to-failures) | | 3 | [Parse a JSON field with parse_json](#3-parse-a-json-field-with-parse_json) | | 4 | [Parse a CSV string in a column](#4-parse-a-csv-string-in-a-column) | | 5 | [Base64-decode an encoded command](#5-base64-decode-an-encoded-command) | | 6 | [Bucket events into 5-minute windows](#6-bucket-events-into-5-minute-windows) | | 7 | [Render a timechart](#7-render-a-timechart) | | 8 | [Top N source IPs by failures](#8-top-n-source-ips-by-failures) | | 9 | [Distinct count of users per host](#9-distinct-count-of-users-per-host) | | 10 | [Join SigninLogs with IdentityInfo](#10-join-signinlogs-with-identityinfo) | | 11 | [externaldata for a lookup table](#11-externaldata-for-a-lookup-table) | | 12 | [Use let for parameterized queries](#12-use-let-for-parameterized-queries) | | 13 | [Sysmon EID 1 process creations](#13-sysmon-eid-1-process-creations) | | 14 | [OfficeActivity mailbox events](#14-officeactivity-mailbox-events) | | 15 | [CloudAppEvents from Defender for Cloud Apps](#15-cloudappevents-from-defender-for-cloud-apps) | | 16 | [DeviceProcessEvents in Defender XDR](#16-deviceprocessevents-in-defender-xdr) | | 17 | [DeviceNetworkEvents](#17-devicenetworkevents) | | 18 | [DeviceLogonEvents](#18-devicelogonevents) | | 19 | [AzureActivity](#19-azureactivity) | | 20 | [Parse XML payloads](#20-parse-xml-payloads) | | 21 | [Failed sign-in burst from one IP](#21-failed-sign-in-burst-from-one-ip) | | 22 | [Impossible-travel via haversine](#22-impossible-travel-via-haversine) | | 23 | [LSASS access (Sysmon EID 10)](#23-lsass-access-sysmon-eid-10) | | 24 | [PowerShell -EncodedCommand](#24-powershell--encodedcommand) | | 25 | [New local admin (4720 + 4732 join)](#25-new-local-admin-4720--4732-join) | | 26 | [Lateral SMB by user + host pair](#26-lateral-smb-by-user--host-pair) | | 27 | [Off-hours admin sign-ins](#27-off-hours-admin-sign-ins) | | 28 | [WMI persistence creation triplet](#28-wmi-persistence-creation-triplet) | | 29 | [Mailbox forwarding rule created](#29-mailbox-forwarding-rule-created) | | 30 | [OAuth high-privilege consent grant](#30-oauth-high-privilege-consent-grant) | | 31 | [Office process spawning PowerShell](#31-office-process-spawning-powershell) | | 32 | [Rundll32 with suspicious parent](#32-rundll32-with-suspicious-parent) | | 33 | [Beacon detection via interval regularity](#33-beacon-detection-via-interval-regularity) | | 34 | [DGA via subdomain entropy](#34-dga-via-subdomain-entropy) | | 35 | [New cloud admin role assignment](#35-new-cloud-admin-role-assignment) | | 36 | [Save query as analytic rule](#36-save-query-as-analytic-rule) | | 37 | [Entity mappings for incident graph](#37-entity-mappings-for-incident-graph) | | 38 | [Create a watchlist via externaldata](#38-create-a-watchlist-via-externaldata) | | 39 | [Schedule a hunting query](#39-schedule-a-hunting-query) | | 40 | [Ingest custom logs via Data Collector](#40-ingest-custom-logs-via-data-collector) | | 41 | [Cross-workspace query](#41-cross-workspace-query) | | 42 | [Define a function (saved KQL helper)](#42-define-a-function-saved-kql-helper) | | 43 | [Materialized view for high-volume table](#43-materialized-view-for-high-volume-table) | | 44 | [ipv4_lookup for ASN enrichment](#44-ipv4_lookup-for-asn-enrichment) | | 45 | [GeoIP enrichment with externaldata](#45-geoip-enrichment-with-externaldata) | | 46 | [Logic App trigger from query result](#46-logic-app-trigger-from-query-result) | | 47 | [Workbook tile from a query](#47-workbook-tile-from-a-query) | | 48 | [Anomaly detection with series_decompose_anomalies](#48-anomaly-detection-with-series_decompose_anomalies) | | 49 | [adx() bridge to Azure Data Explorer](#49-adx-bridge-to-azure-data-explorer) | | 50 | [Compose detection + enrichment in one query](#50-compose-detection--enrichment-in-one-query) | --- ## Telemetry ### 1. List recent SecurityEvent records ```kql SecurityEvent | where TimeGenerated > ago(15m) | take 100 ``` **What it does:** Returns the most recent rows from the Windows Security event table. **Run:** Sentinel Logs blade. Time picker: last 24h is fine, the `where` will narrow further. **Sample output:** rows from `SecurityEvent` with all standard fields. **Tuning:** swap `take` for `top 100 by TimeGenerated desc` if you actually need the *latest* 100 — `take` is non-deterministic order. --- ### 2. Filter SigninLogs to failures ```kql SigninLogs | where TimeGenerated > ago(1h) | where ResultType != 0 | project TimeGenerated, UserPrincipalName, IPAddress, ResultType, ResultDescription ``` **Sample output:** ``` TimeGenerated UserPrincipalName IPAddress ResultType ResultDescription 2026-05-23T09:11Z admin@corp.com 185.220.101.34 50126 Invalid username or password ``` **Tuning:** `AADNonInteractiveUserSignInLogs` has the same shape — query both with `union` when hunting token-based flows. --- ### 3. Parse a JSON field with parse_json ```kql SecurityEvent | where EventID == 4625 | extend Props = parse_json(EventData) | project TimeGenerated, User = tostring(Props.TargetUserName), IP = tostring(Props.IpAddress) ``` **Sample output:** ``` TimeGenerated User IP 2026-05-23T09:11Z admin 185.220.101.34 ``` **Tuning:** `tostring()` after JSON unboxing is cheap and almost always what you want — Sentinel field types default to `dynamic` from JSON. --- ### 4. Parse a CSV string in a column ```kql let raw = datatable(line: string) [ "2026-05-23T09:11Z,jdoe,10.0.4.27,success", "2026-05-23T09:11Z,admin,185.220.101.34,failure" ]; raw | extend parts = split(line, ",") | project ts = tostring(parts[0]), user = tostring(parts[1]), ip = tostring(parts[2]), outcome = tostring(parts[3]) | where outcome == "failure" ``` **Sample output:** ``` ts user ip outcome 2026-05-23T09:11Z admin 185.220.101.34 failure ``` **Tuning:** for quoted CSV (with embedded commas) use `parse_csv` instead of `split`. --- ### 5. Base64-decode an encoded command ```kql DeviceProcessEvents | where ProcessCommandLine matches regex @"(?i)-(?:enc|encodedcommand)\s+([A-Za-z0-9+/=]+)" | extend b64 = extract(@"(?i)-(?:enc|encodedcommand)\s+([A-Za-z0-9+/=]+)", 1, ProcessCommandLine) | extend decoded = base64_decode_tostring(b64) | project TimeGenerated, DeviceName, AccountName, decoded ``` **Sample output:** ``` TimeGenerated DeviceName AccountName decoded ... WIN10-LAB-02 jdoe $s=New-Object Net.WebClient;$s.DownloadString('... ``` **Tuning:** `base64_decode_tostring` in KQL decodes as UTF-8 — for PowerShell which writes UTF-16 LE the output looks "spaced out." Wrap in `replace_string(.., "", "")` to flatten. --- ### 6. Bucket events into 5-minute windows ```kql SigninLogs | where ResultType != 0 | summarize Count = count() by bin(TimeGenerated, 5m), IPAddress | order by Count desc | take 20 ``` **Sample output:** ``` TimeGenerated IPAddress Count 2026-05-23T09:10Z 185.220.101.34 47 ``` **Tuning:** `bin()` is the equivalent of SQL date-truncation. For 1-second buckets use `bin(TimeGenerated, 1s)`. --- ### 7. Render a timechart ```kql SigninLogs | where TimeGenerated > ago(24h) | where ResultType != 0 | summarize FailedCount = count() by bin(TimeGenerated, 15m) | render timechart ``` **Sample output:** an interactive line chart in the Logs blade. **Tuning:** `render columnchart` for bars; `render piechart` for compositions. The render directive is ignored when the query is consumed via API. --- ### 8. Top N source IPs by failures ```kql SigninLogs | where ResultType != 0 | summarize Failures = count(), UniqueUsers = dcount(UserPrincipalName) by IPAddress | top 10 by Failures desc ``` **Sample output:** ``` IPAddress Failures UniqueUsers 185.220.101.34 142 47 ``` **Tuning:** `UniqueUsers` distinguishes spray (high count, many users) from a single user retrying (high count, 1 user). --- ### 9. Distinct count of users per host ```kql SigninLogs | where ResultType == 0 | summarize UserCount = dcount(UserPrincipalName) by AppDisplayName, IPAddress | where UserCount > 10 | order by UserCount desc ``` **Sample output:** ``` AppDisplayName IPAddress UserCount Office 365 Exchange Online 10.0.4.27 45 ``` **Tuning:** `dcount` is an HLL approximation by default. For exact counts on small data use `dcount(..., 4)` (max accuracy). --- ### 10. Join SigninLogs with IdentityInfo ```kql SigninLogs | where TimeGenerated > ago(1h) | where ResultType != 0 | join kind=leftouter ( IdentityInfo | where TimeGenerated > ago(7d) | summarize arg_max(TimeGenerated, Department, JobTitle) by AccountUPN ) on $left.UserPrincipalName == $right.AccountUPN | project TimeGenerated, UserPrincipalName, Department, JobTitle, IPAddress ``` **Sample output:** ``` TimeGenerated UserPrincipalName Department JobTitle IPAddress 2026-05-23T09:11Z admin@corp.com IT Admin 185.220.101.34 ``` **Tuning:** the `arg_max` trick picks the most recent IdentityInfo row per user — `IdentityInfo` is updated every hour with new snapshots. --- ### 11. externaldata for a lookup table ```kql let badIps = externaldata(IP: string) [@"https://raw.githubusercontent.com/firehol/blocklist-ipsets/master/firehol_level1.netset"] with (format="txt", ignoreFirstRecord=false); SigninLogs | where IPAddress in (badIps) | project TimeGenerated, UserPrincipalName, IPAddress ``` **Sample output:** ``` TimeGenerated UserPrincipalName IPAddress 2026-05-23T09:11Z admin@corp.com 185.220.101.34 ``` **Tuning:** `externaldata` fetches once per query — fast enough for hunts but not for high-frequency analytic rules. For those use Watchlists (see #38). --- ### 12. Use let for parameterized queries ```kql let lookback = 1h; let threshold = 10; SigninLogs | where TimeGenerated > ago(lookback) | where ResultType != 0 | summarize Failures = count() by IPAddress | where Failures >= threshold ``` **Sample output:** ``` IPAddress Failures 185.220.101.34 47 ``` **Tuning:** `let` declarations at the top of a query make it easy to ship — copy-paste the query, change parameters at the top, run. Don't bury them inside subqueries. --- ### 13. Sysmon EID 1 process creations ```kql Event | where Source == "Microsoft-Windows-Sysmon" and EventID == 1 | extend EventData = parse_xml(EventData) | extend Image = tostring(EventData.DataItem.EventData.Data[4]["#text"]) | extend Cmd = tostring(EventData.DataItem.EventData.Data[10]["#text"]) | where Image endswith "\\powershell.exe" | project TimeGenerated, Image, Cmd ``` **Sample output:** ``` TimeGenerated Image Cmd 2026-05-23T09:11Z C:\Windows\System32\WindowsPowerShell\v1.0\powershell.exe -nop -enc ... ``` **Tuning:** Defender XDR has `DeviceProcessEvents` which is cleaner. Use Sysmon path only when XDR isn't available. --- ### 14. OfficeActivity mailbox events ```kql OfficeActivity | where OfficeWorkload == "Exchange" | where Operation in ("New-InboxRule", "Set-InboxRule") | project TimeGenerated, UserId, ClientIP, Operation, Parameters ``` **Sample output:** ``` TimeGenerated UserId ClientIP Operation Parameters 2026-05-23T11:09Z cfo@yourdomain.com 185.220.101.34 New-InboxRule {...} ``` **Tuning:** `Parameters` is a stringified JSON — `parse_json` it for further field access. --- ### 15. CloudAppEvents from Defender for Cloud Apps ```kql CloudAppEvents | where Application == "Microsoft Exchange Online" | where ActionType in ("New-InboxRule", "Set-Mailbox") | project Timestamp, AccountId, IPAddress, ActionType, RawEventData ``` **Sample output:** structured CloudAppEvents records. **Tuning:** `CloudAppEvents` is the Defender XDR alternative to Sentinel's `OfficeActivity` — same source data, slightly different schema. --- ### 16. DeviceProcessEvents in Defender XDR ```kql DeviceProcessEvents | where Timestamp > ago(1h) | where InitiatingProcessFileName == "winword.exe" | where FileName in ("powershell.exe", "cmd.exe", "wscript.exe", "cscript.exe", "mshta.exe") | project Timestamp, DeviceName, AccountName, FileName, ProcessCommandLine ``` **Sample output:** ``` Timestamp DeviceName AccountName FileName ProcessCommandLine ... WIN10-LAB-02 jdoe powershell.exe -enc ... ``` **Tuning:** the classic maldoc signature. Combine with `InitiatingProcessVersionInfoCompanyName == "Microsoft Corporation"` if a real Office is the parent. --- ### 17. DeviceNetworkEvents ```kql DeviceNetworkEvents | where ActionType == "ConnectionSuccess" | where RemotePort !in (80, 443, 53) | where InitiatingProcessFileName !in ("svchost.exe", "msedge.exe", "chrome.exe") | project Timestamp, DeviceName, InitiatingProcessFileName, RemoteIP, RemotePort ``` **Sample output:** ``` Timestamp DeviceName InitiatingProcessFileName RemoteIP RemotePort ... WIN10-LAB-02 update.exe 93.184.216.34 8080 ``` **Tuning:** combine with `DeviceProcessEvents` joined on `InitiatingProcessId` to get full process tree context. --- ### 18. DeviceLogonEvents ```kql DeviceLogonEvents | where LogonType in ("Network", "RemoteInteractive") | where ActionType == "LogonSuccess" | project Timestamp, DeviceName, AccountName, LogonType, RemoteIP, RemoteDeviceName ``` **Sample output:** ``` Timestamp DeviceName AccountName LogonType RemoteIP RemoteDeviceName ... DC01 jdoe RemoteInteractive 10.0.4.27 LAPTOP-12 ``` **Tuning:** lateral movement hunts pair this with `DeviceProcessEvents` for "what did they do after logging in." --- ### 19. AzureActivity ```kql AzureActivity | where TimeGenerated > ago(7d) | where OperationNameValue endswith "Microsoft.Authorization/roleAssignments/write" | project TimeGenerated, Caller, ResourceId, ActivityStatusValue ``` **Sample output:** ``` TimeGenerated Caller ResourceId Status 2026-05-23T09:11Z jdoe@yourdomain.com /subscriptions/.../providers/Microsoft.A... Succeeded ``` **Tuning:** for the role assigned, parse `Properties` JSON — `extend props = parse_json(Properties)`. --- ### 20. Parse XML payloads ```kql let xml_str = "12"; print parsed = parse_xml(xml_str) | extend a = parsed.root.a, b = parsed.root.b ``` **Sample output:** ``` parsed a b {root: {a: "1", b: "2"}} 1 2 ``` **Tuning:** `parse_xml` returns a `dynamic` — index it like JSON. For deeply-nested Sysmon XML, write a small `mv-expand` helper. --- ## Detection ### 21. Failed sign-in burst from one IP ```kql let threshold = 10; let lookback = 1h; SigninLogs | where TimeGenerated > ago(lookback) | where ResultType in (50053, 50056, 50126, 50057, 50064, 50034) | summarize Failures = count(), Users = dcount(UserPrincipalName), SampleUsers = make_set(UserPrincipalName, 10) by IPAddress | where Failures >= threshold and Users >= 5 ``` **Sample output:** ``` IPAddress Failures Users SampleUsers 185.220.101.34 47 12 ["jdoe","admin","root",...] ``` **Tuning:** keep `Users >= 5` — single-user retries (high `Failures`, low `Users`) is forgotten-password noise. --- ### 22. Impossible-travel via haversine ```kql let speed_threshold_kmh = 800; let haversine = (lat1:real, lon1:real, lat2:real, lon2:real) { let R = 6371.0; let dLat = (lat2 - lat1) * pi() / 180; let dLon = (lon2 - lon1) * pi() / 180; let a = sin(dLat/2) * sin(dLat/2) + cos(lat1 * pi() / 180) * cos(lat2 * pi() / 180) * sin(dLon/2) * sin(dLon/2); R * 2 * atan2(sqrt(a), sqrt(1 - a)) }; SigninLogs | where ResultType == 0 | where isnotnull(LocationDetails.geoCoordinates) | extend lat = todouble(LocationDetails.geoCoordinates.latitude), lon = todouble(LocationDetails.geoCoordinates.longitude), country = tostring(LocationDetails.countryOrRegion) | order by UserPrincipalName asc, TimeGenerated asc | extend prevTime = prev(TimeGenerated, 1), prevLat = prev(lat, 1), prevLon = prev(lon, 1), prevUser = prev(UserPrincipalName, 1), prevCountry = prev(country, 1) | where UserPrincipalName == prevUser and country != prevCountry | extend hours = (TimeGenerated - prevTime) / 1h | where hours > 0 and hours < 24 | extend dist = haversine(prevLat, prevLon, lat, lon) | extend speed = dist / hours | where speed > speed_threshold_kmh | project TimeGenerated, UserPrincipalName, prevCountry, country, hours, dist, speed ``` **Sample output:** ``` TimeGenerated UserPrincipalName prevCountry country hours dist speed 2026-05-23T09:11Z jdoe@yourdomain.com US Russia 0.27 9100 33700 ``` **Tuning:** for global VPN providers, exclude their egress IPs in a separate `let badIPs = ...` step. --- ### 23. LSASS access (Sysmon EID 10) ```kql DeviceEvents | where ActionType == "OpenProcessApiCall" | extend props = parse_json(AdditionalFields) | where tostring(props.TargetFileName) endswith "lsass.exe" | where tostring(props.GrantedAccess) in ("0x1410", "0x1010", "0x1438", "0x143a") | where InitiatingProcessFileName !in ( "MsMpEng.exe","MsSense.exe","WmiPrvSE.exe","CSFalconService.exe","SentinelAgent.exe" ) | project Timestamp, DeviceName, AccountName, InitiatingProcessFileName, GrantedAccess = tostring(props.GrantedAccess) ``` **Sample output:** ``` Timestamp DeviceName AccountName InitiatingProcessFileName GrantedAccess ... WIN10-LAB-02 jdoe update.exe 0x1410 ``` **Tuning:** for stronger signal, join to `DeviceFileEvents` to require an unsigned source binary. --- ### 24. PowerShell -EncodedCommand ```kql DeviceProcessEvents | where FileName in ("powershell.exe", "pwsh.exe") | where ProcessCommandLine matches regex @"\s-(?:e|ec|enc|encodedcommand)\s+[A-Za-z0-9+/=]{16,}" | extend b64 = extract(@"(?i)\s-(?:enc|encodedcommand)\s+([A-Za-z0-9+/=]+)", 1, ProcessCommandLine) | extend decoded = base64_decode_tostring(b64) | where decoded has_any("IEX", "DownloadString", "FromBase64String") | project Timestamp, DeviceName, decoded ``` **Sample output:** ``` Timestamp DeviceName decoded ... WIN10-LAB-02 $s=New-Object Net.WebClient;$s.DownloadString('http://... ``` **Tuning:** `has_any` is much faster than chained `contains` — Kusto indexes it natively. --- ### 25. New local admin (4720 + 4732 join) ```kql let creates = SecurityEvent | where EventID == 4720 | project CreateTime = TimeGenerated, Computer, TargetUser = TargetUserName; let adds = SecurityEvent | where EventID == 4732 and TargetSid == "S-1-5-32-544" | project AddTime = TimeGenerated, Computer, AddedUser = MemberName; creates | join kind=inner (adds) on Computer | where TargetUser == extract(@"\\(.+)$", 1, AddedUser) | where (AddTime - CreateTime) between (0s .. 5m) | project CreateTime, AddTime, Computer, TargetUser ``` **Sample output:** ``` CreateTime AddTime Computer TargetUser 2026-05-23T09:11Z 2026-05-23T09:11Z FILESERVER-04 aetest_4823 ``` **Tuning:** for domain-level: same pattern with 4728 (added to global group) and SID ending in `-512` (Domain Admins). --- ### 26. Lateral SMB by user + host pair ```kql DeviceLogonEvents | where Timestamp > ago(1d) | where LogonType == "Network" and ActionType == "LogonSuccess" | summarize HostCount = dcount(DeviceName), Hosts = make_set(DeviceName, 10) by AccountName | where HostCount > 10 ``` **Sample output:** ``` AccountName HostCount Hosts admin 23 ["fileserver01","fileserver02","web01",...] ``` **Tuning:** allowlist IT admin accounts via `where AccountName !in (knownAdmins)`. --- ### 27. Off-hours admin sign-ins ```kql let admin_users = IdentityInfo | where AssignedRoles has_any("Global Administrator", "Security Administrator", "Privileged Role Administrator") | summarize by AccountUPN; SigninLogs | where TimeGenerated > ago(7d) | where ResultType == 0 | where UserPrincipalName in (admin_users) | where hourofday(TimeGenerated) !between (7 .. 19) | project TimeGenerated, UserPrincipalName, IPAddress, AppDisplayName ``` **Sample output:** ``` TimeGenerated UserPrincipalName IPAddress AppDisplayName 2026-05-23T03:11Z admin@corp.com 185.220.101.34 Office 365 Exchange Online ``` **Tuning:** convert to user's local time using `bag_extract` on `LocationDetails` for accurate "off-hours." --- ### 28. WMI persistence creation triplet ```kql Event | where Source == "Microsoft-Windows-Sysmon" | where EventID in (19, 20, 21) | summarize EventIDs = make_set(EventID), Times = make_set(TimeGenerated) by Computer | where set_has_element(EventIDs, 19) and set_has_element(EventIDs, 20) and set_has_element(EventIDs, 21) ``` **Sample output:** ``` Computer EventIDs Times WIN10-LAB-02 [19, 20, 21] [...] ``` **Tuning:** add time-window: only fire when all three EIDs occur within 5 minutes of each other. --- ### 29. Mailbox forwarding rule created ```kql OfficeActivity | where Operation in ("New-InboxRule", "Set-InboxRule") | extend params = parse_json(Parameters) | extend forwardTo = tostring(params.ForwardTo) | where isnotempty(forwardTo) | where forwardTo matches regex @"@(?!yourdomain\.com|onmicrosoft\.com)[^,]+" | project TimeGenerated, UserId, ClientIP, forwardTo ``` **Sample output:** ``` TimeGenerated UserId ClientIP forwardTo 2026-05-23T11:09Z cfo@corp.com 185.220.101.34 hr-archive@protonmail.com ``` **Tuning:** the regex's negative lookahead is the allow-list — your tenant domains. Adjust per-org. --- ### 30. OAuth high-privilege consent grant ```kql let highPriv = dynamic([ "mail.read","mail.readwrite","mail.send", "files.read.all","files.readwrite.all","offline_access", "directory.read.all","directory.readwrite.all" ]); AuditLogs | where OperationName == "Consent to application" | extend tr = TargetResources[0] | mv-expand mod = todynamic(tr.modifiedProperties) | where tostring(mod.displayName) == "ConsentAction.Permissions" | extend scopes = extract_all(@"Scope:\s*([\w\.]+)", tolower(tostring(mod.newValue))) | extend matched = set_intersect(scopes, highPriv) | where array_length(matched) > 0 | project TimeGenerated, ConsentingUser = tostring(InitiatedBy.user.userPrincipalName), appName = tostring(tr.displayName), matched ``` **Sample output:** ``` TimeGenerated ConsentingUser appName matched 2026-05-23T11:09Z jdoe@corp.com Document Signer Pro ["mail.read","offline_access"] ``` **Tuning:** combine with app age — newly-registered apps (< 7 days) consenting to high-priv scopes is the strongest signal. --- ### 31. Office process spawning PowerShell ```kql DeviceProcessEvents | where Timestamp > ago(1d) | where InitiatingProcessFileName in~ ("winword.exe","excel.exe","powerpnt.exe","outlook.exe","acrord32.exe") | where FileName in~ ("powershell.exe","pwsh.exe","cmd.exe","wscript.exe","cscript.exe","mshta.exe") | project Timestamp, DeviceName, InitiatingProcessFileName, FileName, ProcessCommandLine ``` **Sample output:** ``` Timestamp DeviceName InitiatingProcessFileName FileName ProcessCommandLine ... WIN10-LAB-02 winword.exe powershell.exe -enc ... ``` **Tuning:** treat every hit as P0. Office parent + scripting child is overwhelmingly malicious. --- ### 32. Rundll32 with suspicious parent ```kql DeviceProcessEvents | where FileName =~ "rundll32.exe" | where InitiatingProcessFileName !in~ ("explorer.exe","services.exe","svchost.exe","msiexec.exe","taskhostw.exe","dllhost.exe","control.exe") | project Timestamp, DeviceName, InitiatingProcessFileName, ProcessCommandLine ``` **Sample output:** ``` Timestamp DeviceName InitiatingProcessFileName ProcessCommandLine ... WIN10-LAB-02 winword.exe rundll32.exe shell32.dll,Control_RunDLL ... ``` **Tuning:** add `ProcessCommandLine has_any` filter for the LOLBAS patterns: `"javascript:"`, `",RunHTMLApplication"`, `",#1"`. --- ### 33. Beacon detection via interval regularity ```kql DeviceNetworkEvents | where Timestamp > ago(24h) | where ActionType == "ConnectionSuccess" | summarize Times = make_list(Timestamp) by DeviceName, RemoteIP, RemotePort | extend Count = array_length(Times) | where Count >= 10 | extend Gaps = array_iff(array_slice(Times, 1, Count-1) > array_slice(Times, 0, Count-2), bin(Times, 1s)) // Coefficient of variation | extend gaps = mv-apply prev = Times to typeof(datetime) on (extend g = Times - prev | project g) // (KQL doesn't have an easy CV — use a Python plug-in or a workbook for this) | project DeviceName, RemoteIP, RemotePort, Count ``` **Tuning:** beacon detection in pure KQL is awkward. Better: export to ADX with a `series_decompose` analysis, or use the Sentinel ML connector. The version above is a starting point for the columns you need. --- ### 34. DGA via subdomain entropy ```kql let shannon = (s:string) { let chars = extract_all(@".", s); let total = array_length(chars); let freqs = chars | mv-expand c = chars | summarize cnt = count() by tostring(c) | extend p = todouble(cnt) / total; -(freqs | extend e = p * log2(p) | summarize sum(e)) }; DnsEvents | where QueryType == "TXT" | extend label = tostring(split(Name, ".")[0]) | where strlen(label) >= 12 // Shannon-entropy implementation above is illustrative; in practice use a workbook function | project TimeGenerated, ClientIP, Name, label ``` **Tuning:** KQL doesn't have a built-in entropy function. The cleanest practical pattern is to compute entropy in the ingestion pipeline and store as a column — then filter via `where shannon_entropy > 3.7`. --- ### 35. New cloud admin role assignment ```kql AuditLogs | where TimeGenerated > ago(7d) | where OperationName == "Add member to role" | extend role = tostring(TargetResources[0].displayName) | where role in ("Global Administrator", "Privileged Role Administrator", "Security Administrator") | project TimeGenerated, role, AssignedTo = tostring(TargetResources[0].userPrincipalName), AssignedBy = tostring(InitiatedBy.user.userPrincipalName) ``` **Sample output:** ``` TimeGenerated role AssignedTo AssignedBy 2026-05-23T11:09Z Global Administrator jdoe@corp.com admin@corp.com ``` **Tuning:** Privileged Identity Management (PIM) eligible-assignments fire a *different* operation — `Add eligible member to role`. Add both to the filter for full coverage. --- ## Automation ### 36. Save query as analytic rule In the Sentinel portal: **Analytics → Create → Scheduled query rule**. Use this query body: ```kql SigninLogs | where ResultType != 0 | summarize Failures = count(), Users = dcount(UserPrincipalName) by IPAddress | where Failures >= 10 and Users >= 5 ``` Settings: - Frequency: 15 min - Period: 1h - Trigger: > 0 - Severity: Medium **Tuning:** see `detection-infra/sentinel/modules/analytic_rule` for the Terraform module that automates this — much easier than the portal for many rules. --- ### 37. Entity mappings for incident graph Inside the analytic-rule creation flow, pin query columns to entity types: | Entity | Identifier | Column | |---|---|---| | Account | Name | `UserPrincipalName` | | Host | HostName | `Computer` | | IP | Address | `IPAddress` | The investigation graph wires these up automatically. Without mappings, the graph shows raw rows; with them, you can pivot. **Tuning:** for `Account.AadUserId` use the `Identity` column from SigninLogs — gives you the Entra GUID, which links to IdentityInfo. --- ### 38. Create a watchlist via externaldata Watchlists are higher-performance than externaldata for analytic rules (they're cached + indexed): 1. Upload a CSV via **Sentinel → Watchlists → New**. 2. Reference as: ```kql let badIPs = _GetWatchlist('BadIPs') | project IPAddress; SigninLogs | where IPAddress in (badIPs) ``` **Tuning:** keep watchlist row count under 10k for snappy lookups. For larger lists (threat-intel feeds with millions of rows), use the `ThreatIntelligenceIndicator` table. --- ### 39. Schedule a hunting query In **Sentinel → Hunting → Queries**, save with a category. Hunts are point-in-time exploratory queries (not analytic rules) — but they can be promoted to rules if a hunt becomes useful. **Tuning:** add `bookmark` capability — when a hunt finds something, click "Add bookmark" to keep the result tied to an investigation. --- ### 40. Ingest custom logs via Data Collector To land custom logs in a `_CL` table: ```python # Python sending to Sentinel import requests, json, base64, hmac, hashlib, datetime workspace_id = "your-guid" shared_key = "your-key" log_type = "PyDetection" body = json.dumps([{"rule_id":"R1","host":"lab"}]) date = datetime.datetime.utcnow().strftime("%a, %d %b %Y %H:%M:%S GMT") string_to_sign = f"POST\n{len(body)}\napplication/json\nx-ms-date:{date}\n/api/logs" sig = base64.b64encode(hmac.new(base64.b64decode(shared_key), string_to_sign.encode(), hashlib.sha256).digest()).decode() requests.post( f"https://{workspace_id}.ods.opinsights.azure.com/api/logs?api-version=2016-04-01", headers={"Content-Type":"application/json", "Authorization":f"SharedKey {workspace_id}:{sig}", "Log-Type":log_type, "x-ms-date":date}, data=body) ``` Then query: `PyDetection_CL | take 10` **Tuning:** Data Collector API is being deprecated for new workspaces in favor of the DCR-based Logs Ingestion API. New deployments should target the latter. --- ### 41. Cross-workspace query ```kql workspace("law-soc-prod").SigninLogs | union workspace("law-soc-staging").SigninLogs | where ResultType != 0 | summarize count() by Workspace = "soc-prod-vs-staging" ``` **Tuning:** cross-workspace queries are billed against the *querying* workspace, but RBAC is enforced per source workspace. Useful for federated SOC setups. --- ### 42. Define a function (saved KQL helper) Functions are saved query snippets you can call as if they were tables: ```kql .create-or-alter function HighSevSignins() { SigninLogs | where ResultType != 0 | where RiskLevelAggregated in ("high", "medium") } ``` Then in any query: ```kql HighSevSignins() | where TimeGenerated > ago(1h) ``` **Tuning:** functions can take parameters: `.create-or-alter function HighSevSignins(lookback:timespan)`. The function approach is *the* DRY tool in KQL — use it whenever you find yourself copy-pasting query fragments. --- ### 43. Materialized view for high-volume table ```kql .create-or-alter materialized-view DailyFailedSignins on table SigninLogs { SigninLogs | where ResultType != 0 | summarize FailedCount = count() by bin(TimeGenerated, 1d), IPAddress } ``` Then: `DailyFailedSignins | top 10 by FailedCount desc`. **Tuning:** materialized views compute incrementally — they're cheap to query but expensive to define on extremely high-volume tables. Worth it when you query the same aggregation repeatedly. --- ### 44. ipv4_lookup for ASN enrichment ```kql let asnTable = datatable(StartIP:string, EndIP:string, ASN:int, Description:string) [ "185.220.100.0","185.220.103.255",4224,"Tor exit", "8.8.8.0","8.8.8.255",15169,"Google DNS" ]; SigninLogs | where ResultType != 0 | evaluate ipv4_lookup(asnTable, IPAddress, StartIP, EndIP) | project TimeGenerated, IPAddress, ASN, Description ``` **Sample output:** ``` TimeGenerated IPAddress ASN Description 2026-05-23T09:11Z 185.220.101.34 4224 Tor exit ``` **Tuning:** keep the ASN table in a Watchlist for production — externaldata adds latency. --- ### 45. GeoIP enrichment with externaldata ```kql let geo = externaldata(network: string, country: string) [@"https://yourdomain.com/blob/geoip.csv"] with (format="csv", ignoreFirstRecord=true); SigninLogs | evaluate ipv4_lookup(geo, IPAddress, network) ``` **Tuning:** for MaxMind GeoLite2 CSV, use `network` as CIDR; `ipv4_lookup` interprets it natively. --- ### 46. Logic App trigger from query result The query *is* the input to Logic Apps. In a Sentinel playbook step ("Run query and list results"): ```kql SigninLogs | where ResultType != 0 | where TimeGenerated > ago(15m) | project TimeGenerated, UserPrincipalName, IPAddress ``` Logic App's subsequent actions iterate over rows. Send Teams message, create JIRA ticket, run `Set-AzureADUser -AccountEnabled $false`, etc. **Tuning:** keep result rows < 10k per playbook run — Logic Apps' "list results" action caps there. --- ### 47. Workbook tile from a query In **Workbooks → Edit → Add query**, paste the query and pick `chart` or `tile` rendering. The query result becomes a dashboard tile that refreshes on the workbook's interval. ```kql SigninLogs | where TimeGenerated > ago(24h) | summarize FailedCount = countif(ResultType != 0), SuccessCount = countif(ResultType == 0) | extend FailRate = 100.0 * FailedCount / (FailedCount + SuccessCount) ``` **Tuning:** workbook parameters (`{lookback}`) let users adjust the time window without editing the query. --- ### 48. Anomaly detection with series_decompose_anomalies ```kql SigninLogs | where ResultType != 0 | make-series count=count() default=0 on TimeGenerated from ago(7d) to now() step 1h by IPAddress | extend (anomalies, score, baseline) = series_decompose_anomalies(count, 1.5) | mv-expand TimeGenerated, count, anomalies, score, baseline | where todouble(anomalies) != 0 | project IPAddress, TimeGenerated, count, score, baseline ``` **Sample output:** ``` IPAddress TimeGenerated count score baseline 185.220.101.34 2026-05-23T09Z 142 8.4 3.1 ``` **Tuning:** `series_decompose_anomalies(input, sensitivity)` — higher sensitivity = more anomalies. Start at 1.5 and adjust. --- ### 49. adx() bridge to Azure Data Explorer If you have additional data in an ADX cluster (often used for high-volume telemetry that's too expensive for Sentinel): ```kql let adxData = adx('cluster.region.kusto.windows.net/database').LongTermLogs | where Timestamp > ago(30d) and EventID == 4625; SigninLogs | union adxData ``` **Tuning:** ADX is roughly 90% cheaper than Sentinel for high-volume telemetry. Many orgs route raw Defender XDR streams to ADX and federate via this pattern. --- ### 50. Compose detection + enrichment in one query ```kql let lookback = 1h; let geoTable = externaldata(network:string, country:string) [@"https://yourdomain.com/geo.csv"] with (format="csv", ignoreFirstRecord=true); let knownVPNs = dynamic(["10.99.0.0/16","192.0.2.0/24"]); SigninLogs | where TimeGenerated > ago(lookback) | where ResultType != 0 | summarize FailedCount = count(), Users = dcount(UserPrincipalName), SampleUsers = make_set(UserPrincipalName, 8) by IPAddress | where FailedCount >= 10 and Users >= 5 | evaluate ipv4_lookup(geoTable, IPAddress, network) | extend IsCorpVPN = iff(IPAddress in (toscalar(knownVPNs)), "yes", "no") | project IPAddress, country, FailedCount, Users, SampleUsers, IsCorpVPN | order by FailedCount desc ``` **Sample output:** ``` IPAddress country FailedCount Users SampleUsers IsCorpVPN 185.220.101.34 Russia 142 12 ["admin","root",...] no ``` **Tuning:** this is the shape of a production analytic rule — threshold + filter + enrichment + entity context, all in one query. Save it as a rule and pipe entity mappings. --- ## License MIT. See [LICENSE](LICENSE).