Correlates edge-access events with lateral movement

// EMPERADOR Hunt: Edge access -> internal lateral movement -> archive staging
// Lookback aligned to assessed 5-14 day dwell time
let Lookback = 14d;
let SuspectExternal = dynamic(["vpn-gateway", "checkpoint", "screenconnect", "rdp-brute"]);
let EdgeHits =
    SigninLogs
    | where TimeGenerated > ago(Lookback)
    | where ResultType == 0
    | where AppDisplayName has_any (SuspectExternal) or IPAddress in (externaldata(string)[h"https://ti-feed.example/iocs.csv"] with (format="csv"))
    | summarize FirstSeen=min(TimeGenerated) by UserPrincipalName, IPAddress;
let Lateral =
    SecurityEvent
    | where TimeGenerated > ago(Lookback)
    | where EventID in (4624, 4672, 7045)
    | where LogonType in (3, 10)
    | where Process has_any ("PSEXESVC", "wmiprvse") or ServiceName has_any ("PSEXESVC", "paexec", "remcom")
    | summarize LateralEvents=count(), Hosts=dcount(Computer), FirstLat=min(TimeGenerated) by Account;
let Staging =
    DeviceProcessEvents
    | where TimeGenerated > ago(Lookback)
    | where FileName in~ ("rar.exe","7z.exe","7za.exe")
    | where ProcessCommandLine has_any (".bak", ".sql", ".mdf", "backup", "ProgramData")
    | summarize StagingHosts=dcount(DeviceName), FirstStage=min(TimeGenerated) by InitiatingProcessAccountName;
EdgeHits
| join kind=inner (Lateral) on $left.UserPrincipalName == $right.Account
| join kind=leftouter (Staging) on $left.UserPrincipalName == $right.InitiatingProcessAccountName
| project UserPrincipalName, EdgeIP=IPAddress, FirstSeen, LateralEvents, Hosts, StagingHosts, FirstStage
| where FirstLat > FirstSeen
| sort by FirstSeen asc

Hunting queries are starting points for threat hunting & detection engineering — validate table/column names against your own workspace schema and tune thresholds before turning any of these into a production alert rule.