基于Python实现网络流量监控的方法详解_Python

一、目的

分析最近一段时间内的网络流量情况，以便更好地理解网络使用情况、识别潜在问题并提出改进建议。网络流量监控对于保障网络稳定性、优化资源分配和提高性能具有重要意义。

二、监控指标

1.接收速率

接收速率是指单位时间内接收数据的速度，通常以 mbps（兆位每秒）或 kb/s（千字节每秒）表示。

2.发送速率

发送速率是指单位时间内发送数据的速度，通常以 mbps（兆位每秒）或 kb/s（千字节每秒）表示。

3.吞吐量

吞吐量是指网络传输数据的总量，通常以 mbps（兆位每秒）或 gb（千兆字节）为单位。

4.丢包率

丢包率是指在传输过程中丢失的数据包的比例，通常以百分比或千分比表示。

当然也根据自己的需求获取更加精准的数据，以便更好知道自己电脑的网络是否安全。

三、实验环境

操作系统：kali linux z 6.5.0-kali3-amd64

语言版本：python 3.11.8 (main, feb 7 2024, 21:52:08) [gcc 13.2.0] on linux

四、功能实现

具体的思路如下：

1. 必要用到的库：我们使用psutil库来获取网络流量和系统信息，使用imet库来计算时间间隔。

2. 设置初始值：记录开始时间、初始接收字节数、初始发送字节数以及初始丢包数。

3. 进入监控循环：使用`while true`进入一个无限循环，每次循环都会执行以下步骤：

休眠一段时间：使用`time.sleep(interval)`来控制每次循环之间的时间间隔。
获取当前的网络流量情况：使用`psutil.net_io_counters()`来获取当前的网络输入输出计数器。
计算流量指标：计算从开始时间到当前时间的时间间隔，然后根据时间间隔计算接收速率、发送速率和吞吐量。同时，计算丢包率（100个包）。
打印监控指标：使用`print`函数打印出接收速率、发送速率、吞吐量和丢包率。
更新初始值：将当前的接收字节数、发送字节数和丢包数更新为下一次循环的初始值。

4. 主程序入口：使用monitor_network_traffic()来确保脚本在直接运行时才执行监控函数。

#安装pip pip是安装python的第三方库 
apt-get install python3-pip
pip install psutil   //安装第三方库
具体代码 ： 
   import psutil
   import time
def monitor_network_traffic(interval=1):
    start_time = time.time()
    start_io = psutil.net_io_counters()
    start_rx = start_io.bytes_recv
    start_tx = start_io.bytes_sent
    start_drop = start_io.dropin  # 接收端的丢包数
    while true:
        time.sleep(interval)
        end_io = psutil.net_io_counters()
        end_rx = end_io.bytes_recv
        end_tx = end_io.bytes_sent
        end_drop = end_io.dropin  # 接收端的丢包数
        duration = time.time() - start_time
        rx_speed = (end_rx - start_rx) / duration
        tx_speed = (end_tx - start_tx) / duration
        packet_loss = (end_drop - start_drop) / 100  # 丢包率（100包）
        print(f"每秒接收速率: {rx_speed:.2f} bytes/s")
        print(f"每秒发送速率: {tx_speed:.2f} bytes/s")
        print(f"吞吐量: {(rx_speed + tx_speed):.2f} bytes/s")
        print(f"丢包率（100包）: {packet_loss:.2%}")
        start_rx = end_rx
        start_tx = end_tx
        start_drop = end_drop
monitor_network_traffic()

五、方法补充

网络流量监控是系统运维、安全分析和应用调试的重要手段。python 凭借丰富的第三方库，可以灵活实现从系统级统计到数据包级别的深度分析。本文将介绍四种主流的实现方法，并给出完整的代码示例和适用场景。

1.使用 psutil 获取系统网络 i/o 统计

适用场景：实时监控本机网络流量总量（上传/下载速度、累计字节数），无需解析数据包内容。

原理：psutil 读取 /proc/net/dev（linux）或性能计数器（windows），返回每个网卡的累计收发字节数、包数等。

安装

pip install psutil

示例代码：实时显示所有网卡的流量速率

import psutil
import time
def get_net_io():
    """获取网络i/o统计"""
    net_io = psutil.net_io_counters(pernic=true)
    return net_io
def format_speed(bytes_per_sec):
    """将字节/秒格式化为可读单位"""
    for unit in ['b', 'kb', 'mb', 'gb']:
        if bytes_per_sec < 1024:
            return f"{bytes_per_sec:.2f} {unit}/s"
        bytes_per_sec /= 1024
    return f"{bytes_per_sec:.2f} tb/s"
