脚本如何监控网络流量

本文目录导读：

1. 使用 Python + scapy
2. 使用 Bash + tcpdump
3. 使用 Python + psutil (更简单)
4. 高级用法：nethogs 风格的进程监控
5. 安装依赖
6. 使用权限
7. 注意事项

使用 Python + scapy

from scapy.all import *
from collections import defaultdict
import time
class NetworkMonitor:
    def __init__(self):
        self.traffic_stats = defaultdict(lambda: {'bytes': 0, 'packets': 0})
    def packet_callback(self, packet):
        if IP in packet:
            src = packet[IP].src
            dst = packet[IP].dst
            size = len(packet)
            # 统计源IP流量
            self.traffic_stats[src]['bytes'] += size
            self.traffic_stats[src]['packets'] += 1
            # 统计目标IP流量
            self.traffic_stats[dst]['bytes'] += size
            self.traffic_stats[dst]['packets'] += 1
    def print_stats(self):
        print(f"\n{time.strftime('%Y-%m-%d %H:%M:%S')}")
        print("IP流量统计（Top 10）：")
        sorted_ip = sorted(
            self.traffic_stats.items(),
            key=lambda x: x[1]['bytes'],
            reverse=True
        )[:10]
        for ip, stats in sorted_ip:
            print(f"{ip:20s} 包数: {stats['packets']:6d}  "
                  f"流量: {stats['bytes']/1024:.2f} KB")
    def monitor(self, interface='eth0', duration=60):
        print(f"开始监控网卡 {interface}，持续 {duration} 秒...")
        sniff(
            iface=interface,
            prn=self.packet_callback,
            store=0,
            timeout=duration
        )
        self.print_stats()
# 使用示例
if __name__ == "__main__":
    monitor = NetworkMonitor()
    monitor.monitor(interface='eth0', duration=30)

使用 Bash + tcpdump

#!/bin/bash
# 网络流量监控脚本
INTERFACE="eth0"
DURATION=60
LOG_FILE="/var/log/network_traffic.log"
monitor_traffic() {
    echo "=== 监控开始: $(date) ===" >> $LOG_FILE
    # 使用tcpdump捕获流量
    tcpdump -i $INTERFACE -n -t 2>/dev/null | while read line; do
        # 提取源IP和目标IP（假设是IP包）
        if [[ $line =~ IP ]]; then
            src_ip=$(echo $line | awk '{print $3}' | cut -d'.' -f1-4)
            dst_ip=$(echo $line | awk '{print $5}' | cut -d'.' -f1-4)
            size=$(echo $line | awk '{print $NF}' | tr -d '()')
            echo "$(date +%H:%M:%S) $src_ip -> $dst_ip (大小: $size bytes)" >> $LOG_FILE
        fi
    done &
    # 每秒显示流量统计
    while [ $SECONDS -lt $DURATION ]; do
        # 使用/proc/net/dev查看接口流量
        rx_bytes=$(cat /proc/net/dev | grep $INTERFACE | awk '{print $2}')
        tx_bytes=$(cat /proc/net/dev | grep $INTERFACE | awk '{print $10}')
        echo "时间: $(date +%H:%M:%S) RX: $((rx_bytes/1024)) KB TX: $((tx_bytes/1024)) KB"
        sleep 1
    done
    echo "=== 监控结束: $(date) ===" >> $LOG_FILE
}
# 实时流量监控
realtime_monitor() {
    echo "实时流量监控（按Ctrl+C退出）"
    while true; do
        rx_bytes=$(cat /proc/net/dev | grep $INTERFACE | awk '{print $2}')
        tx_bytes=$(cat /proc/net/dev | grep $INTERFACE | awk '{print $10}')
        # 计算速度（每秒）
        if [ -n "$last_rx" ]; then
            rx_speed=$(( ($rx_bytes - $last_rx) / 1024 ))
            tx_speed=$(( ($tx_bytes - $last_tx) / 1024 ))
            echo "下载: ${rx_speed}KB/s  上传: ${tx_speed}KB/s"
        fi
        last_rx=$rx_bytes
        last_tx=$tx_bytes
        sleep 1
    done
}
# 主菜单
case "$1" in
    monitor)
        monitor_traffic
        ;;
    realtime)
        realtime_monitor
        ;;
    *)
        echo "用法: $0 {monitor|realtime}"
        exit 1
        ;;
esac

使用 Python + psutil (更简单)

