Python ARP攻击检测软件 - 实时监控局域网安全

以下是一个基于Python实现的ARP攻击检测软件的示例代码，其中包含了上述规则的实现：

import os
import sys
import time
import logging
import tkinter as tk
from tkinter import messagebox
from scapy.all import *
from collections import defaultdict

# 全局变量
ar_requests = defaultdict(set)
ar_responses = defaultdict(set)
active_hosts = defaultdict(str)
logfile = 'arp_attack.log'

# 初始化日志记录
logging.basicConfig(filename=logfile, level=logging.INFO)

# 捕获ARP包并进行分析
def capture_arp():
    def arp_capture_callback(packet):
        if packet[ARP].op == ARP.who_has:
            arp_requests[packet[ARP].psrc].add(packet[ARP].hwsrc)
        elif packet[ARP].op == ARP.is_at:
            arp_responses[packet[ARP].psrc].add(packet[ARP].hwsrc)
        else:
            pass

    sniff(prn=arp_capture_callback, filter='arp', store=0)

# 判断是否遭受ARP攻击
def detect_arp_attack():
    arp_attack = False

    # 检测应答报文
    for ip, macs in arp_responses.items():
        if len(macs) > 1:
            logging.info('IP地址 {} 对应不同的MAC地址 {}'.format(ip, macs))
            messagebox.showerror('ARP攻击检测', 'IP地址 {} 可能遭受ARP攻击'.format(ip))
            arp_attack = True

    # 检测请求报文
    for ip, macs in arp_requests.items():
        if len(macs) > 1:
            logging.info('不同IP地址 {} 对应同一个MAC地址 {}'.format(ip, macs))
            messagebox.showerror('ARP攻击检测', 'IP地址 {} 可能遭受ARP攻击'.format(ip))
            arp_attack = True

    # 检测ARP请求或响应包
    for mac, ips in active_hosts.items():
        if len(ips) > 100:
            logging.info('MAC地址 {} 发出大量数据包'.format(mac))
            messagebox.showerror('ARP攻击检测', 'IP地址 {} 可能遭受ARP攻击'.format(ip))
            arp_attack = True

    if not arp_attack:
        logging.info('未遭受ARP攻击')
        messagebox.showinfo('ARP攻击检测', '未遭受ARP攻击')

# 统计ARP响应报文数量
def count_arp_responses():
    arp_responses_count = defaultdict(int)
    for ip, macs in arp_responses.items():
        arp_responses_count[len(macs)] += 1

    result = 'ARP响应报文数量统计结果：\n'
    for count, num in arp_responses_count.items():
        result += '{} 个IP地址对应 {} 个MAC地址，数量为 {} 个\n'.format(num, count, num*count)

    messagebox.showinfo('ARP响应报文数量统计', result)
    logging.info(result)

# 统计ARP请求报文数量
def count_arp_requests():
    arp_requests_count = defaultdict(int)
    for ip, macs in arp_requests.items():
        arp_requests_count[len(macs)] += 1

    result = 'ARP请求报文数量统计结果：\n'
    for count, num in arp_requests_count.items():
        result += '{} 个IP地址请求 {} 个MAC地址，数量为 {} 个\n'.format(num, count, num*count)

    messagebox.showinfo('ARP请求报文数量统计', result)
    logging.info(result)

# 扫描局域网内的活动主机
def scan_active_hosts():
    active_hosts.clear()
    ans, unans = srp(Ether(dst='ff:ff:ff:ff:ff:ff')/ARP(pdst='192.168.1.0/24'), timeout=2, verbose=False)
    for send, recv in ans:
        active_hosts[recv[Ether].src].add(recv[ARP].psrc)

    result = '局域网内活动主机：\n'
    for mac, ips in active_hosts.items():
        result += 'MAC地址 {} 对应的IP地址有：{}\n'.format(mac, ', '.join(ips))

    messagebox.showinfo('扫描局域网活动主机', result)
    logging.info(result)

# 用户登录
def login():
    username = username_entry.get()
    password = password_entry.get()
    with open('users.txt', 'r') as f:
        for line in f:
            user, pwd = line.strip().split(',')
            if user == username:
                if pwd == password:
                    login_window.destroy()
                    main_window.deiconify()
                    return
                else:
                    messagebox.showerror('登录失败', '用户名或密码错误')
                    return
    messagebox.showerror('登录失败', '查无该用户')

