#安裝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  # 接收端的丟包數(shù)
    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  # 接收端的丟包數(shù)
        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"每秒發(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()

pip install psutil

import psutil
import time
def get_net_io():
    """獲取網(wǎng)絡(luò)I/O統(tǒng)計(jì)"""
    net_io = psutil.net_io_counters(pernic=True)
    return net_io
def format_speed(bytes_per_sec):
    """將字節(jié)/秒格式化為可讀單位"""
    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):
    """監(jiān)控各網(wǎng)卡的實(shí)時(shí)帶寬"""
    print("開始監(jiān)控網(wǎng)絡(luò)流量 (按 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監(jiān)控結(jié)束")
if __name__ == "__main__":
    monitor_bandwidth()

pip install scapy

from scapy.all import sniff, IP, TCP, Raw
def packet_callback(packet):
    """回調(diào)函數(shù)，處理每個(gè)捕獲的數(shù)據(jù)包"""
    if IP in packet and TCP in packet:
        tcp = packet[TCP]
        # 檢查是否是 HTTP 請(qǐng)求（端口 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')
                # 簡(jiǎn)單判斷是否為 HTTP 請(qǐng)求行
                if payload.startswith('GET') or payload.startswith('POST'):
                    print(f"HTTP請(qǐng)求: {payload[:100]}...")
                    print(f"源IP: {packet[IP].src}:{tcp.sport} -> 目標(biāo)IP: {packet[IP].dst}:{tcp.dport}")
# 開始抓包（捕獲 100 個(gè)數(shù)據(jù)包，過濾 HTTP 流量）
sniff(filter="tcp port 80 or tcp port 8080", prn=packet_callback, count=100)

pip install pyshark

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"共發(fā)現(xiàn) {len(connections)} 個(gè) TCP 連接")
    for conn in list(connections)[:10]:
        print(f"{conn[0]}:{conn[1]} -> {conn[2]}:{conn[3]}")
if __name__ == "__main__":
    analyze_pcap("capture.pcap")

pip install pydivert

import pydivert
def block_http():
    with pydivert.WinDivert("tcp.DstPort == 80 or tcp.SrcPort == 80") as w:
        print("開始監(jiān)控并阻止 HTTP 流量...")
        for packet in w:
            print(f"攔截?cái)?shù)據(jù)包: {packet.src_addr}:{packet.src_port} -> {packet.dst_addr}:{packet.dst_port}")
            # 直接丟棄數(shù)據(jù)包（不調(diào)用 packet.send()）
            # 若想放行，則調(diào)用 w.send(packet)
        # 循環(huán)持續(xù)直到進(jìn)程終止
if __name__ == "__main__":
    block_http()