#!/usr/bin/env python3
import psutil
import time
from datetime import datetime
class SimpleNetworkMonitor:
    def __init__(self, interface='eth0'):
        self.interface = interface
        self.last_bytes = None
        self.last_time = None
    def get_network_stats(self):
        net_io = psutil.net_io_counters(pernic=True)
        if self.interface in net_io:
            return net_io[self.interface]
        return None
    def monitor_speed(self, duration=60):
        print(f"监控 {self.interface} 网卡流量速度...")
        # 获取初始值
        init_stats = self.get_network_stats()
        if not init_stats:
            print(f"未找到网卡 {self.interface}")
            return
        last_sent = init_stats.bytes_sent
        last_recv = init_stats.bytes_recv
        last_time = time.time()
        while duration > 0:
            time.sleep(1)
            current_stats = self.get_network_stats()
            if not current_stats:
                break
            now = time.time()
            time_diff = now - last_time
            # 计算速度
            sent_speed = (current_stats.bytes_sent - last_sent) / time_diff / 1024  # KB/s
            recv_speed = (current_stats.bytes_recv - last_recv) / time_diff / 1024  # KB/s
            print(f"\r{datetime.now().strftime('%H:%M:%S')} "
                  f"↑ {sent_speed:.1f} KB/s  ↓ {recv_speed:.1f} KB/s", end='')
            last_sent = current_stats.bytes_sent
            last_recv = current_stats.bytes_recv
            last_time = now
            duration -= 1
        print("\n监控结束")
# 使用示例
if __name__ == "__main__":
    monitor = SimpleNetworkMonitor('eth0')
    monitor.monitor_speed(duration=30)

高级用法：nethogs 风格的进程监控

#!/usr/bin/env python3
import psutil
import time
from collections import defaultdict
class ProcessNetworkMonitor:
    def __init__(self):
        self.process_traffic = defaultdict(lambda: {'sent': 0, 'recv': 0})
    def get_process_connections(self):
        connections = psutil.net_connections()
        process_data = {}
        for conn in connections:
            if conn.status == 'ESTABLISHED' and conn.pid:
                try:
                    process = psutil.Process(conn.pid)
                    process_name = process.name()
                    if process_name not in process_data:
                        process_data[process_name] = {
                            'pid': conn.pid,
                            'connections': []
                        }
                    process_data[process_name]['connections'].append({
                        'local': conn.laddr,
                        'remote': conn.raddr if conn.raddr else None
                    })
                except (psutil.NoSuchProcess, psutil.AccessDenied):
                    continue
        return process_data
    def monitor_process_traffic(self, duration=30):
        print("监控进程网络流量...")
        print(f"{'进程名称':20s} {'PID':8s} {'上传':>10s} {'下载':>10s} {'连接数':>8s}")
        print("-" * 56)
        while duration > 0:
            process_data = self.get_process_connections()
            for process_name, data in process_data.items():
                # 简单统计
                num_connections = len(data['connections'])
                sent = self.process_traffic[process_name]['sent']
                recv = self.process_traffic[process_name]['recv']
                # 更新流量（实际项目需要更复杂的计算）
                sent += 1024  # 示例数据
                recv += 2048  # 示例数据
                print(f"{process_name:20s} {str(data['pid']):8s} "
                      f"{sent/1024:>8.1f}KB {recv/1024:>8.1f}KB "
                      f"{num_connections:>8d}")
            self.cleanup_old_data()
            time.sleep(2)
            duration -= 2
            # 清屏
            print("\033[2J\033[H", end='')
    def cleanup_old_data(self):
        # 清理超过时间的进程数据
        pass
# 使用示例
if __name__ == "__main__":
    monitor = ProcessNetworkMonitor()
    monitor.monitor_process_traffic(duration=30)

安装依赖

# Python依赖
pip install scapy psutil
# 系统工具（Ubuntu/Debian）
sudo apt-get install tcpdump nethogs iftop

使用权限

大多数网络监控需要 root 权限：

# 使用sudo运行脚本
sudo python3 network_monitor.py
sudo bash network_monitor.sh

注意事项

1. 性能影响：持续监控会消耗系统资源
2. 隐私问题：监控他人网络流量可能涉及隐私
3. 权限要求：需要适当的系统权限
4. 数据存储：大量流量数据需要合理存储和处理

选择哪种方法取决于你的具体需求：

• 简单监控：使用 psutil
• 详细分析：使用 scapy
• 命令行工具：使用 tcpdump/nethogs
• 进程级监控：使用 psutil + netstat