"""Detect local network manipulation and man-in-the-middle indicators.""" from __future__ import annotations from collections import defaultdict from typing import Iterable from ..context import CaptureContext, is_special from ..findings import Finding, Category, Severity, Confidence from .base import BaseDetector, register _ARP_REPLY_RATE_PER_MIN = 60 _MIN_ARP_REPLIES = 100 @register class MitmDetector(BaseDetector): name = "MitmDetector" description = "Detects ARP conflicts, ARP floods, rogue DHCP servers and ICMP redirects." def detect(self, ctx: CaptureContext) -> Iterable[Finding]: findings: list[Finding] = [] # Map each IP address to MAC addresses that claimed it, in time order. ip_to_macs: dict[str, dict[str, float]] = defaultdict(dict) packets_for_ip: dict[str, list[int]] = defaultdict(list) def claim(ip, mac, ts, pnum): if not ip or not mac or is_special(ip): return mac = mac.lower() if mac in ("ff:ff:ff:ff:ff:ff", "00:00:00:00:00:00") or mac.startswith(("01:00:5e", "33:33")): return first = ip_to_macs[ip].get(mac) if first is None or (ts is not None and ts < first): ip_to_macs[ip][mac] = ts if ts is not None else float("inf") if pnum: packets_for_ip[ip].append(pnum) for ev in ctx.arp_events: if ev.opcode in (1, 2): claim(ev.sender_ip, ev.sender_mac, ev.timestamp, ev.packet_number) for ip, host in ctx.hosts.items(): if host.is_private and not is_special(ip): for mac in host.macs: claim(ip, mac, host.first_seen, None) for ip, macs in ip_to_macs.items(): if len(macs) > 1: ordered = sorted(macs.items(), key=lambda kv: kv[1]) times = [t for _, t in ordered if t != float("inf")] findings.append(Finding( title="Multiple MAC addresses associated with one IP", category=Category.MITM, severity=Severity.HIGH, confidence=Confidence.MEDIUM, source_ip=ip, protocol="arp", timestamp_first_seen=times[0] if times else None, timestamp_last_seen=times[-1] if times else None, risk_description=( "One IP address was claimed by multiple MAC addresses. This happens with " "ARP spoofing and traffic interception, but also with a DHCP lease that " "moved between devices or a router fail-over." ), recommendation=( "Identify both hosts by MAC address, review ARP tables, and enable " "Dynamic ARP Inspection or port security." ), evidence={"ip": ip, "macs": [m for m, _ in ordered], "device": ctx.is_device(ip), "dedup_key": f"arpdup:{ip}"}, related_packets=sorted(packets_for_ip.get(ip, []))[:10], )) # ARP reply floods: judge by rate, not by the absolute count in a long capture. replies = [e for e in ctx.arp_events if e.opcode == 2] requests = [e for e in ctx.arp_events if e.opcode == 1] minutes = max(ctx.duration / 60.0, 1.0) if len(replies) >= _MIN_ARP_REPLIES and len(replies) / minutes >= _ARP_REPLY_RATE_PER_MIN \ and len(replies) > 2 * max(len(requests), 1): by_sender: dict[str, int] = defaultdict(int) for e in replies: by_sender[e.sender_mac or "?"] += 1 top = sorted(by_sender.items(), key=lambda kv: -kv[1])[:3] findings.append(Finding( title="Unusually high ARP reply rate", category=Category.MITM, severity=Severity.MEDIUM, confidence=Confidence.MEDIUM, protocol="arp", timestamp_first_seen=replies[0].timestamp, timestamp_last_seen=replies[-1].timestamp, risk_description=( "Far more ARP replies than requests were seen at a high rate, which is " "characteristic of active ARP cache poisoning." ), recommendation="Inspect the reply source and investigate possible interception.", evidence={"arp_replies": len(replies), "arp_requests": len(requests), "replies_per_minute": round(len(replies) / minutes, 1), "top_senders": [f"{m} ({n})" for m, n in top], "dedup_key": "arp_reply_flood"}, related_packets=sorted({e.packet_number for e in replies if e.packet_number})[:8], )) # Gratuitous ARP claiming an address that already belongs to another MAC. seen_owner: dict[str, str] = {} for e in sorted(ctx.arp_events, key=lambda e: e.timestamp or 0): if not e.sender_ip or not e.sender_mac or is_special(e.sender_ip): continue owner = seen_owner.setdefault(e.sender_ip, e.sender_mac) if e.is_gratuitous and owner != e.sender_mac: findings.append(Finding( title="Gratuitous ARP overriding an existing address owner", category=Category.MITM, severity=Severity.HIGH, confidence=Confidence.MEDIUM, source_ip=e.sender_ip, protocol="arp", timestamp_first_seen=e.timestamp, timestamp_last_seen=e.timestamp, risk_description=( "A host announced an IP address that another MAC address had been using, " "which redirects traffic for that address to the announcing host." ), recommendation="Identify the announcing host and check for interception.", evidence={"ip": e.sender_ip, "previous_mac": owner, "new_mac": e.sender_mac, "dedup_key": f"garp:{e.sender_ip}:{e.sender_mac}"}, related_packets=[p for p in [e.packet_number] if p], )) # More than one DHCP server answering on the segment. servers: dict[str, list] = defaultdict(list) for d in ctx.dhcp_events: if d.message_type in ("offer", "ack") and (d.server_id or d.src_ip): servers[d.server_id or d.src_ip].append(d) if len(servers) > 1: findings.append(Finding( title="Multiple DHCP servers observed", category=Category.MITM, severity=Severity.MEDIUM, confidence=Confidence.MEDIUM, protocol="dhcp", port=67, risk_description=( "Offers or acknowledgements came from more than one DHCP server. A rogue " "DHCP server can hand out a malicious gateway or DNS resolver." ), recommendation="Verify every DHCP server on the segment and enable DHCP snooping.", evidence={"servers": sorted(servers), "dedup_key": "dhcp_multi_server"}, related_packets=sorted({d.packet_number for evs in servers.values() for d in evs if d.packet_number})[:8], )) # ICMP redirects and IPv6 router advertisements from unexpected sources. redirects = [s for s in ctx.service_events if s.kind == "redirect"] if redirects: srcs = sorted({s.src_ip for s in redirects if s.src_ip}) findings.append(Finding( title="ICMP redirect messages observed", category=Category.MITM, severity=Severity.MEDIUM, confidence=Confidence.MEDIUM, source_ip=srcs[0] if srcs else None, protocol="icmp", timestamp_first_seen=redirects[0].timestamp, timestamp_last_seen=redirects[-1].timestamp, risk_description=( "ICMP redirects change a host's routing table and are a known technique to " "insert a device into the traffic path." ), recommendation="Disable acceptance of ICMP redirects on hosts and verify the sender.", evidence={"count": len(redirects), "sources": srcs[:5], "dedup_key": "icmp_redirect"}, related_packets=sorted({s.packet_number for s in redirects if s.packet_number})[:8], )) ras = [s for s in ctx.service_events if s.kind == "router_advertisement"] ra_sources = sorted({s.src_ip for s in ras if s.src_ip}) if len(ra_sources) > 1: findings.append(Finding( title="IPv6 router advertisements from multiple sources", category=Category.MITM, severity=Severity.MEDIUM, confidence=Confidence.LOW, protocol="icmpv6", risk_description=( "More than one host advertises itself as an IPv6 router. A rogue " "advertisement can redirect all IPv6 traffic through an attacker." ), recommendation="Verify the legitimate router and enable RA Guard on switches.", evidence={"sources": ra_sources[:5], "dedup_key": "rogue_ra"}, related_packets=sorted({s.packet_number for s in ras if s.packet_number})[:8], )) return self._tag(findings)