📚 课程目录第2章日志采集与标准化

日志采集与标准化：从源头搞定数据质量

日志异常检测这件事，说白了就是「垃圾进，垃圾出」。你算法再牛，模型再炫，如果日志采集不全、格式混乱，后面全是白搭。我在项目中见过太多团队，花三个月调模型，最后发现是采集端漏掉了关键字段——那种感觉，嗯，挺崩溃的。

这一章我们就把地基打牢。聊三个核心问题：日志怎么采？采完怎么统一格式？以及，怎么用 Python 把 Windows 事件日志老老实实抓下来。

一、日志采集架构：三种主流姿势

我个人习惯把采集方式分成三类。你根据场景选，没有银弹。

采集方式	原理	适用场景	我踩过的坑
Agent 模式	在目标主机上装一个采集程序，主动拉或被动推日志	服务器、容器环境、需要细粒度控制的场景	Agent 版本不一致导致字段解析失败，曾经排查了两天
Syslog 模式	网络设备、Linux 系统通过 UDP/TCP 514 端口发送日志	网络设备、防火墙、交换机、部分 Linux 服务	UDP 丢包严重，关键告警没收到，后来全切了 TCP
API 模式	调用云平台或应用的 REST API 拉取日志	云服务（AWS CloudTrail、Azure）、SaaS 应用	API 限频，日志量大的时候要设计重试和分页逻辑

核心原则：能 Agent 就 Agent，不能 Agent 再考虑 Syslog，API 作为兜底。Agent 能拿到最全的上下文，Syslog 容易丢字段，API 受限于厂商接口能力。

二、日志标准化：CEF、LEEF、JSON 怎么选？

日志格式不统一，是安全运维的噩梦。你想想看，Windows 事件日志是 EVTX，Linux 是 syslog，防火墙是逗号分隔，Web 服务器是 combined 格式…… 解析规则写到你手软。

所以业界搞了几个标准化格式。我重点讲三个：

1. CEF（Common Event Format）

ArcSight 推的格式，现在 SIEM 圈基本都认。结构是：

CEF:版本|设备厂商|设备产品|设备版本|事件ID|事件名称|严重级别|扩展字段

举个例子：

CEF:0|Microsoft|Windows|10.0.17763|4625|登录失败|5|src=192.168.1.100 dst=10.0.0.1 duser=admin

扩展字段用 key=value 对，空格分隔。注意值里如果有空格，得用反斜杠转义。我曾经因为这个解析 bug，漏掉了 30% 的登录失败日志。

2. LEEF（Log Event Extended Format）

IBM QRadar 的格式。长得像这样：

LEEF:1.0|Microsoft|Windows|10.0|4625|严重级别=5|src=192.168.1.100|dst=10.0.0.1|usrName=admin

跟 CEF 的区别？LEEF 用管道符分隔字段，扩展字段也是 key=value。我个人觉得 LEEF 的解析更稳定，因为管道符在日志里很少出现，不容易误分割。

3. JSON

现在最流行的格式。没有之一。结构清晰，解析简单，几乎所有编程语言都有原生支持。

{
  "event_id": 4625,
  "event_name": "登录失败",
  "severity": 5,
  "src_ip": "192.168.1.100",
  "dst_ip": "10.0.0.1",
  "user": "admin",
  "timestamp": "2025-01-15T08:30:00Z"
}

我的建议：新项目直接上 JSON。CEF 和 LEEF 是历史产物，除非你对接的 SIEM 强制要求，否则别给自己找麻烦。JSON 可读性好，扩展方便，嵌套结构也能表达复杂关系。

三、实战：用 Python 采集 Windows 事件日志

好，理论讲完了，咱们动手。Windows 事件日志是安全审计的重头戏——登录成功/失败、账号创建、权限变更、服务启停，全在里面。

Python 采集 Windows 事件日志，我推荐两个库：

pywin32：底层调用 Windows API，功能全，但代码啰嗦
python-evtx：直接解析 EVTX 文件，适合离线分析

这里我用 pywin32 演示在线采集。为什么？因为实时性要求高的场景，你不能等日志文件轮转。

代码示例：采集安全日志中的登录事件

import win32evtlog
import win32evtlogutil
import datetime

