基于Python开发实现网络流量分析工具详解

from scapy.all import sniff, IP, TCP, UDP
import queue

class PacketCapture:
    def __init__(self, interface='eth0'):
        self.interface = interface
        self.packet_queue = queue.Queue()
        
    def packet_handler(self, packet):
        """处理捕获的数据包"""
        if IP in packet:
            packet_info = {
                'timestamp': float(packet.time),
                'src_ip': packet[IP].src,
                'dst_ip': packet[IP].dst,
                'protocol': packet[IP].proto,
                'length': len(packet)
            }
            
            # TCP/UDP端口信息
            if TCP in packet:
                packet_info['src_port'] = packet[TCP].sport
                packet_info['dst_port'] = packet[TCP].dport
                packet_info['protocol_name'] = 'TCP'
            elif UDP in packet:
                packet_info['src_port'] = packet[UDP].sport
                packet_info['dst_port'] = packet[UDP].dport
                packet_info['protocol_name'] = 'UDP'
                
            self.packet_queue.put(packet_info)
    
    def start_capture(self, count=0):
        """开始捕获数据包"""
        sniff(iface=self.interface, 
              prn=self.packet_handler, 
              count=count, 
              store=False)

from collections import defaultdict
from datetime import datetime

class TrafficAnalyzer:
    def __init__(self):
        self.stats = {
            'protocol_dist': defaultdict(int),
            'ip_traffic': defaultdict(int),
            'port_traffic': defaultdict(int),
            'traffic_timeline': []
        }
    
    def analyze_packet(self, packet_info):
        """分析单个数据包"""
        # 协议分布统计
        protocol = packet_info.get('protocol_name', 'OTHER')
        self.stats['protocol_dist'][protocol] += 1
        
        # IP流量统计
        src_ip = packet_info['src_ip']
        dst_ip = packet_info['dst_ip']
        length = packet_info['length']
        
        self.stats['ip_traffic'][src_ip] += length
        self.stats['ip_traffic'][dst_ip] += length
        
        # 端口流量统计
        if 'dst_port' in packet_info:
            port = packet_info['dst_port']
            self.stats['port_traffic'][port] += 1
        
        # 时间序列数据
        self.stats['traffic_timeline'].append({
            'timestamp': packet_info['timestamp'],
            'bytes': length
        })
    
    def get_top_talkers(self, n=10):
        """获取流量最大的IP地址"""
        sorted_ips = sorted(
            self.stats['ip_traffic'].items(),
            key=lambda x: x[1],
            reverse=True
        )
        return sorted_ips[:n]
    
    def get_protocol_distribution(self):
        """获取协议分布"""
        total = sum(self.stats['protocol_dist'].values())
        return {
            protocol: (count / total) * 100
            for protocol, count in self.stats['protocol_dist'].items()
        }

from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime

Base = declarative_base()

class TrafficRecord(Base):
    __tablename__ = 'traffic_records'
    
    id = Column(Integer, primary_key=True)
    timestamp = Column(DateTime, default=datetime.utcnow)
    src_ip = Column(String(45))
    dst_ip = Column(String(45))
    src_port = Column(Integer, nullable=True)
    dst_port = Column(Integer, nullable=True)
    protocol = Column(String(10))
    packet_length = Column(Integer)
    
class TrafficStats(Base):
    __tablename__ = 'traffic_stats'
    
    id = Column(Integer, primary_key=True)
    timestamp = Column(DateTime, default=datetime.utcnow)
    total_packets = Column(Integer)
    total_bytes = Column(Integer)
    top_protocol = Column(String(10))
    avg_packet_size = Column(Float)

class DatabaseManager:
    def __init__(self, db_url='sqlite:///traffic.db'):
        self.engine = create_engine(db_url)
        Base.metadata.create_all(self.engine)
        Session = sessionmaker(bind=self.engine)
        self.session = Session()
    
    def save_packet(self, packet_info):
        """保存数据包信息"""
        record = TrafficRecord(
            src_ip=packet_info['src_ip'],
            dst_ip=packet_info['dst_ip'],
            src_port=packet_info.get('src_port'),
            dst_port=packet_info.get('dst_port'),
            protocol=packet_info.get('protocol_name', 'UNKNOWN'),
            packet_length=packet_info['length']
        )
        self.session.add(record)
        self.session.commit()

from flask import Flask, jsonify, request
from flask_cors import CORS
from flask_socketio import SocketIO, emit
import threading

app = Flask(__name__)
CORS(app)
socketio = SocketIO(app, cors_allowed_origins="*")

