Lab 18: Threat Hunting with KQL & SPL¶

// KQL queries use double-slash comments
// Tables use PascalCase
TableName
| where TimeGenerated > ago(7d)
| where FieldName == "value"
| summarize count() by FieldName
| order by count_ desc

// Exercise 1.1 — Off-hours authentication events (last 7 days)
SecurityEvent
| where TimeGenerated > ago(7d)
| where EventID == 4624
| where LogonType in (2, 10)  // Interactive and RemoteInteractive
| extend HourOfDay = hourofday(TimeGenerated)
| where HourOfDay < 7 or HourOfDay >= 19
| summarize
    OffHoursLogins = count(),
    DistinctSources = dcount(IpAddress),
    EarliestLogin = min(TimeGenerated),
    LatestLogin = max(TimeGenerated),
    SourceIPs = make_set(IpAddress, 10)
  by Account
| where OffHoursLogins > 2
| order by OffHoursLogins desc

// Exercise 1.2 — New login sources not seen in 30-day baseline
let baseline_period = ago(37d);  // 30 days before hunt window
let hunt_start = ago(7d);
// Build baseline: known source IPs per account
let baseline = SecurityEvent
| where TimeGenerated between (baseline_period .. hunt_start)
| where EventID == 4624
| where LogonType in (2, 3, 10)
| summarize BaselineSources = make_set(IpAddress) by Account;
// Hunt window: find new sources
SecurityEvent
| where TimeGenerated > hunt_start
| where EventID == 4624
| where LogonType in (2, 3, 10)
| join kind=leftouter baseline on Account
| extend IsNewSource = iff(IpAddress !in (BaselineSources), true, false)
| where IsNewSource == true
| summarize
    NewSourceCount = dcount(IpAddress),
    NewSources = make_set(IpAddress, 20),
    LoginCount = count(),
    Workstations = make_set(Computer, 10)
  by Account
| where NewSourceCount > 0
| order by NewSourceCount desc

// Exercise 1.3 — Rare process execution (processes seen < 3 times in 30 days)
let hunt_start = ago(7d);
let rare_threshold = 3;
// Baseline: process frequency per host
let process_baseline = SysmonEvent
| where TimeGenerated between (ago(37d) .. hunt_start)
| where EventID == 1  // Process creation
| summarize BaselineCount = count() by Computer, FileName = tostring(split(Image, "\\")[-1]);
// Hunt: find processes below threshold or never seen
SysmonEvent
| where TimeGenerated > hunt_start
| where EventID == 1
| extend FileName = tostring(split(Image, "\\")[-1])
| join kind=leftouter process_baseline on Computer, FileName
| extend BaselineCount = coalesce(BaselineCount, 0)
| where BaselineCount < rare_threshold
| project
    TimeGenerated,
    Computer,
    User,
    FileName,
    CommandLine,
    ParentImage,
    BaselineCount
| order by TimeGenerated asc

// Exercise 2.1 — Pass-the-Hash indicators
// Key indicators: NTLM auth + KeyLength 0 + LogonType 3 + non-service account
SecurityEvent
| where TimeGenerated > ago(7d)
| where EventID == 4624
| where LogonType == 3
| where AuthenticationPackageName == "NTLM"
| where LogonProcessName == "NtLmSsp"
| where KeyLength == 0
// Exclude machine accounts and known service accounts
| where Account !endswith "$"
| where Account !startswith "svc-"
| summarize
    PtHAttempts = count(),
    TargetHosts = make_set(Computer, 20),
    TargetHostCount = dcount(Computer),
    SourceIPs = make_set(IpAddress, 10),
    FirstSeen = min(TimeGenerated),
    LastSeen = max(TimeGenerated)
  by Account
| where TargetHostCount > 1  // Moving to multiple hosts = lateral movement
| order by TargetHostCount desc

