IEM 已死？从 Kafka 到 UEBA：构建下一代T+0态势感知的技术栈与代码实现，舆情系统中的应用

// 使用 Flink DataStream API
DataStream<LogEvent> stream = env.addSource(new FlinkKafkaConsumer<>("security-logs", ...));

// KeyBy 用户ID，进行状态化处理
DataStream<Alert> alerts = stream
    .keyBy(LogEvent::getUserId)
    // 定义一个“窗口”或“状态机” (CEP - 复杂事件处理)
    .process(new KeyedProcessFunction<String, LogEvent, Alert>() {
        
        // Flink 提供的状态存储 (ValueState)
        private ValueState<Integer> failedLoginCount;
        private ValueState<Long> lastFailedTime;

        @Override
        public void open(Configuration parameters) {
            // 初始化状态
            failedLoginCount = getRuntimeContext().getState(new ValueStateDescriptor<>("failedCount", Integer.class));
            lastFailedTime = getRuntimeContext().getState(new ValueStateDescriptor<>("lastFailedTime", Long.class));
        }

        @Override
        public void processElement(LogEvent log, Context ctx, Collector<Alert> out) throws Exception {
            Integer count = failedLoginCount.value();
            if (count == null) count = 0;

            if (log.getType().equals("LOGIN_FAILED")) {
                count++;
                failedLoginCount.update(count);
                lastFailedTime.update(log.getTimestamp());

                // 注册一个5分钟后的定时器，用于清空状态
                ctx.timerService().registerEventTimeTimer(log.getTimestamp() + 300_000); 
            }

            if (log.getType().equals("LOGIN_SUCCESS") && count >= 3 && log.isRemoteIP()) {
                out.collect(new Alert("HIGH_RISK_LOGIN", log.getUserId()));
                // 清理状态
                failedLoginCount.clear();
                lastFailedTime.clear();
            }
        }
        
        // 定时器触发时，清空状态
        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<Alert> out) throws Exception {
            failedLoginCount.clear();
            lastFailedTime.clear();
        }
    });

from sklearn.ensemble import IsolationForest
import numpy as np

# 1. 模拟一个员工的正常行为数据 (特征工程是关键)
# 特征: [登录时间(小时), 数据下载量(MB), 访问系统数量]
normal_behavior = np.array([
    [9, 10, 3],
    [10, 50, 5],
    [14, 20, 4],
    [17, 5, 2]
])

# 2. 训练模型 (这就是“行为基线”的建立)
# contamination 污染率，即你预估数据中有多大比例是异常的
model = IsolationForest(contamination=0.1, random_state=42)
model.fit(normal_behavior)

# 3. 实时检测新行为
# 行为A: 正常 [11点, 30MB, 4个系统]
behavior_a = np.array([[11, 30, 4]]) 
# 行为B: 异常 [凌晨3点, 10000MB, 20个系统] (10GB)
behavior_b = np.array([[3, 10000, 20]])

# 4. 获得预测结果
# model.predict() 会返回: 1 (正常) 或 -1 (异常)
print(f"行为 A (正常): {model.predict(behavior_a)}") # 输出 [1]
print(f"行为 B (异常): {model.predict(behavior_b)}") # 输出 [-1]

# model.decision_function() 会返回异常分值 (越低越异常)
print(f"行为 A 分值: {model.decision_function(behavior_a)}")
print(f"行为 B 分值: {model.decision_function(behavior_b)}")

---
name: "Dev_Account_Compromise_Response"
trigger:
  type: "Alert"
  source: "UEBA"
  severity: "High"
  tags: ["IsolationForest", "DataExfil"]

steps:
  - name: "Step1_Get_Alert_Details"
    action: "enrichment:get_alert_data"
    params:
      alert_id: "{{ trigger.alert_id }}"

  - name: "Step2_Isolate_Instance"
    description: "调用阿里云API，将EC2实例移入隔离安全组"
    action: "aliyun:ecs:isolate_instance"
    params:
      instance_id: "{{ Step1_Get_Alert_Details.instance_id }}"
      security_group_id: "sg-quarantine-group" # 隔离安全组

  - name: "Step3_Disable_User_Account"
    description: "调用RAM API，冻结该用户的AccessKey"
    action: "aliyun:ram:disable_user"
    params:
      user_name: "{{ Step1_Get_Alert_Details.user_name }}"

  - name: "Step4_Create_ITSM_Ticket"
    action: "servicenow:create_ticket"
    params:
      title: "P0 - Account Compromise: {{ trigger.user_name }}"
      description: "已自动隔离实例和RAM账户，请立即介入。"

  - name: "Step5_Notify_SecOps_Chat"
    action: "webhook:post_to_dingtalk"
    params:
      message: "🔥 P0 告警: {{ trigger.user_name }} 账号疑似被盗，已自动执行隔离！"