import os import sys import time import socket import struct import fcntl import tkinter as tk from tkinter import messagebox

定义常量

ETH_P_ALL = 0x0003 ETH_LEN = 14 ARP_LEN = 28 ARP_REQUEST = 0x0001 ARP_REPLY = 0x0002 ICMP_ECHO_REQUEST = 0x08 TCP_SYN = 0x02

定义全局变量

ip_mac_dict = {} # 存储局域网内活动主机的IP地址和MAC地址 attack_source_dict = {} # 存储攻击源的IP地址和MAC地址

获取本机MAC地址

def get_mac_address(ifname): s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) info = fcntl.ioctl(s.fileno(), 0x8927, struct.pack('256s', bytes(ifname[:15], 'utf-8'))) return ':'.join(['%02x' % b for b in info[18:24]])

发送ARP请求

def send_arp_request(ifname, src_ip, src_mac, dst_ip): s = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(ETH_P_ALL)) s.bind((ifname, ETH_P_ALL)) dst_mac = 'ff:ff:ff:ff:ff:ff' # 广播地址 arp_packet = struct.pack('!6s6s2s2s1s1s2s6s4s6s4s', bytes(dst_mac.replace(':', ''), 'utf-8'), bytes(src_mac.replace(':', ''), 'utf-8'), bytes.fromhex('08 06'), bytes.fromhex('00 01'), bytes.fromhex('08'), bytes.fromhex('00'), bytes.fromhex('06 04'), bytes.fromhex('00 01'), socket.inet_aton(src_ip), bytes.fromhex('00 00 00 00 00 00'), socket.inet_aton(dst_ip)) s.send(arp_packet)

发送ARP响应

def send_arp_reply(ifname, src_ip, src_mac, dst_ip, dst_mac): arp_packet = struct.pack('!6s6s2s2s1s1s2s6s4s6s4s', bytes(dst_mac.replace(':', ''), 'utf-8'), bytes(src_mac.replace(':', ''), 'utf-8'), bytes.fromhex('08 06'), bytes.fromhex('00 02'), bytes.fromhex('08'), bytes.fromhex('00'), bytes.fromhex('06 04'), bytes.fromhex('00 02'), socket.inet_aton(src_ip), bytes.fromhex(dst_mac.replace(':', '')), socket.inet_aton(dst_ip)) s.send(arp_packet)

发送ICMP请求

def send_icmp_request(ifname, src_ip, src_mac, dst_ip): s = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(ETH_P_ALL)) s.bind((ifname, ETH_P_ALL)) dst_mac = ip_mac_dict[dst_ip] # 获取目标主机的MAC地址 icmp_packet = struct.pack('!6s6s2s1s1s2s4s4s', bytes(dst_mac.replace(':', ''), 'utf-8'), bytes(src_mac.replace(':', ''), 'utf-8'), bytes.fromhex('08 00'), bytes.fromhex('00'), bytes.fromhex('00'), bytes.fromhex('00 00'), socket.inet_aton(src_ip), socket.inet_aton(dst_ip)) s.send(icmp_packet)

发送ICMP响应

def send_icmp_reply(ifname, src_ip, src_mac, dst_ip): s = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(ETH_P_ALL)) s.bind((ifname, ETH_P_ALL)) dst_mac = ip_mac_dict[dst_ip] # 获取目标主机的MAC地址 icmp_packet = struct.pack('!6s6s2s1s1s2s4s4s', bytes(dst_mac.replace(':', ''), 'utf-8'), bytes(src_mac.replace(':', ''), 'utf-8'), bytes.fromhex('08 00'), bytes.fromhex('00'), bytes.fromhex('00'), bytes.fromhex('00 00'), socket.inet_aton(src_ip), socket.inet_aton(dst_ip)) s.send(icmp_packet)

发送TCP SYN

def send_tcp_syn(ifname, src_ip, src_mac, dst_ip, dst_port): s = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(ETH_P_ALL)) s.bind((ifname, ETH_P_ALL)) dst_mac = ip_mac_dict[dst_ip] # 获取目标主机的MAC地址 tcp_packet = struct.pack('!6s6s2s2s2s1s1s2s4s4s2s2s', bytes(dst_mac.replace(':', ''), 'utf-8'), bytes(src_mac.replace(':', ''), 'utf-8'), bytes.fromhex('08 00'), bytes.fromhex('45'), bytes.fromhex('00 28'), bytes.fromhex('00 00'), bytes.fromhex('40 00'), bytes.fromhex('40 06'), bytes.fromhex('00 00 00 00'), socket.inet_aton(src_ip), socket.inet_aton(dst_ip), bytes.fromhex('00 00'), bytes.fromhex('00 00'), bytes.fromhex('00 00'), bytes.fromhex('00 00'), bytes.fromhex('50 02'), bytes.fromhex('00 00'), bytes.fromhex('00 00')) tcp_packet += struct.pack('!H', dst_port) tcp_packet += struct.pack('!H', 0x1234) tcp_packet += struct.pack('!I', 0) tcp_packet += struct.pack('!I', 0) tcp_packet += bytes.fromhex('50 02 20 00') s.send(tcp_packet)

解析ARP包

def parse_arp_packet(packet): src_mac = ':'.join(['%02x' % b for b in packet[6:12]]) src_ip = socket.inet_ntoa(packet[28:32]) dst_ip = socket.inet_ntoa(packet[38:42]) arp_op = struct.unpack('!H', packet[20:22])[0] if arp_op == ARP_REQUEST: send_arp_reply(ifname, dst_ip, get_mac_address(ifname), src_ip, src_mac) elif arp_op == ARP_REPLY: ip_mac_dict[src_ip] = src_mac

