📚 课程目录第4章日志格式解析

4 日志格式解析：JSON、XML、CEF、LEEF 等

说实话，我最早接手审计日志的时候，最头疼的还真不是规则怎么写，而是日志长什么样。同一个用户登录事件，在防火墙里是 CEF 格式，到了 Windows 事件日志里变成 XML，发给 SIEM 时又被转成 JSON。你想想看，如果解析器连格式都认不全，后面关联分析就是空中楼阁。

这一章我就把四种主流日志格式掰开了讲——JSON、XML、CEF、LEEF。我会告诉你每种格式长什么样、坑在哪、解析时要注意什么。都是我自己踩过的坑，听完你至少能少走半年弯路。

4.1 JSON —— 最通用的“瑞士军刀”

JSON 是现阶段的默认选择。几乎所有的 REST API 都返回 JSON，现代 SIEM 产品也首选 JSON 作为输入格式。它结构清晰，键值对一一对应，还支持嵌套。

我的经验： 在某银行审计项目里，我每天要处理上亿条 AWS CloudTrail 日志，全是 JSON。一开始我直接用 json.loads() 一股脑全读了，结果发现有的字段名前后大小写不一致——比如 eventTime 和 eventtime 混着来。从那以后，我所有 JSON 解析器第一件事就是字段名规范化：一律转小写。

实战中常见的 JSON 日志结构：

{
  "timestamp": "2025-03-21T14:23:10Z",
  "eventType": "Login",
  "userId": "wangpeng",
  "sourceIP": "10.10.1.100",
  "targetResource": "sftp-server",
  "action": "allow",
  "details": {
    "authMethod": "password",
    "sessionId": "abc123"
  }
}

解析要点：

时间戳标准化：ISO 8601 格式居多，也有带时区的，建议统一转成 UTC + 毫秒时间戳。
嵌套字段展开：比如 details.authMethod 到底要不要作为独立字段？我习惯在解析时拍平（flatten），方便后续 SQL 查询。
特殊字符转义：日志里如果包含反斜杠、双引号，JSON 解析器会自动处理，但有些生产日志会因编码问题导致解析失败——我记得有一次因为一个用户名字符串里混了个非法控制字符，整条日志就被跳过了。所以一定要加上异常处理。

小技巧： 解析前先用 json.loads() 验证格式是否合法；如果发现大量解析错误，先检查日志源是否开启了转义（比如 \\” 误写）。

4.2 XML —— 冗长但严谨的“老派绅士”

XML 审计日志在金融、政府系统中很常见，特别是那些依赖 XSD 验证的合规场景。它的优点是可以定义严格的 schema，字段类型、取值范围都能约束；缺点嘛——太啰嗦。一条 JSON 50 字节能搞定的，XML 可能 500 字节。

举个例子（Windows Security Event Log 导出）：

<Event xmlns="http://schemas.microsoft.com/win/2004/08/events/event">
  <System>
    <EventID>4624</EventID>
    <TimeCreated SystemTime="2025-03-21T14:23:10Z"/>
  </System>
  <EventData>
    <Data Name="TargetUserName">wangpeng</Data>
    <Data Name="IpAddress">10.10.1.100</Data>
  </EventData>
</Event>

解析时我最头疼的就是 命名空间 和 路径表达式。很多人直接用 find() 找节点，但没注意命名空间，结果 find('EventID') 返回 None。我曾经在一个政府项目里排查了一整天，才发现是因为 XML 声明了默认命名空间，必须用 find('{http://...}EventID') 才能查到。

避坑指南：

始终使用带命名空间感知的解析器（如 lxml 的 etree）。
优先用 XPath 提取关键字段，别手写逐层遍历——容易漏掉节点。
注意 XML 转义：&、<、> 等实体在解析后会自动还原，但如果你直接字符串替换，会破坏结构。

我踩过的坑： 有一次我用 xml.etree.ElementTree 解析，发现 TimeCreated 节点明明是 <TimeCreated SystemTime=“...”/>，但 child.attrib 总是空的。后来才意识到该节点是自闭合标签，text 是 None，必须用 child.get('SystemTime') 取属性——而不是取文本。

4.3 CEF —— 网络安全设备的“老兵”

CEF:0|PaloAlto|PAN-OS|10.2.0|1001|Traffic Start|5|src=10.10.1.100 dst=8.8.8.8 sport=12345 dport=80 proto=tcp action=allow

解析 CEF 的关键在于两个点：

头部分割：用竖线 | 分割前 7 段，注意有的设备在版本后可能有多余空白。
扩展字段：用空格分隔键值对，但值里可能包含空格（比如 msg=User logged in from remote）。所以不能简单用 split()——我一般会先匹配 ([^=]+)=([^ ]+.*) 或使用成熟的 CEF 解析库。

个人习惯： 我倾向于用 cef Python 库（pip install cef），它内部处理了转义字符（如 \=、\|、\\）和值内空格，很省心。但如果性能要求极高，我会用正则一次性提取全部字段。