# 用户注册
def register():
    username = username_entry.get()
    password = password_entry.get()
    with open('users.txt', 'r') as f:
        for line in f:
            user, pwd = line.strip().split(',')
            if user == username:
                messagebox.showerror('注册失败', '该用户已注册')
                return
    with open('users.txt', 'a') as f:
        f.write('{},{}
'.format(username, password))
    messagebox.showinfo('注册成功', '注册成功，请登录')

# 创建登录窗口
login_window = tk.Tk()
login_window.title('ARP攻击检测软件 - 登录')
login_window.geometry('300x150')
login_window.resizable(False, False)

username_label = tk.Label(login_window, text='用户名：')
username_label.grid(row=0, column=0, padx=10, pady=10)
username_entry = tk.Entry(login_window)
username_entry.grid(row=0, column=1, padx=10, pady=10)

password_label = tk.Label(login_window, text='密码：')
password_label.grid(row=1, column=0, padx=10, pady=10)
password_entry = tk.Entry(login_window, show='*')
password_entry.grid(row=1, column=1, padx=10, pady=10)

login_button = tk.Button(login_window, text='登录', command=login)
login_button.grid(row=2, column=0, padx=10, pady=10)

register_button = tk.Button(login_window, text='注册', command=register)
register_button.grid(row=2, column=1, padx=10, pady=10)

# 创建主窗口
main_window = tk.Tk()
main_window.title('ARP攻击检测软件')
main_window.geometry('600x400')
main_window.resizable(False, False)
main_window.withdraw()

# 创建攻击检测界面
def attack_detection():
    def analyze_responses():
        result = '同一IP地址对应不同的MAC地址：\n'
        for ip, macs in arp_responses.items():
            if len(macs) > 1:
                result += 'IP地址 {} 对应的MAC地址有：{}\n'.format(ip, ', '.join(macs))
        messagebox.showinfo('分析响应报文', result)
        logging.info(result)

    def analyze_requests():
        result = '不同IP地址对应同一个MAC地址：\n'
        for ip, macs in arp_requests.items():
            if len(macs) > 1:
                result += 'MAC地址 {} 对应的IP地址有：{}\n'.format(', '.join(macs), ip)
        messagebox.showinfo('分析请求报文', result)
        logging.info(result)

    def stop_detection():
        arp_requests.clear()
        arp_responses.clear()
        messagebox.showinfo('停止攻击检测', '攻击检测已停止')

    def count_packets():
        count_window = tk.Toplevel(attack_window)
        count_window.title('统计ARP报文数量')
        count_window.geometry('400x200')
        count_window.resizable(False, False)

        arp_responses_button = tk.Button(count_window, text='计算ARP响应报文数量', command=count_arp_responses)
        arp_responses_button.pack(padx=10, pady=10)

        arp_requests_button = tk.Button(count_window, text='计算ARP请求报文数量', command=count_arp_requests)
        arp_requests_button.pack(padx=10, pady=10)

        back_button = tk.Button(count_window, text='返回功能界面', command=count_window.destroy)
        back_button.pack(padx=10, pady=10)

    attack_window = tk.Toplevel(main_window)
    attack_window.title('ARP攻击检测 - 攻击检测')
    attack_window.geometry('600x400')
    attack_window.resizable(False, False)

    analyze_responses_button = tk.Button(attack_window, text='分析响应报文', command=analyze_responses)
    analyze_responses_button.pack(padx=10, pady=10)

    analyze_requests_button = tk.Button(attack_window, text='分析请求报文', command=analyze_requests)
    analyze_requests_button.pack(padx=10, pady=10)

    back_button = tk.Button(attack_window, text='返回功能界面', command=attack_window.destroy)
    back_button.pack(padx=10, pady=10)

    stop_button = tk.Button(attack_window, text='停止攻击检测', command=stop_detection)
    stop_button.pack(padx=10, pady=10)

    count_button = tk.Button(attack_window, text='统计ARP报文数量', command=count_packets)
    count_button.pack(padx=10, pady=10)

    capture_thread = threading.Thread(target=capture_arp)
    capture_thread.start()

    detect_thread = threading.Thread(target=detect_arp_attack)
    detect_thread.start()

# 创建扫描局域网活动主机界面
def scan_active_hosts_window():
    scan_window = tk.Toplevel(main_window)
    scan_window.title('ARP攻击检测 - 扫描局域网主机活动')
    scan_window.geometry('600x400')
    scan_window.resizable(False, False)

    scan_button = tk.Button(scan_window, text='扫描活动主机', command=scan_active_hosts)
    scan_button.pack(padx=10, pady=10)

    back_button = tk.Button(scan_window, text='返回功能界面', command=scan_window.destroy)
    back_button.pack(padx=10, pady=10)

    result_text = tk.Text(scan_window)
    result_text.pack(padx=10, pady=10)

    def update_result():
        result_text.delete(1.0, tk.END)
        for mac, ips in active_hosts.items():
            result_text.insert(tk.END, 'MAC地址 {} 对应的IP地址有：{}\n'.format(mac, ', '.join(ips)))
        result_text.after(1000, update_result)

    update_result()

# 创建功能界面
attack_detection_button = tk.Button(main_window, text='开始攻击检测', command=attack_detection)
attack_detection_button.pack(padx=10, pady=10)

scan_active_hosts_button = tk.Button(main_window, text='扫描局域网主机活动', command=scan_active_hosts_window)
scan_active_hosts_button.pack(padx=10, pady=10)

# 运行主程序
login_window.mainloop()