detecting-arp-poisoning-in-network-traffic

检测网络流量中的 ARP 欺骗

概述

ARP 欺骗（ARP 伪造）是一种二层攻击，攻击者发送伪造的 ARP 消息，将自己的 MAC 地址与合法主机的 IP 地址关联起来，从而实施中间人（MitM）拦截、会话劫持或拒绝服务攻击。由于 ARP 没有内置的身份验证机制，广播域中的任何设备都可以伪造 ARP 响应。检测需要监控 ARP 流量中的异常情况，例如免费 ARP 洪泛、IP 到 MAC 的映射变化以及重复 IP 地址。本技能涵盖部署多层检测手段，包括 ARPWatch、动态 ARP 检查（DAI）、基于 Wireshark 的分析和自定义 Python 监控工具。

适用场景

• 调查需要检测网络流量中 ARP 欺骗的安全事件时
• 为此领域制定检测规则或威胁搜寻查询时
• SOC 分析师需要针对这类分析制定结构化流程时
• 验证针对相关攻击技术的安全监控覆盖情况时

前置条件

• 可访问目标网络网段（广播域）
• 用于运行 ARPWatch 和自定义监控工具的 Linux 主机
• 支持动态 ARP 检查的托管交换机（Cisco Catalyst、Aruba、Juniper EX）
• 用于数据包捕获的 Wireshark 或 tcpdump
• 已配置 DHCP Snooping（DAI 的前置条件）
• 网络监控基础设施（SIEM、syslog 服务器）

核心概念

ARP 协议基础

ARP 在本地网络网段中将 IP 地址映射到 MAC 地址。该协议以无状态方式运行，且没有身份验证机制：

Normal ARP Process:
1. Host A broadcasts: "Who has 10.0.1.1? Tell 10.0.1.100"
2. Router replies: "10.0.1.1 is at AA:BB:CC:DD:EE:01"
3. Host A caches the mapping

