如何在KQL中为Defender XDR自定义检测规则获取有效ReportId

要实现你的需求，无需类似PowerShell的foreach遍历，直接用KQL的聚合+关联操作就能完成。核心思路是先识别出所有违规的发件人及其对应的24小时滚动窗口，再关联原始EmailEvent数据获取该窗口内的任意一个有效ReportId。

方法一：固定24小时窗口检测（适合快速排查） #

如果可以接受按自然24小时窗口（如00:00-24:00）排查超限情况，用以下查询：

let recipient_threshold = 2000;
let lookback_days = 30d;

// 第一步：筛选过去30天内24小时窗口内外部收件人超标的记录
let violation_windows = EmailEvent
| where TimeGenerated >= ago(lookback_days)
| where RecipientDomain != SenderDomain  // 过滤外部收件人
| extend WindowStart = bin(TimeGenerated, 24h)  // 按24小时划分固定窗口
| extend WindowEnd = WindowStart + 24h
| summarize ExternalRecipientCount = dcount(RecipientEmailAddress) 
    by SenderUPN, WindowStart, WindowEnd
| where ExternalRecipientCount > recipient_threshold;

// 第二步：关联原始邮件事件，获取每个违规窗口的任意ReportId
violation_windows
| join kind=inner (
    EmailEvent
    | where TimeGenerated >= ago(lookback_days)
    | where RecipientDomain != SenderDomain
) on SenderUPN
| where TimeGenerated between (WindowStart .. WindowEnd)
| summarize 
    ExternalRecipientCount = any(ExternalRecipientCount),
    ReportId = take_any(ReportId)  // 取窗口内任意一个有效ReportId
by SenderUPN, WindowStart, WindowEnd
| sort by WindowEnd desc

方法二：严格滚动24小时窗口检测（符合Exchange新规则定义） #

如果需要严格匹配Exchange的连续24小时滚动窗口（即任意时间点往前推24小时内的收件人数量），用scan操作符实现：

let recipient_threshold = 2000;
let lookback_window = 24h;
let lookback_days = 30d;

// 预处理外部邮件事件，按发件人+时间排序
let email_events = EmailEvent
| where TimeGenerated >= ago(lookback_days)
| where RecipientDomain != SenderDomain
| sort by SenderUPN, TimeGenerated asc;

// 用scan检测滚动窗口内的超限情况
let violation_windows = email_events
| scan declare (window_recipients: dynamic=dynamic([]), is_violation: bool=false) with (
    // 同一发件人，且当前邮件在过去24小时窗口内
    step track_window:
        SenderUPN == prev(SenderUPN) and TimeGenerated <= prev(TimeGenerated) + lookback_window =>
            window_recipients = array_concat(prev(window_recipients), dynamic([RecipientEmailAddress])),
            is_violation = dcount(window_recipients) > recipient_threshold;
    // 新的发件人或超出24小时窗口，重置跟踪
    step reset_window:
        true =>
            window_recipients = dynamic([RecipientEmailAddress]),
            is_violation = false;
)
| where is_violation
| summarize 
    WindowStart = min(TimeGenerated) - lookback_window,
    WindowEnd = max(TimeGenerated),
    ExternalRecipientCount = dcount(RecipientEmailAddress)
by SenderUPN;

// 关联获取违规窗口的ReportId
violation_windows
| join kind=inner email_events on SenderUPN
| where TimeGenerated between (WindowStart .. WindowEnd)
| summarize 
    ExternalRecipientCount = any(ExternalRecipientCount),
    ReportId = take_any(ReportId)
by SenderUPN, WindowStart, WindowEnd
| sort by WindowEnd desc

关键说明 #

• take_any(ReportId)会返回违规窗口内的任意一个有效ReportId，完全满足Defender XDR自定义检测规则对ReportId的要求；
• dcount(RecipientEmailAddress)确保统计的是唯一外部收件人数量，和Exchange的速率限制逻辑一致；
• 两种方法都先聚合出违规窗口，再关联原始数据，避免了重复扫描全量数据，性能更优。

内容的提问来源于stack exchange，提问作者JTB