Python实现ARP欺骗检测

import os
import sys
import time
import socket
import struct
import fcntl
import argparse
import threading

# 定义常量
ETH_P_ALL = 0x0003
ETH_HEADER_LEN = 14
ARP_HEADER_LEN = 28
ARP_REQUEST = 1
ARP_REPLY = 2

# 定义全局变量
ip_mac_dict = {}
attack_dict = {}

# 获取本机IP地址
def get_ip_address(ifname):
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    return socket.inet_ntoa(fcntl.ioctl(
        s.fileno(),
        0x8915,  # SIOCGIFADDR
        struct.pack('256s', ifname[:15].encode())
    )[20:24])

# 获取本机MAC地址
def get_mac_address(ifname):
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    info = fcntl.ioctl(s.fileno(), 0x8927, struct.pack('256s', ifname[:15].encode()))
    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))
    src_mac_bytes = bytes.fromhex(src_mac.replace(':', ''))
    src_ip_bytes = socket.inet_aton(src_ip)
    dst_ip_bytes = socket.inet_aton(dst_ip)
    eth_header = struct.pack('!6s6sH', b'\xff\xff\xff\xff\xff\xff', src_mac_bytes, ETH_P_ALL)
    arp_header = struct.pack('!HHBBH6s4s6s4s', 0x0001, 0x0800, 6, 4, ARP_REQUEST, src_mac_bytes, src_ip_bytes, b'\x00\x00\x00\x00\x00\x00', dst_ip_bytes)
    packet = eth_header + arp_header
    s.send(packet)

# 发送ARP响应
def send_arp_reply(ifname, src_ip, src_mac, dst_ip, dst_mac):
    s = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(ETH_P_ALL))
    s.bind((ifname, ETH_P_ALL))
    src_mac_bytes = bytes.fromhex(src_mac.replace(':', ''))
    src_ip_bytes = socket.inet_aton(src_ip)
    dst_mac_bytes = bytes.fromhex(dst_mac.replace(':', ''))
    dst_ip_bytes = socket.inet_aton(dst_ip)
    eth_header = struct.pack('!6s6sH', dst_mac_bytes, src_mac_bytes, ETH_P_ALL)
    arp_header = struct.pack('!HHBBH6s4s6s4s', 0x0001, 0x0800, 6, 4, ARP_REPLY, src_mac_bytes, src_ip_bytes, dst_mac_bytes, dst_ip_bytes)
    packet = eth_header + arp_header
    s.send(packet)

# 处理ARP请求
def handle_arp_request(packet, ifname, my_mac):
    eth_header = packet[:ETH_HEADER_LEN]
    arp_header = packet[ETH_HEADER_LEN:ETH_HEADER_LEN+ARP_HEADER_LEN]
    src_mac = ':'.join(['%02x' % b for b in eth_header[6:12]])
    src_ip = socket.inet_ntoa(arp_header[8:12])
    dst_ip = socket.inet_ntoa(arp_header[20:24])
    send_arp_reply(ifname, dst_ip, my_mac, src_ip, src_mac)

# 处理ARP响应
def handle_arp_reply(packet):
    eth_header = packet[:ETH_HEADER_LEN]
    arp_header = packet[ETH_HEADER_LEN:ETH_HEADER_LEN+ARP_HEADER_LEN]
    src_mac = ':'.join(['%02x' % b for b in eth_header[6:12]])
    src_ip = socket.inet_ntoa(arp_header[8:12])
    dst_mac = ':'.join(['%02x' % b for b in arp_header[18:24]])
    dst_ip = socket.inet_ntoa(arp_header[20:24])
    ip_mac_dict[src_ip] = src_mac
    ip_mac_dict[dst_ip] = dst_mac

# 处理ARP数据包
def handle_arp_packet(packet, ifname, my_mac):
    eth_header = packet[:ETH_HEADER_LEN]
    eth_type = struct.unpack('!H', eth_header[12:14])[0]
    if eth_type == 0x0806:
        arp_op = struct.unpack('!H', packet[ETH_HEADER_LEN+6:ETH_HEADER_LEN+8])[0]
        if arp_op == ARP_REQUEST:
            handle_arp_request(packet, ifname, my_mac)
        elif arp_op == ARP_REPLY:
            handle_arp_reply(packet)

# 监听ARP数据包
def listen_arp_packets(ifname, my_mac):
    s = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(ETH_P_ALL))
    s.bind((ifname, ETH_P_ALL))
    while True:
        packet = s.recv(2048)
        handle_arp_packet(packet, ifname, my_mac)

# 检测ARP欺骗攻击
def detect_arp_spoofing(ifname, interval, my_ip, my_mac):
    while True:
        for ip, mac in ip_mac_dict.items():
            if ip != my_ip and mac != my_mac:
                if ip in attack_dict:
                    attack_dict[ip] += 1
                else:
                    attack_dict[ip] = 1
                if attack_dict[ip] > 10:
                    print('检测到ARP欺骗攻击，攻击者IP地址：', ip, '攻击者MAC地址：', mac)
                    attack_dict[ip] = 0
        time.sleep(interval)

# 主函数
if __name__ == '__main__':
    # 解析命令行参数
    parser = argparse.ArgumentParser()
    parser.add_argument('-i', '--interface', dest='ifname', required=True, help='network interface name')
    parser.add_argument('-t', '--interval', dest='interval', type=int, default=5, help='detection interval')
    args = parser.parse_args()

    # 获取本机IP地址和MAC地址
    my_ip = get_ip_address(args.ifname)
    my_mac = get_mac_address(args.ifname)

    # 启动ARP监听线程和ARP欺骗检测线程
    arp_listen_thread = threading.Thread(target=listen_arp_packets, args=(args.ifname, my_mac))
    arp_detect_thread = threading.Thread(target=detect_arp_spoofing, args=(args.ifname, args.interval, my_ip, my_mac))
    arp_listen_thread.start()
    arp_detect_thread.start()