# 全局对象
capture = PacketCapture()
analyzer = TrafficAnalyzer()
db_manager = DatabaseManager()

@app.route('/api/capture/start', methods=['POST'])
def start_capture():
    """启动数据包捕获"""
    interface = request.json.get('interface', 'eth0')
    
    def capture_thread():
        capture.interface = interface
        capture.start_capture()
    
    thread = threading.Thread(target=capture_thread)
    thread.daemon = True
    thread.start()
    
    return jsonify({'status': 'started', 'interface': interface})

@app.route('/api/stats/protocols', methods=['GET'])
def get_protocol_stats():
    """获取协议分布统计"""
    distribution = analyzer.get_protocol_distribution()
    return jsonify(distribution)

@app.route('/api/stats/top-talkers', methods=['GET'])
def get_top_talkers():
    """获取流量Top IP"""
    n = request.args.get('limit', 10, type=int)
    top_talkers = analyzer.get_top_talkers(n)
    
    result = [
        {'ip': ip, 'bytes': bytes_count}
        for ip, bytes_count in top_talkers
    ]
    return jsonify(result)

@app.route('/api/stats/timeline', methods=['GET'])
def get_timeline():
    """获取流量时间线数据"""
    timeline = analyzer.stats['traffic_timeline'][-100:]
    return jsonify(timeline)

@socketio.on('connect')
def handle_connect():
    """WebSocket连接建立"""
    emit('connected', {'status': 'Connected to traffic analyzer'})

def broadcast_realtime_data():
    """实时广播流量数据"""
    while True:
        if not capture.packet_queue.empty():
            packet = capture.packet_queue.get()
            analyzer.analyze_packet(packet)
            
            # 通过WebSocket推送实时数据
            socketio.emit('packet_update', packet)
            socketio.emit('stats_update', {
                'protocols': analyzer.get_protocol_distribution(),
                'top_talkers': analyzer.get_top_talkers(5)
            })

if __name__ == '__main__':
    # 启动实时数据广播线程
    broadcast_thread = threading.Thread(target=broadcast_realtime_data)
    broadcast_thread.daemon = True
    broadcast_thread.start()
    
    socketio.run(app, host='0.0.0.0', port=5000)

import React, { useState, useEffect } from 'react';
import { io } from 'socket.io-client';
import ReactECharts from 'echarts-for-react';
import axios from 'axios';

const TrafficDashboard = () => {
  const [protocolData, setProtocolData] = useState([]);
  const [topTalkers, setTopTalkers] = useState([]);
  const [realtimePackets, setRealtimePackets] = useState([]);
  
  useEffect(() => {
    // 建立WebSocket连接
    const socket = io('http://localhost:5000');
    
    socket.on('stats_update', (data) => {
      // 更新协议分布
      const protocols = Object.entries(data.protocols).map(([name, value]) => ({
        name,
        value: value.toFixed(2)
      }));
      setProtocolData(protocols);
      
      // 更新Top Talkers
      setTopTalkers(data.top_talkers);
    });
    
    socket.on('packet_update', (packet) => {
      setRealtimePackets(prev => [...prev.slice(-50), packet]);
    });
    
    return () => socket.disconnect();
  }, []);
  
  // 协议分布饼图配置
  const protocolChartOption = {
    title: { text: '协议分布', left: 'center' },
    tooltip: { trigger: 'item' },
    series: [{
      type: 'pie',
      radius: '50%',
      data: protocolData,
      emphasis: {
        itemStyle: {
          shadowBlur: 10,
          shadowOffsetX: 0,
          shadowColor: 'rgba(0, 0, 0, 0.5)'
        }
      }
    }]
  };
  
  // Top Talkers柱状图配置
  const topTalkersChartOption = {
    title: { text: 'Top流量IP', left: 'center' },
    tooltip: { trigger: 'axis' },
    xAxis: {
      type: 'category',
      data: topTalkers.map(t => t.ip)
    },
    yAxis: { type: 'value', name: '字节数' },
    series: [{
      type: 'bar',
      data: topTalkers.map(t => t.bytes),
      itemStyle: { color: '#5470c6' }
    }]
  };
  
  const startCapture = async () => {
    try {
      await axios.post('http://localhost:5000/api/capture/start', {
        interface: 'eth0'
      });
      alert('数据包捕获已启动');
    } catch (error) {
      console.error('启动失败:', error);
    }
  };
  
  return (
    <div className="dashboard">
      <h1>网络流量分析工具</h1>
      
      <button onClick={startCapture}>启动捕获</button>
      
      <div className="charts-container">
        <ReactECharts option={protocolChartOption} style={{ height: 400 }} />
        <ReactECharts option={topTalkersChartOption} style={{ height: 400 }} />
      </div>
      
      <div className="realtime-packets">
        <h3>实时数据包</h3>
        <table>
          <thead>
            <tr>
              <th>源IP</th>
              <th>目标IP</th>
              <th>协议</th>
              <th>长度</th>
            </tr>
          </thead>
          <tbody>
            {realtimePackets.slice(-20).map((packet, idx) => (
              <tr key={idx}>
                <td>{packet.src_ip}</td>
                <td>{packet.dst_ip}</td>
                <td>{packet.protocol_name}</td>
                <td>{packet.length}</td>
              </tr>
            ))}
          </tbody>
        </table>
      </div>
    </div>
  );
};

