基于Python的网络安全工具: ARP欺骗、ICMP泛洪和TCP攻击检测

import tkinter as tk
import os
import sys
import time
import threading
import netifaces
from scapy.all import *
from scapy.layers.l2 import ARP
from tkinter import messagebox
from scapy.layers.l2 import getmacbyip

class ArpDetector:
    def __init__(self, interface, timeout=30):
        self.interface = interface
        self.timeout = timeout
        self.ip_mac_map = {}
        self.attacker_ip = None
        self.attacker_mac = None
        self.thread = None
        self.stop_event = threading.Event()

    def start(self):
        self.thread = threading.Thread(target=self.run)
        self.thread.start()

    def stop(self):
        self.stop_event.set()
        if self.thread:
            self.thread.join()

    def run(self):
        self.attacker_ip = netifaces.ifaddresses(self.interface)[netifaces.AF_INET][0]['addr']
        self.attacker_mac = getmacbyip(self.attacker_ip)
        print(f'攻击者IP: {self.attacker_ip}, MAC: {self.attacker_mac}')

        timer = 0
        while not self.stop_event.is_set():
            self.scan_network()
            self.detect_arp_spoofing()
            time.sleep(5)
            timer += 5
            if timer >= self.timeout:
                print('未检测到ARP欺骗')
                self.stop_event.set()

    def scan_network(self):
        for ip in netifaces.ifaddresses(self.interface)[netifaces.AF_INET][0]['addr'].split('.')[:-1]:
            for i in range(1, 255):
                target_ip = f'{ip}.{i}'
                if target_ip != self.attacker_ip:
                    arp_request = ARP(pdst=target_ip)
                    arp_reply = sr1(arp_request, timeout=1, verbose=0)
                    if arp_reply and arp_reply.hwsrc not in ('00:00:00:00:00:00', self.attacker_mac):
                        self.ip_mac_map[target_ip] = arp_reply.hwsrc

    def detect_arp_spoofing(self):
        for target_ip, target_mac in self.ip_mac_map.items():
            arp_request = ARP(op=1, pdst=target_ip, hwdst=target_mac, psrc=self.attacker_ip, hwsrc=self.attacker_mac)
            arp_reply = sr1(arp_request, timeout=1, verbose=0)
            if arp_reply and arp_reply.hwsrc != target_mac:
                print(f'检测到ARP欺骗: {target_ip} ({target_mac}) -> {arp_reply.hwsrc}')

class IcmpFloodDetector:
    def __init__(self, interface, timeout=30):
        self.interface = interface
        self.timeout = timeout
        self.target_ip = '192.168.1.1'
        self.thread = None
        self.stop_event = threading.Event()

    def start(self):
        self.thread = threading.Thread(target=self.run)
        self.thread.start()

    def stop(self):
        self.stop_event.set()
        if self.thread:
            self.thread.join()

    def run(self):
        timer = 0
        while not self.stop_event.is_set():
            self.detect_icmp_flood()
            time.sleep(5)
            timer += 5
            if timer >= self.timeout:
                print('未检测到ICMP泛洪')
                self.stop_event.set()

    def detect_icmp_flood(self):
        icmp_request = IP(dst=self.target_ip)/ICMP()
        icmp_reply = sr1(icmp_request, timeout=1, verbose=0)
        if icmp_reply:
            print(f'检测到ICMP泛洪: {icmp_reply.src} -> {icmp_reply.dst}')

class TcpAttackDetector:
    def __init__(self, interface, timeout=30):
        self.interface = interface
        self.timeout = timeout
        self.target_ip = '192.168.1.1'
        self.thread = None
        self.stop_event = threading.Event()

    def start(self):
        self.thread = threading.Thread(target=self.run)
        self.thread.start()

    def stop(self):
        self.stop_event.set()
        if self.thread:
            self.thread.join()

    def run(self):
        timer = 0
        while not self.stop_event.is_set():
            self.detect_tcp_attack()
            time.sleep(5)
            timer += 5
            if timer >= self.timeout:
                print('未检测到TCP攻击')
                self.stop_event.set()

    def detect_tcp_attack(self):
        tcp_request = IP(dst=self.target_ip)/TCP()
        tcp_reply = sr1(tcp_request, timeout=1, verbose=0)
        if tcp_reply:
            print(f'检测到TCP攻击: {tcp_reply.src} -> {tcp_reply.dst}')

