Day-275-日志采集技术

# Day 296: 日志采集技术 - Syslog/Agent/API/文件采集

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

---

## 清单 目录

1. [日志采集概述](#日志采集概述)
2. [Syslog 协议采集](#syslog-协议采集)
3. [Agent 代理采集](#agent-代理采集)
4. [API 接口采集](#api-接口采集)
5. [文件采集](#文件采集)
6. [数据库日志采集](#数据库日志采集)
7. [云环境日志采集](#云环境日志采集)
8. [采集架构设计](#采集架构设计)
9. [总结与思考](#总结与思考)
10. [参考资料](#参考资料)

---

## 日志采集概述

### 日志采集的重要性

日志采集是安全运营的基础，没有完整准确的日志采集，后续的分析、检测、响应都无从谈起。

**采集目标**：
- **完整性**：采集所有相关日志，无遗漏
- **及时性**：日志产生后尽快采集（目标：< 5 分钟）
- **准确性**：日志内容完整，无丢失或篡改
- **标准化**：统一日志格式，便于分析

### 日志源分类

**网络设备**：
- 防火墙（Palo Alto、Fortinet、Cisco ASA）
- 路由器/交换机（Cisco、Huawei、H3C）
- 负载均衡（F5、Citrix）
- WAF（Imperva、F5 ASM）

**安全设备**：
- IDS/IPS（Snort、Suricata）
- 终端防护（EDR）
- 邮件安全
- Web 代理

**服务器**：
- Windows 事件日志
- Linux Syslog
- 应用日志（Web 服务器、数据库）

**云服务**：
- AWS CloudTrail、VPC Flow Logs
- Azure Activity Log、NSG Flow Logs
- GCP Cloud Audit Logs

**应用系统**：
- 业务应用日志
- 身份认证系统（AD、LDAP）
- 数据库审计日志

### 采集方法对比

| 方法 | 优点 | 缺点 | 适用场景 |
|------|------|------|----------|
| Syslog | 标准化、广泛支持 | 可能丢失、无加密 | 网络设备、Linux |
| Agent | 可靠、功能强 | 需安装、占用资源 | 服务器、终端 |
| API | 实时、结构化 | 依赖 API 稳定性 | 云服务、SaaS |
| 文件采集 | 简单、无需配置 | 轮询延迟、解析复杂 | 应用日志 |
| 数据库 | 结构化、完整 | 性能影响、配置复杂 | 数据库审计 |

---

## Syslog 协议采集

### Syslog 协议基础

**协议标准**：
- RFC 3164：BSD Syslog（传统格式）
- RFC 5424：Structured Syslog（结构化格式）
- RFC 6587：Syslog over TCP

**消息格式（RFC 3164）**：
```
<PRI>TIMESTAMP HOSTNAME TAG: MSG

示例：
<13>Apr 12 14:30:01 server01 sshd[12345]: Accepted password for user from 192.168.1.100
```

**消息格式（RFC 5424）**：
```
<PRI>VERSION TIMESTAMP HOSTNAME APP-NAME PROCID MSGID STRUCTURED-DATA MSG

示例：
<13>1 2026-04-12T14:30:01.123Z server01 sshd 12345 - [meta sequenceId="1"] Accepted password
```

**优先级计算**：
```
PRI = Facility × 8 + Severity

Facility（设施）:
0 - kernel messages
1 - user-level messages
2 - mail system
3 - system daemons
...
16-23 - local use

Severity（严重性）:
0 - Emergency
1 - Alert
2 - Critical
3 - Error
4 - Warning
5 - Notice
6 - Informational
7 - Debug
```

### Syslog 传输方式

**UDP Syslog**：
- 端口：514
- 优点：简单、低开销
- 缺点：不可靠、可能丢失、无加密
- 适用：内网、非关键日志

**TCP Syslog**：
- 端口：514 或 601
- 优点：可靠、有确认机制
- 缺点：开销稍大
- 适用：关键日志、跨网络

**TLS Syslog**：
- 端口：6514
- 优点：加密、安全
- 缺点：需要证书管理、开销大
- 适用：跨公网、敏感日志

### Syslog 服务器

**开源方案**：
- **rsyslog**：Linux 默认，高性能
- **syslog-ng**：功能丰富，灵活配置
- **syslog**：BSD 原始实现

**商业方案**：
- Splunk Universal Forwarder（接收 Syslog）
- QRadar Syslog Collector
- ArcSight Connector

### rsyslog 配置示例

**基础配置**：
```bash
# /etc/rsyslog.conf

# 加载 TCP 模块
module(load="imtcp")
input(type="imtcp" port="514")

# 加载 UDP 模块
module(load="imudp")
input(type="imudp" port="514")

# 按设施转发
auth,authpriv.*                 /var/log/auth.log
mail.*                          /var/log/mail.log
daemon.*                        /var/log/daemon.log

# 按主机名转发
:hostname, isequal, "firewall01"    /var/log/firewall.log

# 转发到远程 SIEM
*.* @@siem.example.com:514
```

**高级配置（结构化日志）**：
```bash
# 解析结构化 Syslog
template(name="StructuredFormat" type="list") {
    constant(value="{")
    constant(value="\"timestamp\":\"")
    property(name="timestamp" dateFormat="rfc3339")
    constant(value="\",\"host\":\"")
    property(name="hostname")
    constant(value="\",\"app\":\"")
    property(name="programname")
    constant(value="\",\"msg\":\"")
    property(name="msg" controlcharacters="on")
    constant(value="\"}")
    constant(value="\n")
}

# 使用模板输出
*.* @@siem.example.com:514;StructuredFormat
```

---

## Agent 代理采集

### Agent 类型

**通用日志 Agent**：
- Splunk Universal Forwarder
- Elastic Beats（Filebeat、Winlogbeat）
- Fluentd/Fluent Bit
- NXLog

**安全专用 Agent**：
- EDR Agent（CrowdStrike、Defender）
- 主机入侵检测 Agent
- 文件完整性监控 Agent

### Splunk Universal Forwarder

**特点**：
- 轻量级（~50MB 内存）
- 只转发，不索引
- 支持数据压缩和加密
- 集中管理

**安装配置**：
```bash
# 安装
wget -O splunkforwarder.tgz "https://download.splunk.com/products/universalforwarder/releases/9.0.0/linux/splunkforwarder-9.0.0-Linux-x86_64.tgz"
tar -xzf splunkforwarder.tgz -C /opt
cd /opt/splunkforwarder/bin

# 初始化
./splunk start --accept-license
./splunk add forward-server siem.example.com:9997
./splunk enable deploy-server

# 监控日志
./splunk add monitor /var/log/auth.log -index main -sourcetype auth:ssh
./splunk add monitor /var/log/secure -index main -sourcetype auth:secure
```

### Elastic Beats

**Beats 家族**：

| Beat | 用途 | 采集内容 |
|------|------|----------|
| Filebeat | 文件日志 | 应用日志、系统日志 |
| Winlogbeat | Windows 事件 | Windows Event Logs |
| Auditbeat | 审计框架 | Linux Audit、文件完整性 |
| Metricbeat | 指标数据 | 系统指标、服务指标 |
| Packetbeat | 网络数据 | 网络流量、协议分析 |
| Heartbeat | 可用性监控 | 服务可用性、响应时间 |

**Filebeat 配置示例**：
```yaml
# filebeat.yml

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/auth.log
    - /var/log/secure
  fields:
    log_type: auth
  fields_under_root: true

- type: log
  enabled: true
  paths:
    - /var/log/nginx/access.log
    - /var/log/nginx/error.log
  fields:
    log_type: nginx
  fields_under_root: true
  parsers:
    - ndjson:
        target: json
        overwrite_keys: true

output.elasticsearch:
  hosts: ["elasticsearch:9200"]
  index: "filebeat-%{+yyyy.MM.dd}"

# 或输出到 Logstash
output.logstash:
  hosts: ["logstash:5044"]
```

### Winlogbeat 配置

**Windows 事件采集**：
```yaml
# winlogbeat.yml

winlogbeat.event_logs:
  - name: Application
    ignore_older: 72h
    level: critical,error,warning

- name: System
    ignore_older: 72h

- name: Security
    ignore_older: 72h
    event_id: 4624,4625,4634,4647,4648,4656,4657,4660,4661,4662,4663,4670,4672,4673,4674,4688,4689,4697,4698,4699,4700,4701,4702,4703,4704,4705,4719,4720,4722,4723,4724,4725,4726,4731,4732,4733,4734,4735,4738,4740,4741,4742,4743,4744,4745,4746,4767,4768,4769,4770,4771,4774,4775,4776,4778,4779,4781,4782,4793,4798,4800,4801,4802,4803,4902,4904,4905,4906,4907,4908,4909,4910,4911,4912,4913,4928,4929,4930,4931,4932,4933,4934,4935,4946,4947,4948,4950,4951,4952,4953,4954,4955,4956,4957,4958,4964,5038,5058,5059,5060,5061,5062,5120,5121,5122,5123,5124,5125,5126,5127,5136,5137,5138,5139,5140,5141,5142,5143,5144,5145,5148,5150,5151,5152,5153,5154,5155,5156,5157,5158,5159,5168,5169,5170,5171,5172,5173,5174,5175,5176,5177,5178,5179,5180,5181,5182,5183,5184,5185,5186,5187,5188,5189,5190,5191,5192,5193,5194,5195,5196,5197,5198,5199,5200,5201,5202,5203,5204,5205,5206,5207,5208,5209,5210,5211,5212

- name: Windows PowerShell
    event_id: 400,403,600,800

- name: Microsoft-Windows-PowerShell/Operational
    event_id: 4103,4104,4105,4106

output.logstash:
  hosts: ["logstash:5044"]
```

---

## API 接口采集

### 云服务 API 采集

**AWS CloudTrail**：
```python
# 使用 boto3 采集 CloudTrail 日志
import boto3
import json

client = boto3.client('cloudtrail')

# 查询事件
response = client.lookup_events(
    LookupAttributes=[
        {'AttributeKey': 'EventName', 'AttributeValue': 'ConsoleLogin'},
    ],
    StartTime=datetime.now() - timedelta(hours=1),
    EndTime=datetime.now(),
    MaxResults=50
)

for event in response['Events']:
    cloud_trail_event = json.loads(event['CloudTrailEvent'])
    # 发送到 SIEM
    send_to_siem(cloud_trail_event)
```

**AWS VPC Flow Logs**：
```
# Flow Logs 格式
version account-id interface-id srcaddr dstaddr srcport dstport protocol packets bytes start end action log-status

示例：
2 123456789010 eni-12345678 192.168.1.100 10.0.0.50 54321 443 6 10 1500 1649772600 1649772660 ACCEPT OK
```

**Azure Activity Log**：
```python
# 使用 Azure SDK 采集
from azure.monitor.query import LogsQueryClient

client = LogsQueryClient(credential)

# 查询 Azure Activity Log
response = client.query_workspace(
    workspace_id=workspace_id,
    query="AzureActivity | where TimeGenerated > ago(1h)",
    timespan=timedelta(hours=1)
)
```

### SaaS 应用 API 采集

**Office 365 Management API**：
```python
# 获取 O365 审计日志
import requests

# 1. 获取访问令牌
token = get_oauth_token()

# 2. 订阅内容
headers = {'Authorization': f'Bearer {token}'}
response = requests.post(
    'https://manage.office.com/api/v1.0/tenant_id/activity/feed/subscriptions/start?contentType=Audit.AzureActiveDirectory',
    headers=headers
)

# 3. 获取内容列表
response = requests.get(
    'https://manage.office.com/api/v1.0/tenant_id/activity/feed/subscriptions/content?contentType=Audit.AzureActiveDirectory&startTime=2026-04-12T00:00:00&endTime=2026-04-12T23:59:59',
    headers=headers
)

# 4. 下载具体内容
for content_uri in response.json():
    logs = requests.get(content_uri['contentUri']).json()
    send_to_siem(logs)
```

**GitHub Audit Log**：
```python
# GitHub Enterprise Audit Log API
import requests

headers = {
    'Authorization': 'Bearer GITHUB_TOKEN',
    'Accept': 'application/json'
}

response = requests.get(
    'https://api.github.com/orgs/ORG_NAME/audit-log',
    headers=headers,
    params={'phrase': 'action:git.push OR action:repo.create', 'per_page': 100}
)

audit_logs = response.json()
```

---

## 文件采集

### 文件采集方法

**轮询方式**：
- 定期检查文件变化
- 读取新增内容
- 记录读取位置（offset）

**inotify 方式（Linux）**：
- 监听文件变化事件
- 实时触发读取
- 低延迟

### Filebeat 文件采集配置

```yaml
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/*.log
    - /var/log/nginx/*.log
    - /var/log/apache2/*.log
  
  # 多行日志处理（Java 堆栈）
  multiline.pattern: '^[[:space:]]'
  multiline.negate: false
  multiline.match: after
  multiline.max_lines: 500
  
  # JSON 日志解析
  processors:
    - decode_json_fields:
        fields: ["message"]
        target: json
        overwrite_keys: true
  
  # 字段添加
  fields:
    environment: production
    datacenter: dc1
  fields_under_root: true
  
  # 忽略旧文件
  ignore_older: 24h
  
  # 扫描频率
  scan_frequency: 10s
  
  # 最大采集字节数
  max_bytes: 10485760
```

### 常见应用日志格式

**Nginx Access Log**：
```
$remote_addr - $remote_user [$time_local] "$request" $status $body_bytes_sent "$http_referer" "$http_user_agent"

示例：
192.168.1.100 - - [12/Apr/2026:14:30:00 +0800] "GET /api/users HTTP/1.1" 200 1234 "-" "Mozilla/5.0"
```

**Apache Access Log**：
```
%h %l %u %t "%r" %>s %b "%{Referer}i" "%{User-Agent}i"

示例：
192.168.1.100 - admin [12/Apr/2026:14:30:00 +0800] "POST /login HTTP/1.1" 302 512 "https://example.com" "Mozilla/5.0"
```

**Tomcat Access Log**：
```
pattern="%h %l %u %t &quot;%r&quot; %s %b &quot;%{Referer}i&quot; &quot;%{User-Agent}i&quot;"
```

---

## 数据库日志采集

### MySQL 日志采集

**通用日志**：
```sql
-- 启用通用查询日志
SET GLOBAL general_log = 'ON';
SET GLOBAL general_log_file = '/var/log/mysql/mysql.log';

-- 启用慢查询日志
SET GLOBAL slow_query_log = 'ON';
SET GLOBAL slow_query_log_file = '/var/log/mysql/slow.log';
SET GLOBAL long_query_time = 2;
```

**审计日志（MySQL Enterprise）**：
```sql
-- 安装审计插件
INSTALL PLUGIN audit_log SONAME 'audit_log.so';

-- 配置审计
SET GLOBAL audit_log_policy = ALL;
SET GLOBAL audit_log_format = JSON;
```

### PostgreSQL 日志采集

**日志配置**：
```postgresql
# postgresql.conf

log_destination = 'stderr'
logging_collector = on
log_directory = '/var/log/postgresql'
log_filename = 'postgresql-%Y-%m-%d_%H%M%S.log'

# 记录内容
log_connections = on
log_disconnections = on
log_duration = on
log_statement = 'all'
log_line_prefix = '%t [%p]: [%l-1] user=%u,db=%d,app=%a,client=%h '
```

### Oracle 审计日志

**统一审计**：
```sql
-- 创建审计策略
CREATE AUDIT POLICY security_audit_policy
ACTIONS ALL ON HR.EMPLOYEES,
ACTIONS LOGON, LOGOFF,
ACTIONS CREATE USER, ALTER USER, DROP USER;

-- 启用策略
AUDIT POLICY security_audit_policy;
```

---

## 云环境日志采集

### AWS 日志采集架构

```
┌─────────────────────────────────────────────────────────────┐
│                      AWS 环境                                │
│                                                             │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐         │
│  │  CloudTrail │  │  VPC Flow   │  │  GuardDuty  │         │
│  │   Logs      │  │   Logs      │  │   Findings  │         │
│  └──────┬──────┘  └──────┬──────┘  └──────┬──────┘         │
│         │                │                │                 │
│         └────────────────┼────────────────┘                 │
│                          │                                  │
│                   ┌──────▼──────┐                          │
│                   │   S3 Bucket │                          │
│                   │  (Logs)     │                          │
│                   └──────┬──────┘                          │
│                          │                                  │
└──────────────────────────┼──────────────────────────────────┘
                           │
                    ┌──────▼──────┐
                    │   Kinesis   │
                    │   Firehose  │
                    └──────┬──────┘
                           │
                    ┌──────▼──────┐
                    │   SIEM      │
                    │  (外部)     │
                    └─────────────┘
```

### Azure 日志采集

**Azure Monitor Logs**：
```
Azure 资源 → Azure Monitor → Log Analytics Workspace → SIEM 集成
```

**诊断设置**：
```json
{
  "properties": {
    "storageAccountId": "/subscriptions/.../storageAccounts/logstorage",
    "serviceBusRuleId": "",
    "workspaceId": "/subscriptions/.../providers/microsoft.operationalinsights/workspaces/loganalytics",
    "logs": [
      {
        "category": "AuditEvent",
        "enabled": true
      },
      {
        "category": "SecurityEvent",
        "enabled": true
      }
    ]
  }
}
```

---

## 采集架构设计

### 架构模式

**集中式采集**：
- 所有日志直接发送到中央 SIEM
- 简单，适合小规模
- 单点故障风险

**分层采集**：
```
日志源 → 采集代理 → 日志聚合器 → SIEM
```
- 适合大规模部署
- 降低 SIEM 负载
- 提高可靠性

**区域采集**：
```
区域 1: 日志源 → 区域采集器 ┐
区域 2: 日志源 → 区域采集器 ├→ 中央 SIEM
区域 3: 日志源 → 区域采集器 ┘
```
- 适合多地域
- 降低网络延迟
- 数据本地化

### 可靠性设计

**缓冲机制**：
- 采集代理本地缓冲
- 网络中断时暂存
- 恢复后补发

**重试机制**：
- 发送失败自动重试
- 指数退避
- 最大重试次数

**监控告警**：
- 采集延迟监控
- 日志量异常告警
- 采集器状态监控

---

## 总结与思考

### 核心要点回顾

1. **采集方法**：Syslog、Agent、API、文件采集各有适用场景
2. **Syslog**：网络设备标准，注意 UDP 可靠性问题
3. **Agent**：可靠功能强，Splunk UF 和 Elastic Beats 是主流
4. **API**：云服务和 SaaS 应用的主要采集方式
5. **文件采集**：应用日志常用，注意多行和 JSON 处理
6. **数据库**：审计日志对安全分析至关重要
7. **云环境**：利用云原生日志服务，设计合理架构

### 深入思考

**采集的挑战**
- 日志量爆炸式增长
- 格式多样化
- 实时性要求高
- 成本和存储压力

**最佳实践**
- 优先采集安全相关日志
- 标准化日志格式
- 实施采集监控
- 定期审查采集覆盖

---

## 参考资料

### 学习资源

- **RFC 5424**: The Syslog Protocol
- **NIST SP 800-92**: Guide to Computer Security Log Management
- **CIS Benchmarks**: 日志配置基准

### 工具资源

- **rsyslog**: https://www.rsyslog.com
- **Elastic Beats**: https://www.elastic.co/beats
- **Splunk UF**: https://www.splunk.com/en_us/download/universal-forwarder.html
- **Fluentd**: https://www.fluentd.org

### 书籍推荐

- 《Logging with ELK Stack》
- 《Splunk Operational Intelligence Cookbook》
- 《Practical Log Analysis》

---

*Day 296 完成 | 日志采集技术详解 | 字数：约 14,000 字*