界面已本地化;技能介绍与说明文档按所选语言由 AI 自动翻译,代码、命令、路径和专业术语保留原文,以原作者文档为准。
探索技能
效率与协作 · SKILL.md

detecting-arp-poisoning-in-network-traffic

detecting-arp-poisoning-in-network-traffic

【效率与协作】使用 ARPWatch、动态 ARP 检测(Dynamic ARP Inspection)、Wireshark 分析和自定义监控脚本检测和防止 ARP 欺骗(ARP spoofing)攻击,防御中间人(man-in-the-middle)拦截。

独立 GitHub 快照 · 自动分类 · 以原作者文档为准

技能说明

GitHub

name: detecting-arp-poisoning-in-network-traffic description: 使用 ARPWatch、动态 ARP 检测(Dynamic ARP Inspection)、Wireshark 分析和自定义监控脚本检测和防止 ARP 欺骗(ARP spoofing)攻击,防御中间人(man-in-the-middle)拦截。 domain: cybersecurity subdomain: network-security tags: [arp-poisoning, arp-spoofing, mitm, dynamic-arp-inspection, arpwatch, network-security, man-in-the-middle, layer-2-security] version: "1.0" author: mahipal license: Apache-2.0

检测网络流量中的 ARP 投毒

概述

ARP 投毒(ARP poisoning,又称 ARP 欺骗)是一种二层(Layer 2)攻击,攻击者发送伪造的 ARP 消息,将其 MAC 地址与合法主机的 IP 地址关联,从而实现中间人(MITM)拦截、会话劫持或拒绝服务攻击。由于 ARP 没有内置身份验证机制,广播域上的任何设备都可以伪造 ARP 应答。检测需要监控 ARP 流量中的异常,如免费 ARP(gratuitous ARP)洪水、IP 到 MAC 映射变化和重复 IP 地址。本技能涵盖部署多层检测,包括 ARPWatch、动态 ARP 检测(DAI)、基于 Wireshark 的分析和自定义 Python 监控工具。

前置条件

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

核心概念

ARP 协议基础

ARP 在本地网段将 IP 地址映射到 MAC 地址。该协议以无状态方式运行,没有身份验证:

正常 ARP 过程:
1. 主机 A 广播:"谁有 10.0.1.1?请告诉 10.0.1.100"
2. 路由器回复:"10.0.1.1 在 AA:BB:CC:DD:EE:01"
3. 主机 A 缓存该映射

ARP 投毒攻击:
1. 攻击者向主机 A 发送未请求的 ARP 应答:
   "10.0.1.1 在 EV:IL:MA:CA:DD:RR"(攻击者的 MAC)
2. 主机 A 更新缓存,将流量发送给攻击者
3. 攻击者转发给真实网关(中间人位置)

攻击指标

指标描述严重性
MAC 地址反复跳变同一 IP 快速映射到不同 MAC高
免费 ARP 洪水针对多台主机的未请求 ARP 应答高
重复 IP 地址两个不同 MAC 声称相同 IP严重
异常 ARP 数量ARP 每秒数据包数峰值中
来自非 DHCP 源的 ARP来自未知设备的静态 IP 声明中
网关 MAC 变更默认网关 MAC 地址更改严重

实施步骤

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

# 安装 ARPWatch
sudo apt-get install -y arpwatch

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

# 启动监控
sudo systemctl enable arpwatch
sudo systemctl start arpwatch

# 查看当前 ARP 数据库
cat /var/lib/arpwatch/arp.dat

# 监控日志中的变化
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 配置:

! 启用 DHCP 嗅探(DAI 的前置条件)
ip dhcp snooping
ip dhcp snooping vlan 10,20,30

! 配置可信端口(上行链路、DHCP 服务器)
interface GigabitEthernet1/0/1
 description Uplink to Distribution
 ip dhcp snooping trust

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

! 启用动态 ARP 检测
ip arp inspection vlan 10,20,30

! 配置 DAI 的可信端口
interface GigabitEthernet1/0/1
 ip arp inspection trust

! 设置速率限制以防止 ARP 洪水 DoS
interface range GigabitEthernet1/0/2-47
 ip arp inspection limit rate 15

! 启用额外的验证检查
ip arp inspection validate src-mac dst-mac ip

! 为静态 IP 设备配置 ARP ACL(服务器、打印机)
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

! 验证 DAI 状态
show ip arp inspection vlan 10
show ip arp inspection statistics
show ip dhcp snooping binding

步骤 3:Wireshark 检测过滤器

# 检测免费 ARP(发送方和目标 IP 相同)
arp.src.proto_ipv4 == arp.dst.proto_ipv4

# 检测 ARP 应答(关注未请求的)
arp.opcode == 2

# 检测重复 IP 地址声明
arp.duplicate-address-detected

# 检测来自特定攻击者 MAC 的 ARP 数据包
eth.src == ev:il:ma:ca:dd:rr

# 检测 ARP 风暴(高流量)
# 使用 Statistics > I/O Graphs > 显示过滤器:arp