export default TrafficDashboard;

from sklearn.ensemble import IsolationForest
import numpy as np

class AnomalyDetector:
    def __init__(self):
        self.model = IsolationForest(contamination=0.1, random_state=42)
        self.is_trained = False
    
    def train(self, traffic_data):
        """训练异常检测模型"""
        features = self._extract_features(traffic_data)
        self.model.fit(features)
        self.is_trained = True
    
    def _extract_features(self, traffic_data):
        """提取流量特征"""
        features = []
        for packet in traffic_data:
            feature_vector = [
                packet['length'],
                packet.get('src_port', 0),
                packet.get('dst_port', 0),
                hash(packet['src_ip']) % 1000,
                hash(packet['dst_ip']) % 1000
            ]
            features.append(feature_vector)
        return np.array(features)
    
    def detect_anomaly(self, packet):
        """检测单个数据包是否异常"""
        if not self.is_trained:
            return False
        
        features = self._extract_features([packet])
        prediction = self.model.predict(features)
        return prediction[0] == -1  # -1表示异常

from jinja2 import Template
from datetime import datetime

class ReportGenerator:
    def __init__(self, analyzer):
        self.analyzer = analyzer
    
    def generate_html_report(self, output_file='traffic_report.html'):
        """生成HTML格式报告"""
        template_str = """
        <!DOCTYPE html>
        <html>
        <head>
            <title>流量分析报告</title>
            <style>
                body { font-family: Arial, sans-serif; margin: 20px; }
                h1 { color: #333; }
                table { border-collapse: collapse; width: 100%; }
                th, td { border: 1px solid #ddd; padding: 8px; text-align: left; }
                th { background-color: #4CAF50; color: white; }
            </style>
        </head>
        <body>
            <h1>网络流量分析报告</h1>
            <p>生成时间: {{ timestamp }}</p>
            
            <h2>协议分布</h2>
            <table>
                <tr><th>协议</th><th>占比</th></tr>
                {% for protocol, percentage in protocols.items() %}
                <tr><td>{{ protocol }}</td><td>{{ percentage }}%</td></tr>
                {% endfor %}
            </table>
            
            <h2>流量Top 10 IP</h2>
            <table>
                <tr><th>IP地址</th><th>流量(字节)</th></tr>
                {% for ip, bytes in top_talkers %}
                <tr><td>{{ ip }}</td><td>{{ bytes }}</td></tr>
                {% endfor %}
            </table>
        </body>
        </html>
        """
        
        template = Template(template_str)
        
        report_html = template.render(
            timestamp=datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
            protocols=self.analyzer.get_protocol_distribution(),
            top_talkers=self.analyzer.get_top_talkers(10)
        )
        
        with open(output_file, 'w', encoding='utf-8') as f:
            f.write(report_html)

# Dockerfile
FROM python:3.9-slim

WORKDIR /app

# 安装系统依赖
RUN apt-get update && apt-get install -y \
    libpcap-dev \
    gcc \
    && rm -rf /var/lib/apt/lists/*

# 安装Python依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制应用代码
COPY . .

# 暴露端口
EXPOSE 5000

# 启动应用(需要root权限捕获数据包)
CMD ["python", "app.py"]

# docker-compose.yml
version: '3.8'

services:
  traffic-analyzer:
    build: .
    ports:
      - "5000:5000"
    volumes:
      - ./data:/app/data
    environment:
      - DATABASE_URL=postgresql://user:password@db:5432/traffic
    depends_on:
      - db
      - redis
    network_mode: host
    privileged: true

  db:
    image: postgres:13
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: password
      POSTGRES_DB: traffic
    volumes:
      - postgres_data:/var/lib/postgresql/data

  redis:
    image: redis:6-alpine
    ports:
      - "6379:6379"

volumes:
  postgres_data: