基于Linux的ARP攻击检测防护软件设计与实现

本软件基于Python开发，能够检测和防护ICMP Flood、TCP攻击和ARP欺骗攻击。软件提供用户登录注册功能，并通过良好的交互界面展示攻击检测结果和日志记录。

软件功能

攻击检测：
- ICMP Flood攻击检测
- TCP攻击检测
- ARP欺骗攻击检测
攻击防护：
- 针对检测到的攻击，采取相应的防护措施
网络扫描：
- 扫描局域网内活动主机的IP地址和MAC地址
日志记录：
- 记录攻击源的IP地址和MAC地址
- 记录攻击类型和时间戳
用户管理：
- 用户登录和注册
交互界面：
- 提供友好的用户界面，方便用户使用

软件架构

软件架构采用模块化设计，包含以下几个模块：

**攻击检测模块：**负责检测各种攻击类型，包括ICMP Flood、TCP攻击和ARP欺骗攻击。
**攻击防护模块：**负责根据检测到的攻击类型采取相应的防护措施。
**网络扫描模块：**负责扫描局域网内活动主机。
**日志记录模块：**负责将攻击日志存储到数据库中。
**用户管理模块：**负责用户登录注册。
**交互界面模块：**负责提供用户界面，方便用户使用。

代码实现

登录界面(login.html)

<!DOCTYPE html>
<html>
<head>
	<title>Login</title>
</head>
<body>
	<h1>Login</h1>
	<form action='login.py' method='post'>
		<label>Username:</label>
		<input type='text' name='username'><br><br>
		<label>Password:</label>
		<input type='password' name='password'><br><br>
		<input type='submit' value='Login'>
	</form>
	<p>New user? <a href='register.html'>Register here</a></p>
</body>
</html>

注册界面(register.html)

<!DOCTYPE html>
<html>
<head>
	<title>Register</title>
</head>
<body>
	<h1>Register</h1>
	<form action='register.py' method='post'>
		<label>Username:</label>
		<input type='text' name='username'><br><br>
		<label>Password:</label>
		<input type='password' name='password'><br><br>
		<label>Confirm Password:</label>
		<input type='password' name='confirm_password'><br><br>
		<input type='submit' value='Register'>
	</form>
	<p>Already registered? <a href='login.html'>Login here</a></p>
</body>
</html>

主程序代码

import os
import sys
import time
import socket
import struct
import argparse
import threading
import sqlite3 as sql

# Define constants
ETHERNET_HEADER_LENGTH = 14
ARP_HEADER_LENGTH = 28

# Define global variables
arpcache = {}
arplock = threading.Lock()

# Define functions
def get_mac_address(ip_address):
    """
    Get the MAC address of a given IP address using ARP.
    """
    global arpcache
    global arplock

    # Check if the MAC address is already in the ARP cache
    if ip_address in arpcache:
        return arpcache[ip_address]

    # Send an ARP request to get the MAC address
    arp_request = create_arp_request(ip_address)
    with socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.SOCK_RAW) as s:
        s.bind((args.interface, socket.SOCK_RAW))
        s.send(arp_request)
        while True:
            packet = s.recvfrom(2048)[0]
            ethernet_header = packet[:ETHERNET_HEADER_LENGTH]
            arp_header = packet[ETHERNET_HEADER_LENGTH:ETHERNET_HEADER_LENGTH+ARP_HEADER_LENGTH]
            if is_arp_reply(arp_header) and get_sender_ip_address(arp_header) == ip_address:
                mac_address = get_sender_mac_address(arp_header)
                arplock.acquire()
                arpcache[ip_address] = mac_address
                arplock.release()
                return mac_address

def create_arp_request(ip_address):
    """
    Create an ARP request packet.
    """
    ethernet_header = create_ethernet_header(args.interface, 'ff:ff:ff:ff:ff:ff')
    arp_header = struct.pack('!HHBBH6s4s6s4s', 0x0001, 0x0800, 0x06, 0x04, 0x0001, get_mac_address(args.interface), socket.inet_aton('0.0.0.0'), '00:00:00:00:00:00', socket.inet_aton(ip_address))
    return ethernet_header + arp_header

def create_ethernet_header(interface, destination_mac_address):
    """
    Create an Ethernet header.
    """
    source_mac_address = get_mac_address(interface)
    ethernet_header = struct.pack('!6s6sH', destination_mac_address, source_mac_address, 0x0806)
    return ethernet_header

def is_arp_reply(arp_header):
    """
    Check if an ARP packet is a reply.
    """
    return struct.unpack('!H', arp_header[6:8])[0] == 0x0002

def get_sender_mac_address(arp_header):
    """
    Get the MAC address of the sender of an ARP packet.
    """
    return arp_header[8:14]

def get_sender_ip_address(arp_header):
    """
    Get the IP address of the sender of an ARP packet.
    """
    return socket.inet_ntoa(arp_header[14:18])

def icmp_flood_detection():
    """
    Detect ICMP flood attacks.
    """
    print('Checking for ICMP flood attacks...')
    # TODO: Implement ICMP flood detection