def monitor_bandwidth(interval=1):
    """监控各网卡的实时带宽"""
    print("开始监控网络流量 (按 ctrl+c 停止)...")
    prev = get_net_io()
    try:
        while true:
            time.sleep(interval)
            curr = get_net_io()
            for nic, curr_stats in curr.items():
                prev_stats = prev.get(nic)
                if not prev_stats:
                    continue
                sent_rate = (curr_stats.bytes_sent - prev_stats.bytes_sent) / interval
                recv_rate = (curr_stats.bytes_recv - prev_stats.bytes_recv) / interval
                print(f"{nic}: 上传 {format_speed(sent_rate)} | 下载 {format_speed(recv_rate)}")
            prev = curr
    except keyboardinterrupt:
        print("\n监控结束")
if __name__ == "__main__":
    monitor_bandwidth()

关键点

psutil.net_io_counters(pernic=true) 返回字典，键为网卡名，值为 snetio 对象。
计算差值除以时间间隔即得实时速率。
无需管理员权限，轻量级，适合后台常驻监控。

2.使用 scapy 捕获并分析数据包

适用场景：需要解析数据包头部（如 ip、tcp、http）、进行协议分析或自定义过滤。

原理：scapy 可以工作在用户态，利用 pcap 或 socket 抓取链路层或网络层数据包，并提供强大的协议解析能力。

安装

pip install scapy

注意：在 linux 下抓包可能需要 sudo 权限；windows 需要安装 npcap 或 winpcap。

示例代码：捕获并统计 http 请求

from scapy.all import sniff, ip, tcp, raw
def packet_callback(packet):
    """回调函数，处理每个捕获的数据包"""
    if ip in packet and tcp in packet:
        tcp = packet[tcp]
        # 检查是否是 http 请求（端口 80 或 8080）
        if tcp.dport in [80, 8080] or tcp.sport in [80, 8080]:
            if raw in packet:
                payload = packet[raw].load.decode('utf-8', errors='ignore')
                # 简单判断是否为 http 请求行
                if payload.startswith('get') or payload.startswith('post'):
                    print(f"http请求: {payload[:100]}...")
                    print(f"源ip: {packet[ip].src}:{tcp.sport} -> 目标ip: {packet[ip].dst}:{tcp.dport}")
# 开始抓包（捕获 100 个数据包，过滤 http 流量）
sniff(filter="tcp port 80 or tcp port 8080", prn=packet_callback, count=100)

关键点

sniff 函数的 filter 参数使用 berkeley packet filter (bpf) 语法，可以精确控制抓包范围，避免过多无关数据。
回调函数中对 raw 层的内容进行解析，可提取应用层数据。
需要管理员/root 权限。

3.使用 pyshark 解析 pcap 文件（离线分析）

适用场景：已有抓包文件（如 wireshark 保存的 .pcap），需要自动化分析或提取特征。

原理：pyshark 是 wireshark 的 python 封装，利用 tshark 命令行解析 pcap 文件，返回结构化的数据包对象。

安装

pip install pyshark

示例代码：统计 pcap 文件中的 tcp 连接

import pyshark
def analyze_pcap(pcap_file):
    cap = pyshark.filecapture(pcap_file, display_filter="tcp")
    connections = set()
    for packet in cap:
        try:
            src = packet.ip.src
            dst = packet.ip.dst
            sport = packet.tcp.srcport
            dport = packet.tcp.dstport
            connections.add((src, sport, dst, dport))
        except attributeerror:
            continue
    cap.close()
    print(f"共发现 {len(connections)} 个 tcp 连接")
    for conn in list(connections)[:10]:
        print(f"{conn[0]}:{conn[1]} -> {conn[2]}:{conn[3]}")
if __name__ == "__main__":
    analyze_pcap("capture.pcap")

关键点

filecapture 支持实时捕获（使用 livecapture）和离线文件。
可以设置 display_filter 为 wireshark 过滤语法，如 "tcp.port == 443"。
需要系统安装 wireshark（包含 tshark）。

4.使用 pydivert（windows）进行流量拦截和修改

适用场景：在 windows 上实现类似防火墙的实时拦截、重定向或修改数据包内容。

原理：pydivert 封装了 windows 的 windivert 驱动，可以捕获、过滤、修改和重新注入网络数据包。

安装

pip install pydivert

示例代码：记录所有访问 80 端口的流量并阻止

import pydivert
def block_http():
    with pydivert.windivert("tcp.dstport == 80 or tcp.srcport == 80") as w:
        print("开始监控并阻止 http 流量...")
        for packet in w:
            print(f"拦截数据包: {packet.src_addr}:{packet.src_port} -> {packet.dst_addr}:{packet.dst_port}")
            # 直接丢弃数据包（不调用 packet.send()）
            # 若想放行，则调用 w.send(packet)
        # 循环持续直到进程终止
if __name__ == "__main__":
    block_http()

关键点