Day-019-网络流量基线建立

# Day 18: 网络流量基线建立

> 网络安全系列第 18 天 | 预计阅读时间：30 分钟 | 难度：★★★★☆

---

## 清单 目录

1. [引言](#引言)
2. [流量基线概念](#流量基线概念)
3. [基线指标体系](#基线指标体系)
4. [数据收集方法](#数据收集方法)
5. [基线建立流程](#基线建立流程)
6. [异常检测技术](#异常检测技术)
7. [实验环境搭建](#实验环境搭建)
8. [实战演练](#实战演练)
9. [监控与告警](#监控与告警)
10. [总结与思考](#总结与思考)
11. [参考资料](#参考资料)

---

## 引言

### 为什么需要流量基线

在网络安全的日常运营中，最大的挑战之一是区分"正常"和"异常"流量。没有基线，就无法知道什么是异常；没有异常检测，就无法及时发现攻击。

**流量基线的价值**：
- 统计 **异常检测基础**：知道什么是正常，才能发现异常
-  **攻击发现**：DDoS、数据外泄、横向移动等攻击会改变流量模式
- 增长 **容量规划**：了解流量趋势，提前规划扩容
- 目标 **性能优化**：识别瓶颈，优化网络配置
- 说明 **合规要求**：许多标准要求持续监控和基线分析

**真实案例**：
```
案例 1: 数据外泄检测（2020）
- 公司建立了流量基线
- 监控发现夜间流量异常（超出基线 3σ）
- 调查发现内部人员窃取数据
- 及时阻止，避免重大损失

案例 2: DDoS 预警（2021）
- 基线显示正常流量 1Gbps
- 监控发现流量缓慢增长到 1.5Gbps
- 提前扩容，避免服务中断
- 后续遭遇 2Tbps 攻击，因有准备未受影响

案例 3: 横向移动检测（2019）
- 建立服务器间通信基线
- 发现异常的内部扫描流量
- 及时隔离受感染服务器
- 阻止勒索软件扩散
```

---

## 流量基线概念

### 什么是流量基线

```
流量基线定义：
网络在"正常"状态下的流量特征集合，包括：
- 流量大小（带宽使用率）
- 流量模式（时间分布）
- 协议分布
- 连接特征
- 行为模式

基线不是固定值，而是一个范围：
- 平均值（Mean）
- 标准差（Standard Deviation）
- 最小值/最大值
- 百分位数（P95, P99）
```

### 基线类型

```
1. 时间基线
   - 小时基线（24 小时周期）
   - 天基线（7 天周期）
   - 周基线（4 周周期）
   - 月基线（12 月周期）
   - 季节性基线（年度周期）

2. 空间基线
   - 网段基线（不同子网）
   - 应用基线（不同服务）
   - 用户基线（不同部门）
   - 地理位置基线

3. 协议基线
   - TCP 基线
   - UDP 基线
   - ICMP 基线
   - 应用协议基线（HTTP, DNS, SSH 等）
```

### 基线统计模型

```
正态分布模型：
假设流量服从正态分布
正常范围：μ ± 2σ (95% 置信区间)
异常阈值：μ ± 3σ (99.7% 置信区间)

优点：简单、计算快
缺点：网络流量通常不服从正态分布

百分位数模型：
使用 P95、P99 作为阈值
正常：<= P95
警告：P95-P99
异常：> P99

优点：不假设分布
缺点：需要大量数据

时间序列模型：
考虑时间因素
工作日 vs 周末
白天 vs 夜间
业务高峰期 vs 低峰期

优点：更准确
缺点：复杂、需要训练
```

---

## 基线指标体系

### 基础流量指标

```
1. 带宽使用率
   - 入站带宽（bps, Mbps, Gbps）
   - 出站带宽
   - 总带宽
   - 带宽利用率（%）

2. 包速率
   - 包/秒（pps）
   - 入站包速率
   - 出站包速率

3. 连接数
   - 活跃连接数
   - 新建连接速率（conn/s）
   - 并发连接数
   - 连接失败率

4. 流量大小
   - 平均包大小
   - 最大包大小
   - 总字节数
```

### 协议分布指标

```
1. 传输层协议
   - TCP 占比（%）
   - UDP 占比
   - 其他协议占比

2. 应用层协议
   - HTTP/HTTPS 流量
   - DNS 流量
   - SSH/RDP 流量
   - 数据库流量
   - 邮件流量

3. 端口分布
   - Top 10 目标端口
   - Top 10 源端口
   - 非常用端口流量
```

### 行为指标

```
1. 通信模式
   - Top 10 源 IP
   - Top 10 目标 IP
   - Top 10 IP 对
   - 内部 vs 外部流量比

2. 时间模式
   - 小时流量分布
   - 工作日 vs 周末
   - 业务时间 vs 非业务时间

3. 异常行为
   - 端口扫描检测
   - 暴力破解检测
   - 数据外泄检测
   - 横向移动检测
```

---

## 数据收集方法

### 流量捕获工具

```
1. 端口镜像（SPAN）
   交换机配置：
   - 镜像所有端口到监控端口
   - 镜像特定 VLAN
   - 镜像特定端口

优点：无侵入、完整流量
   缺点：需要交换机支持、可能影响性能

2. 网络分路器（TAP）
   物理设备，复制流量
   优点：完全被动、不影响网络
   缺点：需要硬件、成本较高

3. 主机抓包
   tcpdump, Wireshark
   优点：简单、无需额外设备
   缺点：只能看到本机流量

4. 流数据（NetFlow/sFlow）
   路由器/交换机导出流统计
   优点：开销小、覆盖广
   缺点：只有统计信息、无包内容
```

### 工具配置示例

```bash
# 1. tcpdump 长期抓包
tcpdump -i eth0 -s 1500 -C 100 -W 24 \
  -w /var/log/pcap/traffic-%Y%m%d-%H.pcap

# -s 1500: 捕获完整包
# -C 100: 每个文件 100MB
# -W 24: 保留 24 个文件（循环）

# 2. tshark 统计
tshark -i eth0 -q -z io,stat,60 \
  -z conv,tcp -z conv,udp \
  -w /var/log/tshark/stats.log

# 3. ntopng 部署
apt install ntopng
systemctl start ntopng
# 访问 http://localhost:3000

# 4. NetFlow 配置（Cisco）
interface GigabitEthernet0/1
  ip flow ingress
  ip flow egress

ip flow-export destination 192.168.1.100 2055
ip flow-export version 9
```

### 数据标准化

```
数据格式统一：
- 时间戳：ISO 8601 (YYYY-MM-DDTHH:MM:SSZ)
- IP 地址：点分十进制
- 端口：整数
- 协议：标准名称（TCP, UDP, HTTP）

数据清洗：
- 去除无效数据
- 填补缺失值
- 去重
- 格式化

数据聚合：
- 按时间聚合（1 分钟、5 分钟、1 小时）
- 按 IP 聚合
- 按协议聚合
- 按应用聚合
```

---

## 基线建立流程

### 步骤 1: 确定监控范围

```
需要监控的对象：
□ 边界流量（互联网出入口）
□ 关键服务器（数据库、应用服务器）
□ 关键网段（DMZ、内网核心）
□ 关键应用（Web、邮件、数据库）
□ 关键用户（管理员、VIP）

优先级划分：
P0: 必须监控（核心业务）
P1: 重要监控（关键系统）
P2: 一般监控（普通用户）
```

### 步骤 2: 数据收集期

```
收集时长建议：
- 最小：2 周（覆盖工作日和周末）
- 推荐：4 周（覆盖完整业务周期）
- 理想：3 个月（覆盖业务变化）

数据量估算：
1Gbps 链路 × 24 小时 × 14 天 = 21TB 原始数据

实际存储（聚合后）：
- 原始包：压缩后约 5TB
- 流数据：约 500GB
- 聚合统计：约 50GB

建议策略：
- 原始包：保留 7 天
- 流数据：保留 30 天
- 聚合统计：保留 1 年
```

### 步骤 3: 数据分析

```python
# Python 基线分析示例
import pandas as pd
import numpy as np
from datetime import datetime

# 读取流量数据
df = pd.read_csv('traffic_data.csv', parse_dates=['timestamp'])

# 按小时聚合
hourly = df.groupby(df['timestamp'].dt.hour)['bytes'].agg(['mean', 'std', 'min', 'max', 'count'])

# 计算基线
baseline = {
    'hour_0': {'mean': hourly.loc[0, 'mean'], 'std': hourly.loc[0, 'std']},
    'hour_1': {'mean': hourly.loc[1, 'mean'], 'std': hourly.loc[1, 'std']},
    # ...
}

# 计算阈值（3σ原则）
thresholds = {}
for hour in range(24):
    mean = hourly.loc[hour, 'mean']
    std = hourly.loc[hour, 'std']
    thresholds[hour] = {
        'warning': mean + 2 * std,
        'critical': mean + 3 * std
    }

# 保存基线
import json
with open('baseline.json', 'w') as f:
    json.dump(thresholds, f, indent=2)
```

### 步骤 4: 基线验证

```
验证方法：

1. 历史数据回测
   - 使用已知正常时段数据验证
   - 误报率应 < 5%
   - 漏报率应 < 1%

2. 注入测试
   - 模拟已知攻击流量
   - 验证能否检测
   - 调整阈值

3. 业务验证
   - 与业务部门确认
   - 排除业务高峰期
   - 考虑特殊事件

4. 持续调优
   - 每周审查告警
   - 调整误报阈值
   - 更新基线模型
```

---

## 异常检测技术

### 统计方法

```
1. Z-Score 方法
   Z = (X - μ) / σ
   |Z| > 3 → 异常

优点：简单、快速
   缺点：假设正态分布

2. IQR 方法（四分位距）
   IQR = Q3 - Q1
   异常：< Q1 - 1.5×IQR 或 > Q3 + 1.5×IQR

优点：不假设分布
   缺点：需要足够数据

3. EWMA（指数加权移动平均）
   更重视近期数据
   适合动态基线

优点：适应变化
   缺点：参数敏感
```

### 机器学习方法

```
1. 孤立森林（Isolation Forest）
   基于树的方法检测异常
   适合高维数据

2. 单类 SVM
   学习正常数据边界
   边界外为异常

3. 自动编码器
   深度学习重建误差
   误差大为异常

4. 时间序列预测
   LSTM/ARIMA 预测
   实际与预测差异大为异常
```

### 规则方法

```
预定义规则：
- 流量 > 1Gbps → 告警
- 新建连接 > 1000/s → 告警
- 非常用端口流量 > 10% → 告警
- 夜间流量 > 白天 50% → 告警

组合规则：
- (流量异常 AND 连接数异常) → 高优先级
- (流量异常 OR 连接数异常) → 中优先级

上下文规则：
- 工作时间：宽松阈值
- 非工作时间：严格阈值
- 节假日：特殊阈值
```

---

## 实验环境搭建

### 监控环境拓扑

```
┌─────────────────────────────────────────┐
│            生产网络                      │
│                                         │
│  ┌─────────┐    ┌─────────┐            │
│  │  Web    │    │   DB    │            │
│  │ Server  │    │ Server  │            │
│  └────┬────┘    └────┬────┘            │
│       │              │                  │
│       └──────┬───────┘                  │
│              │                           │
│       ┌──────▼──────┐                   │
│       │   交换机     │                   │
│       │ (带端口镜像) │                   │
│       └──────┬──────┘                   │
│              │                           │
│       ┌──────▼──────┐                   │
│       │  监控网络   │                   │
│       └──────┬──────┘                   │
│              │                           │
│    ┌─────────┼─────────┐                │
│    │         │         │                │
│ ┌──▼──┐  ┌───▼───┐ ┌──▼──┐             │
│ │流量 │  │ 分析  │ │ 告警 │             │
│ │收集 │  │ 服务器 │ │ 系统 │             │
│ └─────┘  └───────┘ └─────┘             │
└─────────────────────────────────────────┘
```

### 工具栈部署

```bash
# 1. ELK Stack 部署
# Elasticsearch
apt install elasticsearch
systemctl start elasticsearch

# Logstash
apt install logstash
# 配置 input/filter/output

# Kibana
apt install kibana
systemctl start kibana

# 2. Grafana + Prometheus
apt install prometheus grafana
systemctl start prometheus grafana-server

# 3. ntopng
apt install ntopng
systemctl start ntopng

# 4. 自定义监控脚本
cat > /etc/cron.hourly/traffic-baseline.sh << 'EOF'
#!/bin/bash
# 每小时收集流量统计
INTERFACES="eth0 eth1"
OUTPUT="/var/log/traffic/stats"

for iface in $INTERFACES; do
    # 收集接口统计
    cat /proc/net/dev | grep $iface >> $OUTPUT/$(date +%Y%m%d).log
    
    # 收集连接统计
    netstat -an | awk '{print $6}' | sort | uniq -c >> $OUTPUT/connections.log
    
    # 收集 Top 连接
    netstat -an | awk 'NR>2 {print $5}' | cut -d: -f1 | sort | uniq -c | sort -rn | head -10 >> $OUTPUT/top_ips.log
done
EOF
chmod +x /etc/cron.hourly/traffic-baseline.sh
```

---

## 实战演练

### 实验 1: 基础流量监控

**目标**：建立基础流量监控

**步骤**：

```bash
# 1. 安装监控工具
apt install iftop nethogs vnstat

# 2. 实时监控
iftop -i eth0
# 查看实时流量和 Top 连接

nethogs eth0
# 按进程查看流量

# 3. 历史统计
vnstat
vnstat -d  # 按天
vnstat -m  # 按月

# 4. 配置 vnstat 守护进程
systemctl enable vnstat
systemctl start vnstat

# 5. 生成报告
vnstat --exportdb > /var/log/traffic/vnstat_export.json
```

### 实验 2: 基线计算

**目标**：计算流量基线

**步骤**：

```python
#!/usr/bin/env python3
# baseline_calc.py

import pandas as pd
import numpy as np
import json
from datetime import datetime, timedelta

# 读取历史流量数据
def load_traffic_data(days=30):
    data = []
    for i in range(days):
        date = (datetime.now() - timedelta(days=i)).strftime('%Y%m%d')
        try:
            df = pd.read_csv(f'/var/log/traffic/stats/{date}.log')
            data.append(df)
        except FileNotFoundError:
            continue
    return pd.concat(data)

# 计算基线
def calculate_baseline(df):
    # 按小时分组
    df['hour'] = pd.to_datetime(df['timestamp']).dt.hour
    
    baseline = {}
    for hour in range(24):
        hour_data = df[df['hour'] == hour]['bytes']
        
        baseline[hour] = {
            'mean': hour_data.mean(),
            'std': hour_data.std(),
            'min': hour_data.min(),
            'max': hour_data.max(),
            'p95': hour_data.quantile(0.95),
            'p99': hour_data.quantile(0.99)
        }
    
    return baseline

# 保存基线
def save_baseline(baseline):
    with open('/etc/monitoring/baseline.json', 'w') as f:
        json.dump(baseline, f, indent=2, default=str)

if __name__ == '__main__':
    df = load_traffic_data(30)
    baseline = calculate_baseline(df)
    save_baseline(baseline)
    print("基线计算完成，已保存到 /etc/monitoring/baseline.json")
```

### 实验 3: 异常检测

**目标**：实现异常检测

**步骤**：

```python
#!/usr/bin/env python3
# anomaly_detection.py

import json
import smtplib
from email.mime.text import MIMEText
from datetime import datetime

def load_baseline():
    with open('/etc/monitoring/baseline.json', 'r') as f:
        return json.load(f)

def get_current_traffic():
    # 获取当前流量（从监控工具或 API）
    # 这里简化处理
    import subprocess
    result = subprocess.run(
        ['cat', '/sys/class/net/eth0/statistics/rx_bytes'],
        capture_output=True, text=True
    )
    return int(result.stdout.strip())

def check_anomaly(current, baseline, hour):
    hour_str = str(hour)
    if hour_str not in baseline:
        return None
    
    mean = baseline[hour_str]['mean']
    std = baseline[hour_str]['std']
    
    z_score = (current - mean) / std if std > 0 else 0
    
    if abs(z_score) > 3:
        return {
            'level': 'CRITICAL',
            'z_score': z_score,
            'current': current,
            'expected': mean,
            'deviation': (current - mean) / mean * 100
        }
    elif abs(z_score) > 2:
        return {
            'level': 'WARNING',
            'z_score': z_score,
            'current': current,
            'expected': mean,
            'deviation': (current - mean) / mean * 100
        }
    
    return None

def send_alert(anomaly):
    msg = MIMEText(f"""
流量异常告警

级别：{anomaly['level']}
Z-Score: {anomaly['z_score']:.2f}
当前流量：{anomaly['current']} bytes
预期流量：{anomaly['expected']:.0f} bytes
偏差：{anomaly['deviation']:.1f}%

时间：{datetime.now()}
    """)
    
    msg['Subject'] = f"流量异常告警 - {anomaly['level']}"
    msg['From'] = 'monitor@example.com'
    msg['To'] = 'admin@example.com'
    
    with smtplib.SMTP('localhost') as server:
        server.send_message(msg)

if __name__ == '__main__':
    baseline = load_baseline()
    current = get_current_traffic()
    hour = datetime.now().hour
    
    anomaly = check_anomaly(current, baseline, hour)
    
    if anomaly:
        print(f"检测到异常：{anomaly['level']}")
        send_alert(anomaly)
    else:
        print("流量正常")
```

### 实验 4: 可视化仪表板

**目标**：创建监控仪表板

**步骤**：

```bash
# 1. 安装 Grafana
apt install grafana-server
systemctl start grafana-server

# 2. 配置数据源
# 访问 http://localhost:3000
# 添加 Prometheus 数据源

# 3. 创建仪表板
# 导入或创建以下面板：

# 面板 1: 实时流量
# PromQL: rate(network_rx_bytes[5m])

# 面板 2: 流量趋势
# PromQL: sum(rate(network_rx_bytes[5m])) by (instance)

# 面板 3: Top 连接
# PromQL: topk(10, sum by (dst) (rate(network_bytes[5m])))

# 面板 4: 协议分布
# PromQL: sum by (protocol) (rate(network_bytes[5m]))

# 面板 5: 异常检测
# PromQL: abs((current - baseline) / baseline_std) > 3

# 4. 配置告警
# 设置阈值告警
# 配置通知渠道（邮件、Slack、钉钉等）
```

---

## 监控与告警

### 告警策略

```
告警级别定义：

INFO（信息）:
- 流量略高于基线（1-2σ）
- 无需立即行动
- 记录用于分析

WARNING（警告）:
- 流量明显高于基线（2-3σ）
- 需要关注
- 准备响应

CRITICAL（严重）:
- 流量异常高（>3σ）
- 立即响应
- 启动应急流程

EMERGENCY（紧急）:
- 流量极度异常（>5σ）
- 可能正在遭受攻击
- 立即启动应急预案
```

### 告警渠道

```bash
# 1. 邮件告警
# 配置 Postfix 或 Sendmail
# 使用 Python smtplib 发送

# 2. Slack 告警
curl -X POST -H 'Content-type: application/json' \
  --data '{"text":"流量异常告警"}' \
  https://hooks.slack.com/services/YOUR/WEBHOOK/URL

# 3. 钉钉告警
curl 'https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN' \
  -H 'Content-Type: application/json' \
  -d '{"msgtype":"text","text":{"content":"流量异常告警"}}'

# 4. 短信告警
# 使用云服务商 API（阿里云、腾讯云等）
```

### 告警优化

```
减少误报：
1. 设置告警抑制
   - 相同告警 5 分钟内只发一次
   - 维护窗口暂停告警

2. 多指标确认
   - 流量异常 + 连接数异常 → 确认告警
   - 单一指标异常 → 降低级别

3. 时间窗口
   - 持续 5 分钟异常才告警
   - 避免瞬时波动

4. 基线更新
   - 每周更新基线
   - 考虑业务变化
   - 季节性调整
```

---

## 总结与思考

### 核心要点回顾

1. **基线重要性**
   - 异常检测的基础
   - 需要持续维护
   - 不是一劳永逸

2. **建立流程**
   - 确定监控范围
   - 收集足够数据
   - 计算统计基线
   - 持续验证调优

3. **检测方法**
   - 统计方法（简单快速）
   - 机器学习（准确但复杂）
   - 规则方法（灵活但需维护）

### 深入思考问题

1. **基线漂移**
   - 业务增长导致基线变化
   - 如何区分正常增长和异常？
   - 自适应基线 vs 固定基线？

2. **加密流量**
   - 90%+ 流量已加密
   - 如何检测加密流量中的异常？
   - 元数据分析 vs 解密检测？

3. **云环境**
   - 云原生架构动态变化
   - 传统基线方法适用吗？
   - 需要新的方法论？

### 实战建议

**中小企业**：
1. 从简单工具开始（vnstat, iftop）
2. 建立基础基线
3. 配置关键告警
4. 逐步完善

**大型企业**：
1. 部署专业监控（ELK, Grafana）
2. 建立完整基线体系
3. 机器学习异常检测
4. 自动化响应

**云环境**：
1. 使用云监控（CloudWatch, Azure Monitor）
2. 关注云原生指标
3. 容器流量监控
4. Serverless 监控

---

## 参考资料

### 工具资源
- [ntopng](https://www.ntop.org/)
- [ELK Stack](https://www.elastic.co/)
- [Grafana](https://grafana.com/)
- [Prometheus](https://prometheus.io/)

### 在线资源
- [网络监控最佳实践](https://www.sans.org/)
- [基线建立指南](https://www.nist.gov/)

### 书籍推荐
- 《网络流量分析》
- 《异常检测技术》
- 《网络安全监控》

---

**标记 明日预告**：Day 19 - 网络取证基础

> 本文内容仅供学习和研究使用，请勿用于非法目的。所有实验请在隔离环境中进行。

---

*本文是 365 天信息安全技术系列的第 18 篇，精编版本*