def collect_security_events(log_name="Security", event_ids=[4624, 4625], max_events=100):
    """
    采集 Windows 安全日志中的登录事件
    4624 = 登录成功, 4625 = 登录失败
    """
    hand = win32evtlog.OpenEventLog(None, log_name)
    flags = win32evtlog.EVENTLOG_BACKWARDS_READ | win32evtlog.EVENTLOG_SEQUENTIAL_READ
    
    events = []
    total = 0
    
    while total < max_events:
        records = win32evtlog.ReadEventLog(hand, flags, 0)
        if not records:
            break
        
        for record in records:
            if record.EventID & 0xFFFF in event_ids:
                event_data = {
                    "event_id": record.EventID & 0xFFFF,
                    "time_generated": record.TimeGenerated.Format(),
                    "source": record.SourceName,
                    "computer": record.ComputerName,
                    "message": win32evtlogutil.SafeFormatMessage(record, log_name)
                }
                events.append(event_data)
                total += 1
                if total >= max_events:
                    break
    
    win32evtlog.CloseEventLog(hand)
    return events

# 使用示例
if __name__ == "__main__":
    login_events = collect_security_events(event_ids=[4624, 4625], max_events=50)
    for evt in login_events:
        print(f"[{evt['time_generated']}] EventID={evt['event_id']} - {evt['computer']}")
        # 这里可以输出 JSON 格式，方便后续处理
        # import json; print(json.dumps(evt, ensure_ascii=False))

注意：采集安全日志需要管理员权限。我曾经在客户现场忘了提权，脚本跑了一小时一条日志都没抓到，排查半天才发现是权限问题。另外，EventID 在 32 位系统上可能超过 16 位范围，记得用 record.EventID & 0xFFFF 取低 16 位。

标准化输出：把 Windows 日志转成 JSON

采集到原始数据后，我习惯立刻转成标准 JSON 格式。这样下游不管是写 ES、写 Kafka 还是直接分析，都方便。

import json

def normalize_windows_event(raw_event):
    """
    将 Windows 事件日志标准化为 JSON 格式
    """
    normalized = {
        "event_source": "windows_security",
        "event_id": raw_event["event_id"],
        "event_name": "登录成功" if raw_event["event_id"] == 4624 else "登录失败",
        "timestamp": raw_event["time_generated"],
        "host": raw_event["computer"],
        "raw_message": raw_event["message"],
        "severity": 3 if raw_event["event_id"] == 4624 else 5,
        "tags": ["windows", "security", "login"]
    }
    
    # 从消息中提取关键字段（这里简化处理，实际需要正则匹配）
    if "登录失败" in raw_event["message"]:
        normalized["reason"] = "密码错误或账号不存在"
    
    return normalized

# 批量标准化
raw_events = collect_security_events(event_ids=[4624, 4625], max_events=10)
json_events = [normalize_windows_event(e) for e in raw_events]

# 输出 JSON 数组
print(json.dumps(json_events, indent=2, ensure_ascii=False))

四、整体架构图：从采集到标准化的完整链路

下面这张图，是我做项目时画给团队看的。你照着搭，基本不会出大问题。

实战经验：我在做金融客户项目时，日志量每天 50TB。标准化层用了 Fluentd + 自定义 Python 插件，把 CEF 转 JSON。核心逻辑就一条——保留原始日志全文，标准化字段只做索引用。万一标准化逻辑有 bug，还能从原始日志回溯。

五、避坑指南：我踩过的三个坑

时间戳时区问题：Windows 事件日志默认本地时间，Syslog 是 UTC。不统一时区，关联分析全乱套。我建议统一转成 UTC+0 的 ISO 8601 格式。
日志截断：Windows 事件日志单条消息有 32KB 限制。超过的部分会被截断。我曾经因为这个漏掉了攻击者的完整命令行。解决方案是采集时检查消息长度，截断时记录标记。
性能问题：用 pywin32 实时采集时，ReadEventLog 的轮询间隔别设太短。我一般设 5 秒，配合 Windows 事件转发（WEF）做分布式采集，避免单点压力。

嗯，这一章的内容就到这。日志采集和标准化是安全运维的「脏活累活」，但也是最有价值的部分。你把这一步做扎实了，后面的异常检测才有意义。