并简洁。

首先,我们需要导入所需的模块:

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

然后,我们定义一些常量和全局变量:

# ARP欺骗攻击检测的缓存
arp_cache = {}

# ICMP flood攻击检测的缓存
icmp_cache = {}

# TCP攻击检测的缓存
tcp_cache = {}

# ARP欺骗攻击检测的阈值
arp_threshold = 10

# ICMP flood攻击检测的阈值
icmp_threshold = 1000

# TCP攻击检测的阈值
tcp_threshold = 1000

# 局域网内活动主机的IP地址和MAC地址
active_hosts = []

# 记录攻击源的IP地址和MAC地址的文件名
attack_log_file = 'attack.log'

# 是否开启ARP欺骗攻击检测
detect_arp = True

# 是否开启ICMP flood攻击检测
detect_icmp = True

# 是否开启TCP攻击检测
detect_tcp = True

接下来,我们定义一些函数来实现具体的功能:

  1. 获取本机IP地址和MAC地址
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])


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]])
  1. 发送ARP请求获取主机MAC地址
def get_mac_address_from_ip(ip_address):
    try:
        ans, unans = srp(Ether(dst='ff:ff:ff:ff:ff:ff')/ARP(pdst=ip_address), timeout=2, retry=2)
        for s, r in ans:
            return r[Ether].src
    except:
        pass
    return None
  1. 检测ARP欺骗攻击
def detect_arp_spoofing(pkt):
    global arp_cache
    if pkt.haslayer(ARP):
        arp = pkt[ARP]
        if arp.op == ARP.who-has:
            if arp.psrc in arp_cache:
                if arp_cache[arp.psrc] != arp.hwsrc:
                    print('ARP spoofing detected from %s (%s) to %s (%s)' % (arp_cache[arp.psrc], arp.psrc, arp.hwsrc, arp.pdst))
                    with open(attack_log_file, 'a') as f:
                        f.write('ARP spoofing detected from %s (%s) to %s (%s)\n' % (arp_cache[arp.psrc], arp.psrc, arp.hwsrc, arp.pdst))
            else:
                arp_cache[arp.psrc] = arp.hwsrc
                if len(arp_cache) > arp_threshold:
                    arp_cache.popitem(last=False)
  1. 检测ICMP flood攻击
def detect_icmp_flood(pkt):
    global icmp_cache
    if pkt.haslayer(ICMP):
        icmp = pkt[ICMP]
        if icmp.type == ICMP.ECHO_REQUEST:
            if icmp.src in icmp_cache:
                icmp_cache[icmp.src] += 1
                if icmp_cache[icmp.src] > icmp_threshold:
                    print('ICMP flood detected from %s (%d packets)' % (icmp.src, icmp_cache[icmp.src]))
                    with open(attack_log_file, 'a') as f:
                        f.write('ICMP flood detected from %s (%d packets)\n' % (icmp.src, icmp_cache[icmp.src]))
            else:
                icmp_cache[icmp.src] = 1
                if len(icmp_cache) > icmp_threshold:
                    icmp_cache.popitem(last=False)
  1. 检测TCP攻击
def detect_tcp(pkt):
    global tcp_cache
    if pkt.haslayer(TCP):
        tcp = pkt[TCP]
        if tcp.flags & (TCP.SYN | TCP.ACK) == TCP.SYN:
            if tcp.dport in (80, 443):
                if tcp.src in tcp_cache:
                    tcp_cache[tcp.src] += 1
                    if tcp_cache[tcp.src] > tcp_threshold:
                        print('TCP attack detected from %s (%d packets)' % (tcp.src, tcp_cache[tcp.src]))
                        with open(attack_log_file, 'a') as f:
                            f.write('TCP attack detected from %s (%d packets)\n' % (tcp.src, tcp_cache[tcp.src]))
                else:
                    tcp_cache[tcp.src] = 1
                    if len(tcp_cache) > tcp_threshold:
                        tcp_cache.popitem(last=False)
  1. 扫描局域网内的主机
def scan_network():
    global active_hosts
    active_hosts = []
    for i in range(1, 255):
        ip_address = '192.168.1.%d' % i
        mac_address = get_mac_address_from_ip(ip_address)
        if mac_address is not None:
            active_hosts.append((ip_address, mac_address))

最后,我们定义一个主函数来启动程序:

def main():
    parser = argparse.ArgumentParser(description='ARP attack detection and protection software')
    parser.add_argument('--interface', '-i', type=str, default='eth0', help='network interface to use (default: eth0)')
    parser.add_argument('--no-arp', '-na', action='store_true', help='disable ARP spoofing detection')
    parser.add_argument('--no-icmp', '-ni', action='store_true', help='disable ICMP flood detection')
    parser.add_argument('--no-tcp', '-nt', action='store_true', help='disable TCP attack detection')
    args = parser.parse_args()

    global detect_arp, detect_icmp, detect_tcp
    detect_arp = not args.no_arp
    detect_icmp = not args.no_icmp
    detect_tcp = not args.no_tcp

    print('Starting ARP attack detection and protection software...')

    # 获取本机IP地址和MAC地址
    ip_address = get_ip_address(args.interface)
    mac_address = get_mac_address(args.interface)

    # 扫描局域网内的主机
    scan_network()

    # 输出本机IP地址和MAC地址
    print('IP address: %s' % ip_address)
    print('MAC address: %s' % mac_address)

    # 输出局域网内的活动主机
    print('Active hosts:')
    for ip_address, mac_address in active_hosts:
        print('  %s (%s)' % (ip_address, mac_address))

    # 开始监听网络流量
    sniff(filter='arp or icmp or tcp', prn=lambda pkt: detect_arp_spoofing(pkt) if detect_arp else None)
    sniff(filter='icmp', prn=lambda pkt: detect_icmp_flood(pkt) if detect_icmp else None)
    sniff(filter='tcp', prn=lambda pkt: detect_tcp(pkt) if detect_tcp else None)

我们可以使用以下命令来启动程序:

python arp_attack_detection.py

如果要禁用某些检测模块,可以使用以下命令:

python arp_attack_detection.py --no-arp --no-icmp --no-tcp

这样就可以根据实际需求来配置程序了

基于Linux的ARP攻击检测防护软件的设计独立的模块包括icmp flood、tcp攻击和ARP欺骗攻击检测若有则输出该攻击的类型若没有则输出并没有遭到该类攻击扫描和记录局域网内活动主机的IP地址和MAC地址能够记录攻击源的IP地址和MAC地址通过中文按钮分别实现web前端将所以记录保存在本地文件上。将模块结合起来通过python基本代码详细实现尽量美观

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

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