WSO2 Siddhi：统计符合条件事件数及登录暴力破解检测问题

问题分析 #

你当前的代码里，count()的行为是符合Siddhi默认聚合逻辑的：每当有新事件进入窗口时，都会计算当前窗口内的总事件数并输出一条记录，所以attempts会从1开始逐步递增。这就导致#login_attempts表中会生成多条记录，每条的attempts是累计到当前的数值，而非一次性统计窗口内总数并触发告警，和你预期的效果不符。

解决方案：正确统计窗口内登录尝试次数并触发告警 #

我们可以简化逻辑，直接在分区内对窗口事件做聚合，当计数超过阈值时生成告警；如果需要保留登录尝试明细，再单独插入到明细表里：

1. 仅触发阈值告警（推荐）

partition with (Target_IP4 of I_Events) begin
  -- 统计5秒窗口内当前Target_IP4的登录尝试次数
  from I_Events[Category == 'Attempt.Login']#window.time(5 sec)
  select 
    Target_IP4,
    count() as attempts,
    max(meta_EventTime) as latest_attempt_time,
    collect(Source_IP4) as source_ips, -- 可选：收集所有来源IP
    collect(correlation__id) as correlation_ids -- 可选：收集所有关联ID
  having attempts > 20 -- 仅当次数超过20时输出告警
  insert into login_brute_force_alert;
end;

说明：

• having attempts > 20 会过滤掉次数未达阈值的聚合结果，只有当窗口内登录尝试超过20次时，才会生成告警事件。
• 每次窗口内的事件数量变化时（比如从20到21、21到22），都会输出一条告警。如果希望仅在首次超过阈值时触发，可以添加状态变量跟踪已告警IP，比如用Siddhi的state表记录状态，窗口过期后清除。

2. 保留登录尝试明细+触发告警

如果你需要留存所有登录尝试的明细，同时触发告警，可以拆分两个查询：

partition with (Target_IP4 of I_Events) begin
  -- 插入所有登录尝试明细到临时表
  from I_Events[Category == 'Attempt.Login']
  insert into login_attempts;

  -- 统计5秒窗口内的尝试次数，超过阈值则告警
  from login_attempts#window.time(5 sec)
  select 
    Target_IP4,
    count() as attempts,
    max(meta_EventTime) as latest_attempt_time
  having attempts > 20
  insert into login_brute_force_alert;
end;

关于“选中流中所有元素”的问题 #

如果需要在某个条件满足时（比如attempts > 20）获取当前窗口内的所有事件，有两种常用方式：

方式1：用collect()聚合所有事件字段

partition with (Target_IP4 of I_Events) begin
  from I_Events[Category == 'Attempt.Login']#window.time(5 sec)
  select 
    Target_IP4,
    count() as attempts,
    collect(I_Events) as all_attempt_events -- 收集窗口内所有登录事件
  having attempts > 20
  insert into login_brute_force_alert_with_details;
end;

collect(I_Events)会把窗口内的所有事件打包成一个列表，后续处理时可以解析这个列表获取每个事件的明细。

方式2：用Join关联告警与明细事件

如果需要单独输出每个触发阈值的明细事件，可以用join将告警信号与窗口内的事件关联：

-- 先定义一个触发告警的流
partition with (Target_IP4 of I_Events) begin
  from I_Events[Category == 'Attempt.Login']#window.time(5 sec)
  select Target_IP4, count() as attempts
  having attempts > 20
  insert into alert_trigger;
end;

-- 关联告警触发信号与原始登录事件，输出所有明细
from alert_trigger as t
join I_Events[Category == 'Attempt.Login']#window.time(5 sec) as e
on t.Target_IP4 == e.Target_IP4
select e.*, t.attempts as total_attempts
insert into brute_force_attempt_details;

额外优化建议 #

• 注意你代码里的Target_Hostnmae拼写错误，应该是Target_Hostname，避免字段映射失败。
• 如果要避免同一Target_IP4在5秒内重复触发告警，可以用#window.length(1)配合distinct，或者用state表记录已告警IP的状态，窗口过期后清除。

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