class NetworkScanner:
    def __init__(self, interface, timeout=30):
        self.interface = interface
        self.timeout = timeout
        self.ip_mac_map = {}
        self.thread = None
        self.stop_event = threading.Event()

    def start(self):
        self.thread = threading.Thread(target=self.run)
        self.thread.start()

    def stop(self):
        self.stop_event.set()
        if self.thread:
            self.thread.join()

    def run(self):
        timer = 0
        while not self.stop_event.is_set() and timer < self.timeout:
            self.scan_network()
            time.sleep(5)
            timer += 5

    def scan_network(self):
        for ip in netifaces.ifaddresses(self.interface)[netifaces.AF_INET][0]['addr'].split('.')[:-1]:
            for i in range(1, 255):
                target_ip = f'{ip}.{i}'
                arp_request = ARP(pdst=target_ip)
                arp_reply = sr1(arp_request, timeout=1, verbose=0)
                if arp_reply and arp_reply.hwsrc not in ('00:00:00:00:00:00', get_if_hwaddr(self.interface)):
                    self.ip_mac_map[target_ip] = arp_reply.hwsrc
        self.show_result()

    def show_result(self):
        result = ''
        for ip, mac in self.ip_mac_map.items():
            result += f'{ip} ({mac})
'
        print('扫描结果:\n' + result)

class ArpDetectorPage(tk.Frame):
    def __init__(self, master):
        super().__init__(master)
        self.master = master
        self.frame = tk.Frame(self.master)
        self.frame.pack()

        self.status_label = tk.Label(self.frame, text='空闲')
        self.status_label.pack()

        self.start_button = tk.Button(self.frame, text='开始', command=self.start_detection)
        self.start_button.pack()

        self.stop_button = tk.Button(self.frame, text='停止', command=self.stop_detection, state=tk.DISABLED)
        self.stop_button.pack()

        self.arp_detector = None

    def start_detection(self):
        self.status_label.config(text='运行中')
        self.start_button.config(state=tk.DISABLED)
        self.stop_button.config(state=tk.NORMAL)

        self.arp_detector = ArpDetector('ens33')
        self.arp_detector.start()

    def stop_detection(self):
        self.status_label.config(text='空闲')
        self.start_button.config(state=tk.NORMAL)
        self.stop_button.config(state=tk.DISABLED)

        if self.arp_detector:
            self.arp_detector.stop()

