• 日期: 2028-02-21
  • 标签: 常规

ARP攻击检测与防御：Python实现与安全防护指南

引言

随着互联网的普及，网络安全问题日益突出，其中ARP攻击作为一种常见的网络攻击手段，对个人隐私和网络安全构成严重威胁。本文将介绍如何使用Python开发ARP攻击检测与防御软件，旨在帮助读者了解ARP攻击的原理和危害，并提供相应的防御措施。

ARP攻击原理

ARP（地址解析协议）攻击是指攻击者通过伪造ARP报文，将攻击者的MAC地址与目标主机的IP地址绑定，从而拦截目标主机与网关之间的数据流量，实现窃取信息、篡改数据等恶意目的。常见的ARP攻击类型包括ARP欺骗、ARP泛洪等。

Python实现ARP攻击检测与防御

1. 环境准备

  • Python 3.x
  • scapy库
  • tkinter库

2. 代码实现

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)
    print('Sent ARP request to', dst_ip)
    response = s.recvfrom(65535)[0]
    eth_header = struct.unpack('!6s6sH', response[:14])
    if eth_header[2] == ETH_P_ALL:
        arp_header = struct.unpack('2s2s1s1s2s6s4s6s4s', response[14:42])
        if arp_header[4] == bytes.fromhex('00 02'):
            return {'src_ip': socket.inet_ntoa(arp_header[6]), 'src_mac': ':'.join(['%02x' % b for b in arp_header[5]])}
    return None

# 发送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))
    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.fromhex(dst_mac.replace(':', '')), bytes.fromhex(src_mac.replace(':', '')), 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, dst_ip, get_mac_address(ifname), '192.168.197.1', get_mac_address(ifname))
        if response is not None:
            active_hosts[response['src_ip']] = response['src_mac']
    return active_hosts

# 保存记录到本地文件
ifname = 'ens33'
active_hosts = scan_network(ifname)
def save_records():
    active_hosts = scan_network(ifname)
    print('Active hosts:', active_hosts)  # 添加打印语句
    with open('records.txt', 'w') as f:
        f.write('活动主机IP地址		MAC地址
')
        for ip, mac in active_hosts.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():
        global ip_mac_dict
        ip_mac_dict = scan_network(ifname)
        messagebox.showinfo('提示', '扫描完成')

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

    # 检测按钮
    def detect_button_click():
        global attack_source_dict
        attack_source_dict = {}
        for ip, mac in ip_mac_dict.items():
            if mac != get_mac_address(ifname):
                send_arp_request(ifname, '192.168.197.1', get_mac_address(ifname), ip)
                time.sleep(0.1)
                response = send_arp_reply(ifname, timeout=1)
                if response is not None and response['src_mac'] != mac:
                    attack_source_dict[ip] = mac
        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 <interface>')
        sys.exit(1)
    create_gui()
    save_records()

3. 代码说明

  • 导入模块: 导入所需的Python模块，包括os、sys、time、socket、struct、fcntl、tkinter等。

  • 定义常量: 定义网络协议相关的常量，例如以太网类型、ARP操作码、ICMP类型等。

  • 定义全局变量: 定义存储活动主机信息和攻击源信息的全局变量。

  • 获取本机MAC地址: 定义函数get_mac_address，用于获取指定网卡的MAC地址。

  • 发送ARP请求和响应: 定义函数send_arp_request和send_arp_reply，用于发送ARP请求和响应报文。

  • 发送ICMP请求和响应: 定义函数send_icmp_request和send_icmp_reply，用于发送ICMP请求和响应报文。

  • 发送TCP SYN: 定义函数send_tcp_syn，用于发送TCP SYN报文。

  • 解析数据包: 定义函数parse_arp_packet、parse_icmp_packet和parse_tcp_packet，用于解析ARP、ICMP和TCP数据包，并根据不同的协议类型进行相应的处理。

  • 监听网络数据包: 定义函数sniff_packets，用于监听指定网卡上的网络数据包，并调用相应的数据包解析函数进行处理。

  • 检测网络攻击: 定义函数detect_icmp_flood、detect_tcp_attack和detect_arp_spoofing，用于检测ICMP Flood攻击、TCP攻击和ARP欺骗攻击。

  • 扫描局域网: 定义函数scan_network，用于扫描局域网内活动主机的IP地址和MAC地址。

  • 保存记录: 定义函数save_records，用于将扫描结果和检测结果保存到本地文件。

  • 创建可视化界面: 定义函数create_gui，用于创建简单的图形用户界面，方便用户进行扫描、检测和保存操作。

  • 主函数: 程序入口点，调用create_gui函数创建图形用户界面，并调用save_records函数保存初始记录。

安全防护建议

  • 部署ARP防火墙: 使用专业的ARP防火墙软件，对网络中的ARP流量进行监控和过滤，防止ARP欺骗攻击。

  • 静态绑定ARP表项: 将重要主机的IP地址和MAC地址进行静态绑定，避免ARP缓存被攻击者篡改。

  • 使用网络安全工具: 使用网络安全扫描工具定期扫描网络，及时发现并修复安全漏洞。

  • 提高安全意识: 加强网络安全意识教育，避免点击不明链接或打开可疑附件，防止恶意软件入侵。

总结

本文介绍了如何使用Python开发ARP攻击检测与防御软件，并提供了一些安全防护建议，希望能够帮助读者提升网络安全意识，有效防御ARP攻击。

ARP攻击检测与防御：Python实现与安全防护指南

原文地址: https://www.cveoy.top/t/topic/jnbX 著作权归作者所有。请勿转载和采集!

免费AI点我,无需注册和登录