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=5):
self.interface = interface
self.ip_mac_map = {}
self.attacker_ip = None
self.attacker_mac = None
self.thread = None
self.stop_event = threading.Event()
self.timeout = timeout
self.attack_count = 0
def start(self):
self.thread = threading.Thread(target=self.run)
self.thread.start()
def stop(self):
self.stop_event.set()
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'Attacker IP: {self.attacker_ip}, MAC: {self.attacker_mac}')
while not self.stop_event.is_set():
self.attack_count = 0 # 重置攻击计数器
self.scan_network()
self.detect_arp_spoofing()
if self.attack_count == 0:
print(f'在{self.timeout}秒内未检测到ARP欺骗攻击')
time.sleep(self.timeout)
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}')
self.attack_count += 1
class IcmpFloodDetector:
def __init__(self, interface, timeout=5):
self.interface = interface
self.attacker_ip = None
self.thread = None
self.stop_event = threading.Event()
self.timeout = timeout
self.attack_count = 0
def start(self):
self.thread = threading.Thread(target=self.run)
self.thread.start()
def stop(self):
self.stop_event.set()
self.thread.join()
def run(self):
self.attacker_ip = netifaces.ifaddresses(self.interface)[netifaces.AF_INET][0]['addr']
print(f'Attacker IP: {self.attacker_ip}')
while not self.stop_event.is_set():
self.attack_count = 0 # 重置攻击计数器
self.detect_icmp_flood()
if self.attack_count == 0:
print(f'在{self.timeout}秒内未检测到ICMP泛洪攻击')
time.sleep(self.timeout)
def detect_icmp_flood(self):
icmp_packets = sniff(filter=f'icmp and src host {self.attacker_ip}', timeout=1, count=10)
if len(icmp_packets) == 10:
print('检测到ICMP泛洪攻击')
self.attack_count += 1
class TcpAttackDetector:
def __init__(self, interface, timeout=5):
self.interface = interface
self.attacker_ip = None
self.thread = None
self.stop_event = threading.Event()
self.timeout = timeout
self.attack_count = 0
def start(self):
self.thread = threading.Thread(target=self.run)
self.thread.start()
def stop(self):
self.stop_event.set()
self.thread.join()
def run(self):
self.attacker_ip = netifaces.ifaddresses(self.interface)[netifaces.AF_INET][0]['addr']
print(f'Attacker IP: {self.attacker_ip}')
while not self.stop_event.is_set():
self.attack_count = 0 # 重置攻击计数器
self.detect_tcp_attack()
if self.attack_count == 0:
print(f'在{self.timeout}秒内未检测到TCP攻击')
time.sleep(self.timeout)
def detect_tcp_attack(self):
tcp_packets = sniff(filter=f'tcp and src host {self.attacker_ip}', timeout=1, count=10)
if len(tcp_packets) == 10:
print('检测到TCP攻击')
self.attack_count += 1
class NetworkScanner:
def __init__(self, interface):
self.interface = interface
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()
self.thread.join()
def run(self):
while not self.stop_event.is_set():
self.scan_network()
time.sleep(10)
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 not in self.ip_mac_map:
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', 'ff:ff:ff:ff:ff:ff'):
self.ip_mac_map[target_ip] = arp_reply.hwsrc
print(f'发现设备: {target_ip} ({arp_reply.hwsrc})')
self.save_to_file(target_ip, arp_reply.hwsrc)
def save_to_file(self, ip, mac):
with open('network_scan.txt', 'a') as f:
f.write(f'{ip},{mac}
')
class LoginWindow:
def __init__(self):
self.root = tk.Tk()
self.root.title('登录')
self.root.geometry('300x150')
tk.Label(self.root, text='用户名').place(x=50, y=30)
self.username_entry = tk.Entry(self.root)
self.username_entry.place(x=120, y=30)
tk.Label(self.root, text='密码').place(x=50, y=60)
self.password_entry = tk.Entry(self.root, show='*')
self.password_entry.place(x=120, y=60)
self.login_button = tk.Button(self.root, text='登录', command=self.login)
self.login_button.place(x=100, y=100)
self.register_button = tk.Button(self.root, text='注册', command=self.register)
self.register_button.place(x=170, y=100)
self.root.mainloop()
def show_main_window(self):
self.root.destroy()
MainWindow()
def login(self):
username = self.username_entry.get()
password = self.password_entry.get()
for user in 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 users:
if user['username'] == username:
# 用户名已存在,弹出错误提示
messagebox.showerror('错误', '用户名已存在')
return
# 用户名不存在,将新用户添加到用户列表中
users.append({'username': username, 'password': password})
# 注册成功,弹出成功提示
messagebox.showinfo('提示', '注册成功,请登录')
class MainWindow:
def __init__(self):
self.root = tk.Tk()
self.root.title('网络安全工具')
self.root.geometry('400x300')
self.status_label = tk.Label(self.root, text='空闲')
self.status_label.pack()
self.start_button = tk.Button(self.root, text='全部开始', command=self.start_detection)
self.start_button.pack()
self.stop_button = tk.Button(self.root, text='全部停止', command=self.stop_detection, state=tk.DISABLED)
self.stop_button.pack()
self.arp_button = tk.Button(self.root, text='ARP欺骗检测', command=self.toggle_arp_detection, state=tk.DISABLED)
self.arp_button.pack()
self.icmp_button = tk.Button(self.root, text='ICMP泛洪检测', command=self.toggle_icmp_detection, state=tk.DISABLED)
self.icmp_button.pack()
self.tcp_button = tk.Button(self.root, text='TCP攻击检测', command=self.toggle_tcp_detection, state=tk.DISABLED)
self.tcp_button.pack()
self.network_button = tk.Button(self.root, text='网络扫描器', command=self.toggle_network_scanner, state=tk.DISABLED)
self.network_button.pack()
self.root.mainloop()
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_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)
self.arp_detector = ArpDetector('ens33')
self.arp_detector.start()
self.icmp_flood_detector = IcmpFloodDetector('ens33')
self.icmp_flood_detector.start()
self.tcp_attack_detector = TcpAttackDetector('ens33')
self.tcp_attack_detector.start()
self.network_scanner = NetworkScanner('ens33')
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.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.stop()
self.icmp_flood_detector.stop()
self.tcp_attack_detector.stop()
self.network_scanner.stop()
def toggle_arp_detection(self):
if self.arp_detector.thread is None or not self.arp_detector.thread.is_alive():
self.arp_detector = ArpDetector('ens33')
self.arp_detector.start()
self.arp_button.config(text='停止ARP欺骗检测')
else:
self.arp_detector.stop()
self.arp_button.config(text='开始ARP欺骗检测')
def toggle_icmp_detection(self):
if self.icmp_flood_detector.thread is None or not self.icmp_flood_detector.thread.is_alive():
self.icmp_flood_detector = IcmpFloodDetector('ens33')
self.icmp_flood_detector.start()
self.icmp_button.config(text='停止ICMP泛洪检测')
else:
self.icmp_flood_detector.stop()
self.icmp_button.config(text='开始ICMP泛洪检测')
def toggle_tcp_detection(self):
if self.tcp_attack_detector.thread is None or not self.tcp_attack_detector.thread.is_alive():
self.tcp_attack_detector = TcpAttackDetector('ens33')
self.tcp_attack_detector.start()
self.tcp_button.config(text='停止TCP攻击检测')
else:
self.tcp_attack_detector.stop()
self.tcp_button.config(text='开始TCP攻击检测')
def toggle_network_scanner(self):
if self.network_scanner.thread is None or not self.network_scanner.thread.is_alive():
self.network_scanner = NetworkScanner('ens33')
self.network_scanner.start()
self.network_button.config(text='停止网络扫描器')
else:
self.network_scanner.stop()
self.network_button.config(text='开始网络扫描器')
if __name__ == '__main__':
users = [{'username': 'admin', 'password': 'admin'}]
login_window = LoginWindow()
# main_window = MainWindow() # 不应该在登录窗口后直接创建主窗口
原文地址: https://www.cveoy.top/t/topic/jnVi 著作权归作者所有。请勿转载和采集!