# 检测网关冒充
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
"""
使用数据包捕获进行实时 ARP 投毒检测。
监控 ARP 流量中的欺骗指标并在异常时发出告警。
"""

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 映射
        self.arp_history = defaultdict(list)  # IP -> (MAC, 时间戳) 列表
        self.alerts = []
        self.arp_count = defaultdict(int)  # 源 MAC -> 每间隔计数
        self.last_reset = time.time()
        self.arp_rate_threshold = 50  # 每 10 秒的 ARP 数据包数

    def alert(self, severity: str, message: str, details: dict):
        """为检测到的异常生成告警。"""
        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):
        """检查是否有人在伪造网关。"""
        if src_ip == self.gateway_ip and src_mac != self.gateway_mac:
            self.alert('CRITICAL', '检测到网关 ARP 欺骗', {
                'gateway_ip': self.gateway_ip,
                'expected_mac': self.gateway_mac,
                'spoofed_mac': src_mac,
                'action': '默认网关上的潜在中间人攻击',
            })
            return True
        return False

    def check_mac_change(self, src_ip: str, src_mac: str):
        """检查 IP 到 MAC 的映射是否已更改。"""
        if src_ip in self.arp_table:
            known_mac = self.arp_table[src_ip]
            if known_mac != src_mac:
                self.alert('HIGH', 'ARP 缓存投毒尝试', {
                    'ip_address': src_ip,
                    'previous_mac': known_mac,
                    'new_mac': src_mac,
                    'action': 'IP 到 MAC 映射意外更改',
                })
                return True
        return False

    def check_flip_flop(self, src_ip: str, src_mac: str):
        """检查 MAC 地址反复跳变(主动攻击指标)。"""
        self.arp_history[src_ip].append((src_mac, time.time()))

        # 仅保留最近 60 秒的历史
        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 地址跳变(主动攻击)', {
                '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):
        """检查 ARP 洪水(DoS 或扫描)。"""
        self.arp_count[src_mac] += 1

        # 每 10 秒重置计数器
        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 洪水', {
                        '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):
        """处理捕获的 ARP 数据包。"""
        if not packet.haslayer(ARP):
            return

        arp = packet[ARP]

        # 仅处理 ARP 应答(操作码 2)和请求(操作码 1)
        if arp.op not in (1, 2):
            return

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

        # 运行检测检查
        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)

        # 更新 ARP 表
        self.arp_table[src_ip] = src_mac

    def start_monitoring(self):
        """启动实时 ARP 监控。"""
        print(f"[*] 在 {self.interface} 上启动 ARP 投毒检测")
        print(f"[*] 网关:{self.gateway_ip}({self.gateway_mac})")
        print(f"[*] 监控中...(Ctrl+C 停止)\n")

        if SCAPY_AVAILABLE:
            sniff(
                iface=self.interface,
                filter="arp",
                prn=self.process_packet,
                store=False,
            )
        else:
            print("[-] Scapy 不可用。安装:pip install scapy")
            print("[*] 回退到基于 tcpdump 的监控...")
            self._monitor_with_tcpdump()

    def _monitor_with_tcpdump(self):
        """使用 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:
        """生成检测到的异常摘要报告。"""
        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("用法:python process.py <interface> <gateway_ip> <gateway_mac>")
        print("示例: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[*] 监控已停止。")
        report = detector.generate_report()
        print(f"[*] 生成的告警总数:{report['total_alerts']}")
        print(f"[*] ARP 表条目:{report['arp_table_size']}")

预防措施

二层控制

  1. 动态 ARP 检测(DAI) - 根据 DHCP 嗅探绑定表验证 ARP 数据包
  2. DHCP 嗅探 - 构建可信的 IP-MAC-端口绑定数据库
  3. 端口安全 - 限制每个端口的 MAC 地址数
  4. 私有 VLAN - 限制同一 VLAN 中主机之间的通信

网络控制

  1. 静态 ARP 条目 - 用于关键基础设施(网关、DNS、DHCP)
  2. 网络分段 - 使用 VLAN 减小广播域大小
  3. 802.1X 认证 - 在网络访问前对设备进行身份验证
  4. 加密协议 - 使用 SSH、HTTPS、TLS 保护数据,即使被拦截也是安全的

最佳实践

  • 纵深防御 - 结合 DAI、ARPWatch 和自定义监控实现全面覆盖
  • 先配置 DHCP 嗅探 - 在 DAI 之前始终启用 DHCP 嗅探(DAI 依赖嗅探数据库)
  • 为网关配置静态 ARP - 在关键服务器上为默认网关配置静态 ARP 条目
  • 监控免费 ARP - 特别关注未请求的 ARP 应答
  • 小广播域 - 使用 VLAN 限制 ARP 攻击的范围
  • 定期审计 - 定期比较各设备上的 ARP 表以识别异常

参考资料

  • NIST SP 800-54 - 边界网关协议安全
  • Cisco DAI 配置指南
  • MITRE ATT&CK T1557.002 — ARP 缓存投毒
我的收藏