解析ICMP包

def parse_icmp_packet(packet): src_mac = ':'.join(['%02x' % b for b in packet[6:12]]) src_ip = socket.inet_ntoa(packet[26:30]) dst_ip = socket.inet_ntoa(packet[30:34]) icmp_type = struct.unpack('!B', packet[34:35])[0] icmp_code = struct.unpack('!B', packet[35:36])[0] if icmp_type == ICMP_ECHO_REQUEST: send_icmp_reply(ifname, dst_ip, get_mac_address(ifname), src_ip)

解析TCP包

def parse_tcp_packet(packet): src_mac = ':'.join(['%02x' % b for b in packet[6:12]]) src_ip = socket.inet_ntoa(packet[26:30]) dst_ip = socket.inet_ntoa(packet[30:34]) tcp_flags = struct.unpack('!B', packet[47:48])[0] if tcp_flags == TCP_SYN: attack_source_dict[src_ip] = src_mac

监听网络数据包

def sniff_packets(ifname): s = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(ETH_P_ALL)) s.bind((ifname, ETH_P_ALL)) while True: packet = s.recvfrom(2048)[0] if len(packet) < ETH_LEN: continue eth_type = struct.unpack('!H', packet[12:14])[0] if eth_type == 0x0806: parse_arp_packet(packet) elif eth_type == 0x0800: ip_proto = struct.unpack('!B', packet[23:24])[0] if ip_proto == 0x01: parse_icmp_packet(packet) elif ip_proto == 0x06: parse_tcp_packet(packet)

检测ICMP Flood攻击

def detect_icmp_flood(): if len(ip_mac_dict) == 0: return '并没有遭到ICMP Flood攻击' for ip in ip_mac_dict.keys(): count = 0 for src_ip in attack_source_dict.keys(): if src_ip == ip: count += 1 if count > 100: return '遭到ICMP Flood攻击' return '并没有遭到ICMP Flood攻击'

检测TCP攻击

def detect_tcp_attack(): if len(ip_mac_dict) == 0: return '并没有遭到TCP攻击' for ip in ip_mac_dict.keys(): count = 0 for src_ip in attack_source_dict.keys(): if src_ip == ip: count += 1 if count > 10: return '遭到TCP攻击' return '并没有遭到TCP攻击'

检测ARP欺骗攻击

def detect_arp_spoofing(): if len(ip_mac_dict) == 0: return '并没有遭到ARP欺骗攻击' for ip, mac in ip_mac_dict.items(): if mac != get_mac_address(ifname): return '遭到ARP欺骗攻击' return '并没有遭到ARP欺骗攻击'

扫描局域网内活动主机的IP地址和MAC地址

def scan_network(ifname): active_hosts = {} for i in range(1, 255): dst_ip = '192.168.197.' + str(i) send_arp_request(ifname, '192.168.197.1', get_mac_address(ifname), dst_ip) time.sleep(0.1) response = send_arp_reply(ifname, timeout=1) if response is not None: active_hosts[response['src_ip']] = response['src_mac'] return active_hosts

保存记录到本地文件

def save_records(): active_hosts = scan_network(ifname) with open('records.txt', 'w') as f: f.write('活动主机IP地址 MAC地址 ') for ip, mac in ip_mac_dict.items(): f.write(ip + ' ' + mac + ' ') f.write(' 攻击源IP地址 MAC地址 ') for ip, mac in attack_source_dict.items(): f.write(ip + ' ' + mac + ' ') f.write(' ICMP Flood攻击：' + detect_icmp_flood() + ' ') f.write('TCP攻击：' + detect_tcp_attack() + ' ') f.write('ARP欺骗攻击：' + detect_arp_spoofing() + ' ') f.write('')

创建可视化界面

def create_gui(): root = tk.Tk() root.title('ARP攻击检测防护软件') root.geometry('300x200')

# 扫描按钮
def scan_button_click():
    scan_network(ifname)
    messagebox.showinfo('提示', '扫描完成')

scan_button = tk.Button(root, text='扫描', command=scan_button_click)
scan_button.pack(pady=10)

# 检测按钮
def detect_button_click():
    result = 'IP地址		MAC地址

' for ip, mac in ip_mac_dict.items(): result += ip + ' ' + mac + ' ' result += ' 攻击源IP地址 MAC地址 ' for ip, mac in attack_source_dict.items(): result += ip + ' ' + mac + ' ' result += ' ICMP Flood攻击：' + detect_icmp_flood() + ' ' result += 'TCP攻击：' + detect_tcp_attack() + ' ' result += 'ARP欺骗攻击：' + detect_arp_spoofing() + ' ' messagebox.showinfo('检测结果', result)

detect_button = tk.Button(root, text='检测', command=detect_button_click)
detect_button.pack(pady=10)

# 保存按钮
def save_button_click():
    save_records()
    messagebox.showinfo('提示', '记录已保存到本地文件')

save_button = tk.Button(root, text='保存', command=save_button_click)
save_button.pack(pady=10)

root.mainloop()

主函数

if name == 'main': if len(sys.argv) < 2: print('Usage: python arp_attack_detection.py ') sys.exit(1) ifname = sys.argv[1] create_gui()