// Exercise 2.2 — RDP pivot chain detection
// Find users who RDP to host A, then RDP FROM host A to host B
let rdp_sessions = SecurityEvent
| where TimeGenerated > ago(7d)
| where EventID == 4624
| where LogonType == 10  // RemoteInteractive = RDP
| project
    TimeGenerated,
    Account,
    SourceIP = IpAddress,
    TargetHost = Computer;
// Self-join: find cases where a target becomes a source
rdp_sessions
| join kind=inner (
    rdp_sessions
    | project
        TimeGenerated2 = TimeGenerated,
        Account,
        SourceIP2 = SourceIP,
        TargetHost2 = TargetHost
) on Account
// The source of the second hop must be in the same subnet as the first target
| where TargetHost != TargetHost2
| where TimeGenerated2 > TimeGenerated
| where datetime_diff('hour', TimeGenerated2, TimeGenerated) < 24
| project
    Account,
    Hop1_Time = TimeGenerated,
    Hop1_Source = SourceIP,
    Hop1_Target = TargetHost,
    Hop2_Time = TimeGenerated2,
    Hop2_Source = SourceIP2,
    Hop2_Target = TargetHost2
| order by Account, Hop1_Time asc

// Exercise 2.3 — Administrative share access from workstations
SecurityEvent
| where TimeGenerated > ago(7d)
| where EventID == 5140  // Network share accessed
| where ShareName in ("\\\\*\\C$", "\\\\*\\ADMIN$", "\\\\*\\IPC$")
| where IpAddress startswith "10.50.1."  // From workstation subnet
   or IpAddress startswith "10.50.50."   // or executive subnet
| summarize
    AccessCount = count(),
    SharesAccessed = make_set(ShareName),
    TargetHosts = make_set(Computer, 10),
    DistinctTargets = dcount(Computer)
  by Account, IpAddress
| where DistinctTargets > 1
| order by DistinctTargets desc

// Exercise 2.4 — Explicit credential usage (runas, credential theft)
SecurityEvent
| where TimeGenerated > ago(7d)
| where EventID == 4648  // Logon with explicit credentials
| where TargetAccount != Account  // Using a DIFFERENT account's creds
| project
    TimeGenerated,
    SourceAccount = Account,
    TargetAccount,
    TargetHost = TargetServerName,
    SourceIP = IpAddress,
    ProcessName
| order by TimeGenerated asc

// Exercise 3.1 — Top talkers to external IPs (connection frequency)
NetworkCommunicationEvents
| where TimeGenerated > ago(7d)
| where DestinationPort in (80, 443)
// Exclude internal-to-internal
| where not(ipv4_is_in_range(RemoteIP, "10.0.0.0/8"))
| where not(ipv4_is_in_range(RemoteIP, "172.16.0.0/12"))
| where not(ipv4_is_in_range(RemoteIP, "192.168.0.0/16"))
| summarize
    ConnectionCount = count(),
    TotalBytesSent = sum(SentBytes),
    TotalBytesReceived = sum(ReceivedBytes),
    AvgBytesSent = avg(SentBytes),
    StdevBytesSent = stdev(SentBytes),
    FirstSeen = min(TimeGenerated),
    LastSeen = max(TimeGenerated),
    DurationHours = datetime_diff('hour', max(TimeGenerated), min(TimeGenerated))
  by LocalIP, RemoteIP, DestinationPort
| where ConnectionCount > 50
| extend ConnectionsPerHour = round(toreal(ConnectionCount) / max_of(DurationHours, 1), 2)
| where ConnectionsPerHour > 10
| order by ConnectionsPerHour desc

// Exercise 3.2 — Beacon jitter analysis using time-series
// Step 1: Get connection times for each src-dst pair
let connection_times = NetworkCommunicationEvents
| where TimeGenerated > ago(7d)
| where DestinationPort in (80, 443)
| where not(ipv4_is_in_range(RemoteIP, "10.0.0.0/8"))
| project TimeGenerated, LocalIP, RemoteIP
| order by LocalIP, RemoteIP, TimeGenerated asc;
// Step 2: Calculate intervals using serialize and prev()
connection_times
| serialize
| extend PrevTime = prev(TimeGenerated, 1),
         PrevLocal = prev(LocalIP, 1),
         PrevRemote = prev(RemoteIP, 1)
