PyQt5 网络嗅探器 - 实时捕获和分析网络数据包

import sys from PyQt5.QtWidgets import QApplication, QMainWindow, QPushButton, QComboBox, QTableWidget, QTableWidgetItem, QVBoxLayout, QHBoxLayout, QWidget, QTextEdit, QHeaderView from scapy.all import sniff import threading

class Sniffer(QMainWindow): def init(self): super().init() self.initUI()

def initUI(self):
    self.setWindowTitle('PyQt5 Network Sniffer')
    self.setGeometry(100, 100, 1200, 800)

    mainLayout = QVBoxLayout()

    controlLayout = QHBoxLayout()

    self.comboBox = QComboBox(self)
    self.comboBox.addItems(['All', 'HTTP', 'HTTPS', 'FTP', 'SMTP', 'IMAP', 'POP3', 'DNS'])
    controlLayout.addWidget(self.comboBox)

    self.startButton = QPushButton('Start Sniffing', self)
    self.startButton.clicked.connect(self.start_sniffing)
    controlLayout.addWidget(self.startButton)

    self.stopButton = QPushButton('Stop Sniffing', self)
    self.stopButton.clicked.connect(self.stop_sniffing)
    controlLayout.addWidget(self.stopButton)

    self.resetButton = QPushButton('Reset', self)
    self.resetButton.clicked.connect(self.reset_table)
    controlLayout.addWidget(self.resetButton)

    mainLayout.addLayout(controlLayout)

    self.tableWidget = QTableWidget(self)
    self.tableWidget.setColumnCount(5)
    self.tableWidget.setHorizontalHeaderLabels(['源IP', '源端口', '目标IP', '目标端口', '长度'])
    self.tableWidget.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch)
    self.tableWidget.clicked.connect(self.show_packet_detail)
    mainLayout.addWidget(self.tableWidget)

    self.textEdit = QTextEdit(self)
    mainLayout.addWidget(self.textEdit)

    container = QWidget()
    container.setLayout(mainLayout)
    self.setCentralWidget(container)

    self.stop_sniffing_flag = False
    self.packets = []

def packet_callback(self, packet):
    self.packets.append(packet)
    src_ip = packet['IP'].src if packet.haslayer('IP') else '-'
    dst_ip = packet['IP'].dst if packet.haslayer('IP') else '-'
    src_port = packet[1].sport if hasattr(packet[1], 'sport') else '-'
    dst_port = packet[1].dport if hasattr(packet[1], 'dport') else '-'
    length = len(packet)

    rowPosition = self.tableWidget.rowCount()
    self.tableWidget.insertRow(rowPosition)
    self.tableWidget.setItem(rowPosition, 0, QTableWidgetItem(src_ip))
    self.tableWidget.setItem(rowPosition, 1, QTableWidgetItem(str(src_port)))
    self.tableWidget.setItem(rowPosition, 2, QTableWidgetItem(dst_ip))
    self.tableWidget.setItem(rowPosition, 3, QTableWidgetItem(str(dst_port)))
    self.tableWidget.setItem(rowPosition, 4, QTableWidgetItem(str(length)))

def start_sniffing(self):
    self.startButton.setEnabled(False)  # Disable the startButton
    protocol = self.comboBox.currentText().lower()
    if protocol == 'all':
        filter_str = 'ip'
    else:
        filter_str = {
            'http': 'tcp port 80',
            'https': 'tcp port 443',
            'ftp': 'tcp port 21',
            'smtp': 'tcp port 25',
            'imap': 'tcp port 143',
            'pop3': 'tcp port 110',
            'dns': 'udp port 53'
        }.get(protocol, 'ip')

    self.stop_sniffing_flag = False
    t = threading.Thread(target=self.sniff_thread, args=(filter_str,))
    t.start()

def sniff_thread(self, filter_str):
    sniff(prn=self.packet_callback, filter=filter_str, stop_filter=lambda x: self.stop_sniffing_flag)

def stop_sniffing(self):
    self.stop_sniffing_flag = True
    self.startButton.setEnabled(True)  # Enable the startButton after stopping sniffing

def reset_table(self):
    self.tableWidget.setRowCount(0)
    self.textEdit.clear()
    self.packets = []

def show_packet_detail(self):
    row = self.tableWidget.currentRow()
    packet = self.packets[row]
    self.textEdit.clear()
    self.textEdit.append(packet.show(dump=True))

if name == 'main': app = QApplication(sys.argv) mainWin = Sniffer() mainWin.show() sys.exit(app.exec_())