ARP Poisoning Attack:
1. Attacker sends unsolicited ARP reply to Host A:
   "10.0.1.1 is at EV:IL:MA:CA:DD:RR" (attacker's MAC)
2. Host A updates cache, sends traffic to attacker
3. Attacker forwards to real gateway (MitM position)

攻击指标

指标	描述	严重性
MAC 地址来回切换	同一 IP 快速映射到不同的 MAC 地址	高
免费 ARP 洪泛	向多个主机发送未经请求的 ARP 响应	高
重复 IP 地址	两个不同的 MAC 地址声明同一 IP 地址	严重
ARP 流量异常增多	每秒 ARP 数据包数量激增	中
来自非 DHCP 源的 ARP	未知设备声明静态 IP	中
网关 MAC 地址变化	默认网关的 MAC 地址发生变化	严重

工作流

步骤 1：部署 ARPWatch 进行持续监控

# Install ARPWatch
sudo apt-get install -y arpwatch

# Configure ARPWatch
sudo vi /etc/default/arpwatch
# INTERFACES="eth0"
# ARGS="-N -p -i eth0 -f /var/lib/arpwatch/arp.dat"

# Start monitoring
sudo systemctl enable arpwatch
sudo systemctl start arpwatch

# View current ARP database
cat /var/lib/arpwatch/arp.dat

# Monitor logs for changes
tail -f /var/log/syslog | grep arpwatch

ARPWatch 告警类型：

• new station - 此前未见过的 MAC 地址
• changed ethernet address - IP 映射到不同的 MAC 地址（可能存在欺骗）
• flip flop - MAC 地址在两个地址之间交替（正在进行的攻击）
• reused old ethernet address - 重新出现此前记录过的映射

步骤 2：在交换机上配置动态 ARP 检查（DAI）

Cisco Catalyst 配置：

! Enable DHCP snooping (prerequisite for DAI)
ip dhcp snooping
ip dhcp snooping vlan 10,20,30

! Configure trusted ports (uplinks, DHCP servers)
interface GigabitEthernet1/0/1
 description Uplink to Distribution
 ip dhcp snooping trust

interface GigabitEthernet1/0/48
 description DHCP Server
 ip dhcp snooping trust

! Enable Dynamic ARP Inspection
ip arp inspection vlan 10,20,30

! Configure trusted ports for DAI
interface GigabitEthernet1/0/1
 ip arp inspection trust

! Set rate limits to prevent ARP flood DoS
interface range GigabitEthernet1/0/2-47
 ip arp inspection limit rate 15

! Enable additional validation checks
ip arp inspection validate src-mac dst-mac ip

! Configure ARP ACL for static IP devices (servers, printers)
arp access-list STATIC-ARP-ENTRIES
 permit ip host 10.0.10.100 mac host 0011.2233.4455
 permit ip host 10.0.10.101 mac host 0011.2233.4456

ip arp inspection filter STATIC-ARP-ENTRIES vlan 10

! Verify DAI status
show ip arp inspection vlan 10
show ip arp inspection statistics
show ip dhcp snooping binding

步骤 3：Wireshark 检测过滤器

# Detect gratuitous ARP (sender and target IP are the same)
arp.src.proto_ipv4 == arp.dst.proto_ipv4

# Detect ARP replies (focus on unsolicited)
arp.opcode == 2

# Detect duplicate IP address claims
arp.duplicate-address-detected

# Detect ARP packets from specific attacker MAC
eth.src == ev:il:ma:ca:dd:rr

# Detect ARP storms (high volume)
# Use Statistics > I/O Graphs > Display filter: arp

# Detect gateway impersonation
arp.src.proto_ipv4 == 10.0.1.1 && arp.src.hw_mac != aa:bb:cc:dd:ee:01

步骤 4：自定义 Python ARP 监控器

#!/usr/bin/env python3
"""
Real-time ARP poisoning detection using packet capture.
Monitors ARP traffic for spoofing indicators and alerts on anomalies.
"""

import subprocess
import sys
import json
import time
from collections import defaultdict
from datetime import datetime

try:
    from scapy.all import sniff, ARP, Ether, get_if_hwaddr, conf
    SCAPY_AVAILABLE = True
except ImportError:
    SCAPY_AVAILABLE = False


class ARPPoisonDetector:
    def __init__(self, interface: str, gateway_ip: str, gateway_mac: str):
        self.interface = interface
        self.gateway_ip = gateway_ip
        self.gateway_mac = gateway_mac.lower()
        self.arp_table = {}  # IP -> MAC mapping
        self.arp_history = defaultdict(list)  # IP -> list of (MAC, timestamp)
        self.alerts = []
        self.arp_count = defaultdict(int)  # Source MAC -> count per interval
        self.last_reset = time.time()
        self.arp_rate_threshold = 50  # ARP packets per 10 seconds

    def alert(self, severity: str, message: str, details: dict):
        """Generate alert for detected anomaly."""
        alert_data = {
            'timestamp': datetime.now().isoformat(),
            'severity': severity,
            'message': message,
            'details': details,
        }
        self.alerts.append(alert_data)
        print(f"\n[{severity}] {datetime.now().strftime('%H:%M:%S')} - {message}")
        for key, value in details.items():
            print(f"  {key}: {value}")

    def check_gateway_spoofing(self, src_ip: str, src_mac: str):
        """Check if someone is spoofing the gateway."""
        if src_ip == self.gateway_ip and src_mac != self.gateway_mac:
            self.alert('CRITICAL', 'Gateway ARP Spoofing Detected', {
                'gateway_ip': self.gateway_ip,
                'expected_mac': self.gateway_mac,
                'spoofed_mac': src_mac,
                'action': 'Potential MitM attack on default gateway',
            })
            return True
        return False

    def check_mac_change(self, src_ip: str, src_mac: str):
        """Check if IP-to-MAC mapping has changed."""
        if src_ip in self.arp_table:
            known_mac = self.arp_table[src_ip]
            if known_mac != src_mac:
                self.alert('HIGH', 'ARP Cache Poisoning Attempt', {
                    'ip_address': src_ip,
                    'previous_mac': known_mac,
                    'new_mac': src_mac,
                    'action': 'IP-to-MAC mapping changed unexpectedly',
                })
                return True
        return False

    def check_flip_flop(self, src_ip: str, src_mac: str):
        """Check for MAC address flip-flopping (active attack indicator)."""
        self.arp_history[src_ip].append((src_mac, time.time()))

        # Keep only last 60 seconds of history
        cutoff = time.time() - 60
        self.arp_history[src_ip] = [
            (mac, ts) for mac, ts in self.arp_history[src_ip]
            if ts > cutoff
        ]

        unique_macs = set(mac for mac, ts in self.arp_history[src_ip])
        if len(unique_macs) > 2:
            self.alert('CRITICAL', 'ARP Flip-Flop Detected (Active Attack)', {
                'ip_address': src_ip,
                'mac_addresses': list(unique_macs),
                'changes_in_60s': len(self.arp_history[src_ip]),
            })
            return True
        return False

    def check_arp_rate(self, src_mac: str):
        """Check for ARP flood (DoS or scanning)."""
        self.arp_count[src_mac] += 1

        # Reset counters every 10 seconds
        if time.time() - self.last_reset > 10:
            for mac, count in self.arp_count.items():
                if count > self.arp_rate_threshold:
                    self.alert('MEDIUM', 'ARP Flood Detected', {
                        'source_mac': mac,
                        'arp_packets_10s': count,
                        'threshold': self.arp_rate_threshold,
                    })
            self.arp_count.clear()
            self.last_reset = time.time()

    def process_packet(self, packet):
        """Process captured ARP packet."""
        if not packet.haslayer(ARP):
            return

        arp = packet[ARP]

        # Only process ARP replies (opcode 2) and requests (opcode 1)
        if arp.op not in (1, 2):
            return

        src_ip = arp.psrc
        src_mac = arp.hwsrc.lower()

        # Run detection checks
        self.check_gateway_spoofing(src_ip, src_mac)
        self.check_mac_change(src_ip, src_mac)
        self.check_flip_flop(src_ip, src_mac)
        self.check_arp_rate(src_mac)

        # Update ARP table
        self.arp_table[src_ip] = src_mac

    def start_monitoring(self):
        """Start real-time ARP monitoring."""
        print(f"[*] Starting ARP Poison Detection on {self.interface}")
        print(f"[*] Gateway: {self.gateway_ip} ({self.gateway_mac})")
        print(f"[*] Monitoring... (Ctrl+C to stop)\n")

        if SCAPY_AVAILABLE:
            sniff(
                iface=self.interface,
                filter="arp",
                prn=self.process_packet,
                store=False,
            )
        else:
            print("[-] Scapy not available. Install with: pip install scapy")
            print("[*] Falling back to tcpdump-based monitoring...")
            self._monitor_with_tcpdump()

    def _monitor_with_tcpdump(self):
        """Fallback monitoring using tcpdump."""
        cmd = ['tcpdump', '-i', self.interface, '-l', '-n', 'arp']
        proc = subprocess.Popen(cmd, stdout=subprocess.PIPE,
                                stderr=subprocess.DEVNULL, text=True)
        try:
            for line in proc.stdout:
                parts = line.strip().split()
                if 'is-at' in parts:
                    try:
                        ip_idx = parts.index('is-at') - 1
                        mac_idx = parts.index('is-at') + 1
                        src_ip = parts[ip_idx]
                        src_mac = parts[mac_idx].lower()
                        self.check_gateway_spoofing(src_ip, src_mac)
                        self.check_mac_change(src_ip, src_mac)
                        self.arp_table[src_ip] = src_mac
                    except (IndexError, ValueError):
                        continue
        except KeyboardInterrupt:
            proc.terminate()

    def generate_report(self) -> dict:
        """Generate summary report of detected anomalies."""
        return {
            'monitoring_interface': self.interface,
            'gateway': {'ip': self.gateway_ip, 'mac': self.gateway_mac},
            'total_alerts': len(self.alerts),
            'arp_table_size': len(self.arp_table),
            'alerts': self.alerts,
        }


if __name__ == '__main__':
    if len(sys.argv) < 4:
        print("Usage: python process.py <interface> <gateway_ip> <gateway_mac>")
        print("Example: python process.py eth0 10.0.1.1 aa:bb:cc:dd:ee:01")
        sys.exit(1)

    detector = ARPPoisonDetector(
        interface=sys.argv[1],
        gateway_ip=sys.argv[2],
        gateway_mac=sys.argv[3],
    )

    try:
        detector.start_monitoring()
    except KeyboardInterrupt:
        print("\n\n[*] Monitoring stopped.")
        report = detector.generate_report()
        print(f"[*] Total alerts generated: {report['total_alerts']}")
        print(f"[*] ARP table entries: {report['arp_table_size']}")

预防措施

二层控制措施

1. 动态 ARP 检查（DAI） - 根据 DHCP Snooping 绑定表验证 ARP 数据包
2. DHCP Snooping - 建立可信的 IP-MAC-端口绑定数据库
3. 端口安全 - 限制每个端口的 MAC 地址数量
4. 私有 VLAN - 限制同一 VLAN 内主机之间的通信

网络控制措施

1. 静态 ARP 条目 - 用于关键基础设施（网关、DNS、DHCP）
2. 网络分段 - 使用 VLAN 缩小广播域
3. 802.1X 身份验证 - 在设备访问网络前对其进行身份验证
4. 加密协议 - 使用 SSH、HTTPS、TLS 保护数据，即使数据被拦截也不受影响

最佳实践

• 纵深防御 - 结合 DAI、ARPWatch 和自定义监控，实现全面防护
• 优先启用 DHCP Snooping - 始终先启用 DHCP snooping，再启用 DAI（DAI 依赖 snooping 数据库）
• 为网关配置静态 ARP - 在关键服务器上为默认网关配置静态 ARP 条目
• 监控免费 ARP - 特别留意未经请求的 ARP 响应
• 使用小型广播域 - 使用 VLAN 限制基于 ARP 的攻击范围
• 定期审计 - 定期比较各设备上的 ARP 表，以识别异常

参考资料

• NIST SP 800-54 - 边界网关协议安全
• Cisco DAI 配置指南
• Comparitech ARP 欺骗指南
• Okta ARP 欺骗技术