| where LocalIP == PrevLocal and RemoteIP == PrevRemote
| extend IntervalSeconds = datetime_diff('second', TimeGenerated, PrevTime)
| where IntervalSeconds > 0 and IntervalSeconds < 3600  // Ignore gaps > 1hr
| summarize
    ConnectionCount = count(),
    MeanInterval = avg(IntervalSeconds),
    StdevInterval = stdev(IntervalSeconds),
    MinInterval = min(IntervalSeconds),
    MaxInterval = max(IntervalSeconds)
  by LocalIP, RemoteIP
| where ConnectionCount > 20
| extend CoefficientOfVariation = round(StdevInterval / MeanInterval, 4)
// Low CoV = highly regular = likely beacon
// Typical beacons: CoV < 0.20 (20% jitter or less)
| where CoefficientOfVariation < 0.20
| extend EstimatedBeaconInterval = strcat(tostring(round(MeanInterval, 0)), "s +/- ",
                                          tostring(round(StdevInterval, 1)), "s")
| project LocalIP, RemoteIP, ConnectionCount, EstimatedBeaconInterval,
          CoefficientOfVariation, MeanInterval, MinInterval, MaxInterval
| order by CoefficientOfVariation asc

// Exercise 3.3 — DNS beaconing detection
// High-frequency queries to a single parent domain with high subdomain entropy
DnsEvents
| where TimeGenerated > ago(7d)
| where QueryType in ("TXT", "CNAME", "MX")  // Commonly abused types
| extend ParentDomain = strcat(
    tostring(split(Name, ".")[-2]), ".",
    tostring(split(Name, ".")[-1]))
| extend SubdomainLabel = tostring(split(Name, ".")[0])
| extend SubdomainLength = strlen(SubdomainLabel)
| summarize
    QueryCount = count(),
    UniqueSubdomains = dcount(SubdomainLabel),
    AvgSubdomainLength = round(avg(SubdomainLength), 1),
    MaxSubdomainLength = max(SubdomainLength),
    SampleQueries = make_set(Name, 5),
    QueryingHosts = make_set(ClientIP, 10)
  by ParentDomain
| where QueryCount > 100
| where UniqueSubdomains > 50
| where AvgSubdomainLength > 8  // Base64 labels tend to be long
| order by QueryCount desc

// Exercise 3.4 — Time-series visualization of beacon traffic
NetworkCommunicationEvents
| where TimeGenerated > ago(2d)
| where LocalIP == "10.50.1.42"
| where RemoteIP == "192.0.2.41"
| make-series ConnectionCount = count()
    on TimeGenerated
    from ago(2d) to now()
    step 5m
| render timechart
// Expected: a flat, consistent line of ~5 connections per 5-min bucket
// (one connection per minute = 5 per 5-min bucket)

// Exercise 4.1 — Outbound data volume anomalies
// Calculate per-host daily upload volume and flag outliers
let daily_uploads = NetworkCommunicationEvents
| where TimeGenerated > ago(7d)
| where not(ipv4_is_in_range(RemoteIP, "10.0.0.0/8"))
| where not(ipv4_is_in_range(RemoteIP, "172.16.0.0/12"))
| where not(ipv4_is_in_range(RemoteIP, "192.168.0.0/16"))
| summarize
    DailyBytesSent = sum(SentBytes),
    ConnectionCount = count(),
    DistinctDestinations = dcount(RemoteIP)
  by LocalIP, Day = bin(TimeGenerated, 1d);
