182 lines
9.4 KiB
Python
182 lines
9.4 KiB
Python
"""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)
|