Day-276-日志标准化与解析

# Day 297: 日志标准化与解析 - CEF/LEEF/自定义解析规则

> 安全运维系列第 7 天 | 预计阅读时间：25 分钟 | 难度：★★★☆☆

---

## 清单 目录

1. [日志标准化概述](#日志标准化概述)
2. [通用日志格式](#通用日志格式)
3. [CEF 格式详解](#cef-格式详解)
4. [LEEF 格式详解](#leef-格式详解)
5. [自定义解析规则](#自定义解析规则)
6. [解析引擎技术](#解析引擎技术)
7. [标准化实践](#标准化实践)
8. [总结与思考](#总结与思考)
9. [参考资料](#参考资料)

---

## 日志标准化概述

### 为什么需要标准化

**问题背景**：
- 不同厂商日志格式各异
- 同一厂商不同产品格式不同
- 字段命名和含义不统一
- 时间格式多样化

**标准化价值**：
- 统一分析基础
- 简化关联规则
- 提高分析效率
- 降低运维成本

### 标准化层次

**字段级标准化**：
- 统一字段命名（src_ip → srcIp）
- 统一数据类型（IP、端口、时间戳）
- 统一枚举值（事件类型、严重性）

**语义级标准化**：
- 统一事件分类
- 统一动作语义
- 统一对象类型

**架构级标准化**：
- 采用标准格式（CEF、LEEF、ECS）
- 遵循行业规范

---

## 通用日志格式

### CEF (Common Event Format)

**定义**：ArcSight 开发的通用事件格式，被广泛支持。

**字段说明**：
| 字段 | 说明 | 示例 |
|------|------|------|
| Version | CEF 版本 | 0 |
| Device Vendor | 设备厂商 | Symantec |
| Device Product | 设备产品 | IDS |
| Device Version | 设备版本 | 1.0 |
| Device Event Class ID | 事件类 ID | 100 |
| Name | 事件名称 | SSH Login Attempt |
| Severity | 严重性 (0-10) | 5 |
| Extension | 扩展键值对 | src=192.168.1.1 dst=10.0.0.1 |

**示例**：
```
CEF:0|Symantec|IDS|1.0|100|SSH Login Attempt|5|src=192.168.1.100 dst=10.0.0.50 spt=54321 dpt=22 cn1=1 cn1Label=Attempts rt=1649772600000
```

### LEEF (Log Event Extended Format)

**定义**：IBM QRadar 开发的日志扩展格式。

### ECS (Elastic Common Schema)

**定义**：Elastic 开发的通用日志模式。

**核心字段**：
```json
{
  "@timestamp": "2026-04-12T14:30:00.000Z",
  "agent": {
    "hostname": "server01",
    "version": "8.0.0"
  },
  "source": {
    "ip": "192.168.1.100",
    "port": 54321,
    "address": "192.168.1.100"
  },
  "destination": {
    "ip": "10.0.0.50",
    "port": 22
  },
  "user": {
    "name": "admin",
    "id": "1000"
  },
  "event": {
    "action": "login",
    "category": "authentication",
    "type": "start",
    "severity": 3
  }
}
```

---

## CEF 格式详解

### 扩展字段规范

**标准扩展字段**：
| 字段 | 说明 | 类型 |
|------|------|------|
| rt | 接收时间 | 时间戳 |
| src | 源 IP | IP 地址 |
| spt | 源端口 | 端口号 |
| dst | 目的 IP | IP 地址 |
| dpt | 目的端口 | 端口号 |
| suser | 源用户名 | 字符串 |
| duser | 目的用户名 | 字符串 |
| act | 动作 | 字符串 |
| msg | 消息 | 字符串 |
| cn1/cn2/cn3 | 自定义数字 | 数字 |
| cs1/cs2/cs3 | 自定义字符串 | 字符串 |

**自定义字段**：
```
任意键值对，格式：key=value
多个字段用空格分隔
特殊字符需要转义
```

### CEF 解析规则

**正则表达式**：
```regex
^CEF:(\d+)\|([^|]+)\|([^|]+)\|([^|]+)\|([^|]+)\|([^|]+)\|(\d+)\|(.*)$
```

**解析示例（Python）**：
```python
import re

def parse_cef(cef_string):
    pattern = r'^CEF:(\d+)\|([^|]+)\|([^|]+)\|([^|]+)\|([^|]+)\|([^|]+)\|(\d+)\|(.*)$'
    match = re.match(pattern, cef_string)
    
    if not match:
        return None
    
    groups = match.groups()
    
    event = {
        'version': groups[0],
        'device_vendor': groups[1],
        'device_product': groups[2],
        'device_version': groups[3],
        'event_class_id': groups[4],
        'name': groups[5],
        'severity': int(groups[6]),
        'extension': parse_extension(groups[7])
    }
    
    return event

def parse_extension(ext_string):
    # 解析键值对，处理转义字符
    ext = {}
    # 简化实现，实际需要考虑转义
    pairs = ext_string.split()
    for pair in pairs:
        if '=' in pair:
            key, value = pair.split('=', 1)
            ext[key] = value
    return ext
```

---

## LEEF 格式详解

### 属性字段

**必需字段**：
- cat: 类别
- devname: 设备名称

**常用字段**：
| 字段 | 说明 |
|------|------|
| src | 源 IP |
| dst | 目的 IP |
| sPort | 源端口 |
| dPort | 目的端口 |
| usrName | 用户名 |
| severity | 严重性 |
| desc | 描述 |

### LEEF 解析示例

```python
def parse_leef(leef_string):
    lines = leef_string.split('\t')
    header = lines[0].split('|')
    
    event = {
        'version': header[0].replace('LEEF:', ''),
        'vendor': header[1],
        'product': header[2],
        'version': header[3],
        'event_id': header[4],
        'attributes': {}
    }
    
    # 解析属性
    for line in lines[1:]:
        if '=' in line:
            key, value = line.split('=', 1)
            event['attributes'][key] = value
    
    return event
```

---

## 自定义解析规则

### Grok 模式

**Grok 语法**：
```
%{PATTERN:field_name}
%{PATTERN:field_name:datatype}
```

**常用模式**：
```
IP: 匹配 IP 地址
NUMBER: 匹配数字
WORD: 匹配单词
DATA: 匹配任意数据
TIMESTAMP_ISO8601: 匹配 ISO 时间戳
USER: 匹配用户名
```

**示例**：
```grok
# Apache Access Log
%{IPORHOST:clientip} %{USER:ident} %{USER:auth} \[%{HTTPDATE:timestamp}\] "%{WORD:verb} %{DATA:request} HTTP/%{NUMBER:httpversion}" %{NUMBER:response} %{NUMBER:bytes}

# Syslog
%{SYSLOGTIMESTAMP:timestamp} %{SYSLOGHOST:hostname} %{DATA:program}(?:\[%{POSINT:pid}\])?: %{GREEDYDATA:message}
```

### Logstash 解析配置

```ruby
filter {
  # Apache 日志解析
  if [log_type] == "apache" {
    grok {
      match => { "message" => '%{IPORHOST:clientip} %{USER:ident} %{USER:auth} \[%{HTTPDATE:timestamp}\] "%{WORD:verb} %{DATA:request} HTTP/%{NUMBER:httpversion}" %{NUMBER:response} %{NUMBER:bytes}' }
    }
    date {
      match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
      target => "@timestamp"
    }
  }
  
  # Syslog 解析
  if [log_type] == "syslog" {
    grok {
      match => { "message" => '%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}' }
    }
  }
  
  # 标准化字段
  if [clientip] {
    mutate {
      rename => { "clientip" => "src_ip" }
    }
  }
  
  # 添加标准字段
  mutate {
    add_field => {
      "event_category" => "web"
      "datacenter" => "dc1"
    }
  }
}
```

### Fluentd 解析配置

```xml
<parse>
  @type regexp
  expression /^(?<time>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) \[(?<level>\w+)\] (?<message>.*)$/
  time_format %Y-%m-%d %H:%M:%S
</parse>

# JSON 解析
<parse>
  @type json
  time_key timestamp
  time_format %Y-%m-%dT%H:%M:%S.%LZ
</parse>
```

---

## 解析引擎技术

### 正则表达式引擎

**性能考虑**：
- 预编译正则表达式
- 避免贪婪匹配
- 使用原子组
- 减少回溯

**优化示例**：
```regex
# 低效：大量回溯
.*(\d{3}).*

# 高效：精确匹配
[^\d]*(\d{3})
```

### 多行日志处理

**问题**：Java 堆栈、长日志跨多行

**解决方案**：
```yaml
# Filebeat 多行配置
multiline.pattern: '^[[:space:]]'
multiline.negate: false
multiline.match: after
multiline.max_lines: 500

# Logstash 多行
filter {
  multiline {
    pattern => "^[[:space:]]"
    negate => false
    what => "previous"
  }
}
```

### JSON 日志解析

**自动检测**：
```javascript
// 检测并解析 JSON
if (message.startsWith('{')) {
  try {
    const parsed = JSON.parse(message);
    event.fields = parsed;
  } catch (e) {
    // 解析失败，保持原始消息
  }
}
```

**嵌套 JSON 处理**：
```ruby
filter {
  json {
    source => "message"
    target => "json"
  }
  # 扁平化嵌套字段
  mutate {
    rename => {
      "[json][user][name]" => "user_name"
      "[json][user][id]" => "user_id"
    }
  }
}
```

---

## 标准化实践

### 字段映射表

**源字段 → 标准字段**：
| 源字段 | 标准字段 | 说明 |
|--------|----------|------|
| src_ip, source_ip, srcIp | src.ip | 源 IP |
| dst_ip, dest_ip, dstIp | dst.ip | 目的 IP |
| src_port, source_port | src.port | 源端口 |
| dst_port, dest_port | dst.port | 目的端口 |
| user, username, usr | user.name | 用户名 |
| timestamp, time, @timestamp | @timestamp | 时间戳 |
| severity, sev, priority | event.severity | 严重性 |

### 标准化流程

```
原始日志 → 解析 → 字段映射 → 数据验证 → 标准化输出
```

**实施步骤**：
1. 识别日志源和格式
2. 定义解析规则
3. 创建字段映射
4. 实施数据验证
5. 测试和验证
6. 部署和监控

### 质量检查

**完整性检查**：
- 必需字段是否存在
- 字段类型是否正确
- 枚举值是否合法

**一致性检查**：
- IP 地址格式
- 时间戳格式
- 端口范围

**示例检查规则**：
```python
def validate_event(event):
    errors = []
    
    # 检查必需字段
    required_fields = ['@timestamp', 'src.ip', 'dst.ip']
    for field in required_fields:
        if field not in event:
            errors.append(f"Missing required field: {field}")
    
    # 检查 IP 格式
    import ipaddress
    for ip_field in ['src.ip', 'dst.ip']:
        if ip_field in event:
            try:
                ipaddress.ip_address(event[ip_field])
            except ValueError:
                errors.append(f"Invalid IP format: {ip_field}")
    
    # 检查端口范围
    for port_field in ['src.port', 'dst.port']:
        if port_field in event:
            port = event[port_field]
            if not (0 <= port <= 65535):
                errors.append(f"Port out of range: {port_field}")
    
    return errors
```

---

## 总结与思考

### 核心要点回顾

1. **标准化必要性**：统一分析基础，简化关联规则
2. **通用格式**：CEF、LEEF、ECS 是主流标准
3. **CEF 格式**：ArcSight 开发，被广泛支持
4. **LEEF 格式**：IBM QRadar 使用
5. **ECS**：Elastic 开发的现代 schema
6. **解析技术**：Grok、正则、JSON 解析
7. **质量检查**：完整性、一致性验证

### 深入思考

**标准化的挑战**
- 厂商私有格式
- 格式频繁变更
- 性能与灵活性平衡

**最佳实践**
- 优先采用标准格式
- 建立字段映射文档
- 实施解析质量监控
- 定期审查和更新

---

## 参考资料

### 学习资源

- **CEF Reference Guide**: ArcSight CEF 格式参考
- **Elastic Common Schema**: https://www.elastic.co/guide/en/ecs/current/index.html
- **Grok Patterns**: https://github.com/logstash-plugins/logstash-patterns-core

### 工具资源

- **Logstash**: https://www.elastic.co/logstash
- **Fluentd**: https://www.fluentd.org
- **Grok Debugger**: https://grokdebugger.com

---

*Day 297 完成 | 日志标准化与解析详解 | 字数：约 10,000 字*