class MainWindow(tk.Tk):
    def __init__(self):
        super().__init__()
        self.title('网络安全工具箱')
        self.geometry('400x350')

        self.arp_button = tk.Button(self, text='ARP欺骗检测', command=self.show_arp_page)
        self.arp_button.pack()

        self.icmp_button = tk.Button(self, text='ICMP泛洪检测', command=self.show_icmp_page)
        self.icmp_button.pack()

        self.tcp_button = tk.Button(self, text='TCP攻击检测', command=self.show_tcp_page)
        self.tcp_button.pack()

        self.network_button = tk.Button(self, text='网络扫描', command=self.show_network_page)
        self.network_button.pack()

        self.timeout_label = tk.Label(self, text='检测时间 (秒)')
        self.timeout_label.pack()
        self.timeout_entry = tk.Entry(self)
        self.timeout_entry.insert(0, '30')  # 默认检测时间为30秒
        self.timeout_entry.pack()

        self.set_timeout_button = tk.Button(self, text='设置检测时间', command=self.set_timeout)
        self.set_timeout_button.pack()

        self.status_label = tk.Label(self, text='空闲')
        self.status_label.pack()

        self.start_button = tk.Button(self, text='开始检测', command=self.start_detection)
        self.start_button.pack()

        self.stop_button = tk.Button(self, text='停止检测', command=self.stop_detection, state=tk.DISABLED)
        self.stop_button.pack()

        self.arp_detector = None
        self.icmp_flood_detector = None
        self.tcp_attack_detector = None
        self.network_scanner = None

    def show_arp_page(self):
        arp_page = tk.Toplevel(self)
        arp_page.title('ARP欺骗检测')
        ArpDetectorPage(arp_page)

    def show_icmp_page(self):
        icmp_page = tk.Toplevel(self)
        icmp_page.title('ICMP泛洪检测')
        IcmpFloodDetectorPage(icmp_page)

    def show_tcp_page(self):
        tcp_page = tk.Toplevel(self)
        tcp_page.title('TCP攻击检测')
        TcpAttackDetectorPage(tcp_page)

    def show_network_page(self):
        network_page = tk.Toplevel(self)
        network_page.title('网络扫描')
        NetworkScannerPage(network_page)

    def set_timeout(self):
        try:
            timeout = int(self.timeout_entry.get())
            if timeout > 0:
                self.timeout = timeout
                messagebox.showinfo('提示', f'检测时间已设置为{timeout}秒')
            else:
                messagebox.showerror('错误', '检测时间必须为正整数')
        except ValueError:
            messagebox.showerror('错误', '请输入有效的整数')

    def start_detection(self):
        timeout = int(self.timeout_entry.get())
        self.status_label.config(text='运行中')
        self.start_button.config(state=tk.DISABLED)
        self.stop_button.config(state=tk.NORMAL)
        self.arp_button.config(state=tk.DISABLED)
        self.icmp_button.config(state=tk.DISABLED)
        self.tcp_button.config(state=tk.DISABLED)
        self.network_button.config(state=tk.DISABLED)

        self.arp_detector = ArpDetector('ens33', timeout)
        self.arp_detector.start()

        self.icmp_flood_detector = IcmpFloodDetector('ens33', timeout)
        self.icmp_flood_detector.start()

        self.tcp_attack_detector = TcpAttackDetector('ens33', timeout)
        self.tcp_attack_detector.start()

        self.network_scanner = NetworkScanner('ens33', timeout)
        self.network_scanner.start()

    def stop_detection(self):
        self.status_label.config(text='空闲')
        self.start_button.config(state=tk.NORMAL)
        self.stop_button.config(state=tk.DISABLED)
        self.arp_button.config(state=tk.NORMAL)
        self.icmp_button.config(state=tk.NORMAL)
        self.tcp_button.config(state=tk.NORMAL)
        self.network_button.config(state=tk.NORMAL)

        if self.arp_detector:
            self.arp_detector.stop()
        if self.icmp_flood_detector:
            self.icmp_flood_detector.stop()
        if self.tcp_attack_detector:
            self.tcp_attack_detector.stop()
        if self.network_scanner:
            self.network_scanner.stop()

class LoginWindow(tk.Tk):
    def __init__(self):
        super().__init__()
        self.title('登录')
        self.geometry('300x150')

        self.users = [{'username': 'admin', 'password': 'admin'}],

        tk.Label(self, text='用户名').place(x=50, y=30)
        self.username_entry = tk.Entry(self)
        self.username_entry.place(x=120, y=30)

        tk.Label(self, text='密码').place(x=50, y=60)
        self.password_entry = tk.Entry(self, show='*')
        self.password_entry.place(x=120, y=60)

        self.login_button = tk.Button(self, text='登录', command=self.login)
        self.login_button.place(x=100, y=100)

        self.register_button = tk.Button(self, text='注册', command=self.register)
        self.register_button.place(x=170, y=100)

    def show_main_window(self):
        self.destroy()
        MainWindow().mainloop()

    def login(self):
        username = self.username_entry.get()
        password = self.password_entry.get()
        for user in self.users:
            if user['username'] == username and user['password'] == password:
                self.show_main_window()
                return
        messagebox.showinfo('错误', '用户名或密码错误')

    def register(self):
        username = self.username_entry.get()
        password = self.password_entry.get()
        for user in self.users:
            if user['username'] == username:
                messagebox.showerror('错误', '用户名已存在')
                return
        self.users.append({'username': username, 'password': password})
        messagebox.showinfo('提示', '注册成功，请登录')

if __name__ == '__main__':
    login_window = LoginWindow()
    login_window.mainloop()