// Calculate baseline stats per host
let host_baselines = daily_uploads
| summarize
    AvgDailyUpload = avg(DailyBytesSent),
    StdDailyUpload = stdev(DailyBytesSent)
  by LocalIP;
// Flag days where upload exceeds baseline by 3+ standard deviations
daily_uploads
| join kind=inner host_baselines on LocalIP
| where StdDailyUpload > 0
| extend ZScore = round((DailyBytesSent - AvgDailyUpload) / StdDailyUpload, 2)
| where ZScore > 3
| extend DailyUploadMB = round(toreal(DailyBytesSent) / 1048576, 2)
| project Day, LocalIP, DailyUploadMB, ZScore,
          ConnectionCount, DistinctDestinations
| order by ZScore desc

// Exercise 4.2 — DNS tunneling volume estimation
DnsEvents
| where TimeGenerated > ago(7d)
| where Name endswith "cdn-analytics.example.com"
| extend SubdomainLabel = tostring(split(Name, ".")[0])
| extend PayloadBytes = strlen(SubdomainLabel) * 6 / 8  // Base64: 6 bits per char
| summarize
    TotalQueries = count(),
    EstimatedExfilBytes = sum(PayloadBytes),
    UniqueLabels = dcount(SubdomainLabel),
    QueriesPerHour = count() / max_of(datetime_diff('hour',
        max(TimeGenerated), min(TimeGenerated)), 1),
    FirstQuery = min(TimeGenerated),
    LastQuery = max(TimeGenerated)
  by ClientIP
| extend EstimatedExfilKB = round(toreal(EstimatedExfilBytes) / 1024, 2)
| project ClientIP, TotalQueries, EstimatedExfilKB,
          UniqueLabels, QueriesPerHour, FirstQuery, LastQuery

// Exercise 4.3 — Upload/Download ratio anomalies (inverted traffic patterns)
NetworkCommunicationEvents
| where TimeGenerated > ago(7d)
| where DestinationPort in (80, 443)
| where not(ipv4_is_in_range(RemoteIP, "10.0.0.0/8"))
| where not(ipv4_is_in_range(RemoteIP, "172.16.0.0/12"))
| where not(ipv4_is_in_range(RemoteIP, "192.168.0.0/16"))
| summarize
    TotalSent = sum(SentBytes),
    TotalReceived = sum(ReceivedBytes),
    ConnectionCount = count()
  by LocalIP, RemoteIP
| where TotalSent > 1048576  // > 1 MB sent
| extend UploadDownloadRatio = round(toreal(TotalSent) / max_of(TotalReceived, 1), 2)
| where UploadDownloadRatio > 5.0  // Sending 5x more than receiving
| extend TotalSentMB = round(toreal(TotalSent) / 1048576, 2),
         TotalReceivedMB = round(toreal(TotalReceived) / 1048576, 2)
| project LocalIP, RemoteIP, TotalSentMB, TotalReceivedMB,
          UploadDownloadRatio, ConnectionCount
| order by UploadDownloadRatio desc

// Exercise 4.4 — Cloud storage exfiltration via proxy logs
CommonSecurityLog
| where TimeGenerated > ago(7d)
| where DeviceVendor == "BlueCoat" or DeviceVendor == "Zscaler"
| where RequestURL has_any (
    "blob.core.windows.net",
    "s3.amazonaws.com",
    "storage.googleapis.com",
    "drive.google.com",
    "dropbox.com",
    "mega.nz",
    "pastebin.com"
)
| where RequestMethod in ("POST", "PUT")
| summarize
    UploadCount = count(),
    TotalBytesSent = sum(SentBytes),
    DistinctURLs = dcount(RequestURL),
    SampleURLs = make_set(RequestURL, 5)
  by SourceIP, DestinationHostName
| where TotalBytesSent > 5242880  // > 5 MB uploaded
| extend TotalSentMB = round(toreal(TotalBytesSent) / 1048576, 2)
| order by TotalSentMB desc