4.4 LEEF —— IBM QRadar 的“私房菜”

LEEF:1.0|CheckPoint|Firewall|R81|traffic|Session accepted|5|src=10.10.1.100|dst=8.8.8.8|proto=tcp

我第一次看到 LEEF 时还纳闷：为什么扩展字段也用竖线？后来才意识到，IBM 为了对齐 CEF 的字段顺序，但想保持扩展字段的紧凑性。解析时要注意：

LEEF 的版本号不是固定整数，可能是 1.0、2.0 等。
扩展字段的 key=value 之间是竖线，但值里不能包含竖线——如果有，必须用 \| 转义，这点跟 CEF 一样。
我曾在某个项目中遇到设备输出的 LEEF 字符串里有 |\| 三个字符，我直接 split('|') 后把键值对拆烂了。后来加了转义预处理：先把 \| 替换成占位符，再分割。

建议： 对于 LEEF，最好先提取头部 7 个字段，再处理剩余部分。因为有的设备会在尾部附加自定义字段，只靠 split('|', maxsplit=7) 会更安全。

4.5 四种格式对比速查表

维度	JSON	XML	CEF	LEEF
结构	键值对/嵌套	树形元素+属性	固定头+扩展k=v	固定头+扩展k=v
可读性	高	中（标签冗余）	中（需要解头）	中
编码/转义	JSON 标准反斜杠	XML 实体	\=, \\|, \\	\=, \\|, \\
解析复杂度	低	中（命名空间敏感）	中（分隔符歧义）	中（分隔符歧义）
典型场景	Web API、CloudTrail	Windows EVTX、XSD合规	防火墙、IDS	QRadar 生态
字段顺序要求	不要求	不要求（XSD可约束）	头部严格顺序	头部严格顺序

4.6 解析通用原则与工具

不管你用什么格式，我的习惯是先过一道 归一化：把时间戳、IP、用户名等映射成统一的字段名（如 @timestamp、src_ip、user_name）。这样后续关联规则就可以写在同一套字段上。

几种常用解析工具：

Python：json、xml.etree.ElementTree（或 lxml）、cef 库、自定义正则。
Logstash：提供 json、xml、cef 等 filter 插件，配置即可。
自研解析引擎：如果吞吐量要求高（百万 EPS），建议用 C++ 或者 Go 实现，关键在字符串处理上避免内存拷贝。

性能警告： 不要用递归遍历 XML 或 JSON 的嵌套结构做全量展开——如果存在深度达 100 层的异常数据，递归会很容易栈溢出。提前设置最大深度或者用迭代方式解析。

4.7 一个实际解析流程示例（Python）

假设我们需要从一段混合日志流中自动识别格式并解析：

import json, re, xml.etree.ElementTree as ET

def parse_log(line):
    # 优先尝试 JSON
    try:
        return json.loads(line), 'json'
    except json.JSONDecodeError:
        pass

    # 尝试 XML（检测是否以 < 开头）
    if line.strip().startswith('<'):
        try:
            root = ET.fromstring(line)
            # 提取关键信息，略
            return root, 'xml'
        except ET.ParseError:
            pass

    # 尝试 CEF
    if line.startswith('CEF:'):
        parts = line.split('|', 7)
        if len(parts) >= 7:
            header = parts[:7]
            ext = parts[-1] if len(parts) == 8 else ''
            ext_dict = dict(re.findall(r'(\w+)=([^|]*)', ext))
            return (header, ext_dict), 'cef'
    
    # 尝试 LEEF
    if line.startswith('LEEF:'):
        parts = line.split('|', 7)
        if len(parts) >= 7:
            header = parts[:7]
            ext = parts[-1] if len(parts) == 8 else ''
            ext_dict = dict(re.findall(r'(\w+)=([^|]*)', ext))
            return (header, ext_dict), 'leef'

    return None, 'unknown'

注意：这只是一个演示，生产环境一定要加更健壮的转义处理（比如 CEF 中的 \= 和 \|）。

4.8 我看到的几个典型踩坑点

时间戳时区丢失：JSON 里没有时区隐式当作 UTC，但有的设备输出了本地时间且不带 TZ——我曾经把上海时间当 UTC 存了，导致关联分析时差乱了 8 小时。解决方案：强制要求日志源在字段中明确时区，或者在采集层配置时区信息。
CEF 扩展字段值含等号：例如 msg=User=wangpeng logged in，如果用 split('=') 会多出字段。用正则 ([^=]+)=(.+) 或者先匹配到第一个 = 再分割。
XML 的 CDATA：某些审计日志会把敏感信息包在 <![CDATA[...]]> 里，解析后内容会包含标签。记得在解析后做一次 xml.etree.CDATA 处理。

好了，格式解析这块我就讲这么多。你可能会问：为什么一定要掌握这么多格式？因为真实环境里你永远不知道下一台设备会吐出什么格式的日志。多掌握一种，边界关联就少一个盲区。后面几章我们就会基于这些解析结果来做跨数据源的关联分析，打好这个基础，后面就顺了。