def tcp_attack_detection():
    """
    Detect TCP attacks.
    """
    print('Checking for TCP attacks...')
    # TODO: Implement TCP attack detection

def arp_spoof_detection():
    """
    Detect ARP spoofing attacks.
    """
    print('Checking for ARP spoofing attacks...')
    # TODO: Implement ARP spoof detection

def scan_network():
    """
    Scan the network for active hosts.
    """
    print('Scanning the network for active hosts...')
    # TODO: Implement network scanning

def log_attack(source_ip_address, source_mac_address, attack_type):
    """
    Log an attack in the database.
    """
    with sql.connect('attacks.db') as conn:
        cursor = conn.cursor()
        cursor.execute('INSERT INTO attacks (source_ip_address, source_mac_address, attack_type, timestamp) VALUES (?, ?, ?, ?)', (source_ip_address, source_mac_address, attack_type, int(time.time())))

def main():
    # Parse command line arguments
    parser = argparse.ArgumentParser()
    parser.add_argument('-i', '--interface', help='the network interface to use', required=True)
    args = parser.parse_args()

    # Create the database if it doesn't exist
    with sql.connect('attacks.db') as conn:
        cursor = conn.cursor()
        cursor.execute('CREATE TABLE IF NOT EXISTS attacks (id INTEGER PRIMARY KEY AUTOINCREMENT, source_ip_address TEXT, source_mac_address TEXT, attack_type TEXT, timestamp INTEGER)')

    # Start the detection and prevention threads
    icmp_flood_thread = threading.Thread(target=icmp_flood_detection)
    icmp_flood_thread.start()
    tcp_attack_thread = threading.Thread(target=tcp_attack_detection)
    tcp_attack_thread.start()
    arp_spoof_thread = threading.Thread(target=arp_spoof_detection)
    arp_spoof_thread.start()

    # Start the scanning thread
    scan_thread = threading.Thread(target=scan_network)
    scan_thread.start()

    # Start the logging thread
    while True:
        try:
            packet = s.recvfrom(2048)[0]
            ethernet_header = packet[:ETHERNET_HEADER_LENGTH]
            arp_header = packet[ETHERNET_HEADER_LENGTH:ETHERNET_HEADER_LENGTH+ARP_HEADER_LENGTH]
            if is_arp_reply(arp_header):
                source_mac_address = get_sender_mac_address(arp_header)
                source_ip_address = get_sender_ip_address(arp_header)
                log_attack(source_ip_address, source_mac_address, 'ARP spoofing')
        except KeyboardInterrupt:
            sys.exit()

if __name__ == '__main__':
    main()

登录处理代码(login.py)

import cgi
import sqlite3 as sql

# Get the form data
form = cgi.FieldStorage()
username = form.getvalue('username')
password = form.getvalue('password')

# Check if the username and password are correct
with sql.connect('users.db') as conn:
    cursor = conn.cursor()
    cursor.execute('SELECT * FROM users WHERE username = ? AND password = ?', (username, password))
    if cursor.fetchone() is not None:
        print('Content-Type: text/html')
        print()
        print('<html>')
        print('<head>')
        print('<title>ARP Attack Detection and Prevention</title>')
        print('</head>')
        print('<body>')
        print('<h1>Welcome, {}!</h1>'.format(username))
        print('<p><a href="main.py">Go to main page</a></p>')
        print('</body>')
        print('</html>')
    else:
        print('Content-Type: text/html')
        print()
        print('<html>')
        print('<head>')
        print('<title>ARP Attack Detection and Prevention</title>')
        print('</head>')
        print('<body>')
        print('<h1>Invalid username or password</h1>')
        print('<p><a href="login.html">Go back to login page</a></p>')
        print('</body>')
        print('</html>')

注册处理代码(register.py)

import cgi
import sqlite3 as sql

# Get the form data
form = cgi.FieldStorage()
username = form.getvalue('username')
password = form.getvalue('password')
confirm_password = form.getvalue('confirm_password')

# Check if the passwords match
if password != confirm_password:
    print('Content-Type: text/html')
    print()
    print('<html>')
    print('<head>')
    print('<title>ARP Attack Detection and Prevention</title>')
    print('</head>')
    print('<body>')
    print('<h1>Passwords do not match</h1>')
    print('<p><a href="register.html">Go back to registration page</a></p>')
    print('</body>')
    print('</html>')
else:
    # Add the user to the database
    with sql.connect('users.db') as conn:
        cursor = conn.cursor()
        cursor.execute('INSERT INTO users (username, password) VALUES (?, ?)', (username, password))
    print('Content-Type: text/html')
    print()
    print('<html>')
    print('<head>')
    print('<title>ARP Attack Detection and Prevention</title>')
    print('</head>')
    print('<body>')
    print('<h1>Registration successful</h1>')
    print('<p><a href="login.html">Go to login page</a></p>')
    print('</body>')
    print('</html>')

注意： 以上代码仅为示例，需要根据实际情况进行修改和完善。