n0mad1k/bigbrother
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add active ARP scanning to bypass AP proxy on WiFi networks

Browse Source 
Passive ARP sniffing fails on modern WiFi because the AP intercepts
ARP broadcasts and responds on behalf of clients. Active scanning
sends who-has to all 254 /24 addresses and collects replies directly,
making presence detection reliable regardless of AP configuration.

Also adds monitor mode sniffer thread that activates automatically if
a second WiFi interface is present (e.g. USB dongle). Reads raw 802.11
frames, bypassing the AP stack entirely for station-transmitted data
and management frames.
This commit is contained in:
Cobra
2026-04-15 12:58:59 -04:00
parent a9d43ce4d3
commit 9b2f747775
1 changed files with 201 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
net_alerter/net_alerter.py
+201 -1
View File
@@ -32,6 +32,7 @@ OUI_DB_PATH = "/opt/net_alerter/oui.db"
ENV_PATH = Path(os.getenv("NET_ALERTER_ENV", "/opt/net_alerter/.env"))
BLE_NAMES_FILE = Path("/opt/net_alerter/ble_names.json")
BLE_CORRELATION_WINDOW = 60 # seconds — how far back to look for nearby BLE devices
ARP_SCAN_INTERVAL = int(os.getenv("NET_ALERTER_ARP_SCAN_INTERVAL", "30"))
known_devices = {} # {mac: {ip, hostname, vendor, first_seen, last_seen, infrastructure (opt)}}
known_lock = threading.Lock()
@@ -1309,6 +1310,199 @@ def seed_infrastructure_ips(iface: str) -> None:
logging.info(f"Seeded extra infrastructure IP {ip} from env")
def _get_iface_mac(iface: str) -> bytes:
"""Return interface MAC as 6 bytes, or zero bytes on failure."""
try:
raw = open(f"/sys/class/net/{iface}/address").read().strip()
return bytes(int(x, 16) for x in raw.split(':'))
except Exception:
return b'\x00' * 6
def arp_scanner(iface: str) -> None:
"""Active ARP scan thread.
Sends ARP who-has to every IP in the local /24 subnet every ARP_SCAN_INTERVAL
seconds. Every device that replies triggers on_arrival, bypassing AP ARP proxy.
This is the reliable presence detection path for devices with existing DHCP leases
that never broadcast ARP on their own.
"""
logging.info(f"ARP scanner started on {iface} (interval={ARP_SCAN_INTERVAL}s)")
src_mac = _get_iface_mac(iface)
while True:
try:
src_ip = get_local_ip(iface)
if not src_ip:
time.sleep(ARP_SCAN_INTERVAL)
continue
prefix = '.'.join(src_ip.split('.')[:3])
src_ip_b = socket.inet_aton(src_ip)
broadcast = b'\xff\xff\xff\xff\xff\xff'
sock = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(0x0806))
sock.bind((iface, 0))
sock.settimeout(0.1)
# Send ARP who-has for every IP in the /24
for i in range(1, 255):
target_ip = f"{prefix}.{i}"
if target_ip == src_ip:
continue
eth = broadcast + src_mac + b'\x08\x06'
arp = struct.pack('!HHBBH', 1, 0x0800, 6, 4, 1)
arp += src_mac + src_ip_b + b'\x00' * 6 + socket.inet_aton(target_ip)
try:
sock.send(eth + arp)
except Exception:
pass
# Collect replies for 2 seconds
deadline = time.time() + 2.0
while time.time() < deadline:
try:
data = sock.recv(60)
if len(data) < 42 or data[12:14] != b'\x08\x06':
continue
if struct.unpack('!H', data[20:22])[0] != 2: # opcode must be reply
continue
sender_mac = ':'.join(f'{b:02x}' for b in data[22:28])
sender_ip = socket.inet_ntoa(data[28:32])
if sender_mac in ('00:00:00:00:00:00', 'ff:ff:ff:ff:ff:ff'):
continue
_update_last_seen(sender_mac)
on_arrival(sender_mac, sender_ip)
except socket.timeout:
continue
except Exception:
break
sock.close()
except Exception as e:
logging.error(f"ARP scanner error: {e}")
time.sleep(ARP_SCAN_INTERVAL)
def _find_monitor_iface(primary_iface: str) -> str:
"""Return a second WiFi interface name for monitor mode, or '' if none found."""
try:
result = subprocess.run(['iw', 'dev'], capture_output=True, text=True, timeout=5)
for line in result.stdout.splitlines():
line = line.strip()
if line.startswith('Interface '):
iface = line.split()[1]
if iface != primary_iface:
return iface
except Exception:
pass
return ''
def _enable_monitor_mode(iface: str) -> bool:
"""Put iface into 802.11 monitor mode. Returns True on success."""
try:
subprocess.run(['ip', 'link', 'set', iface, 'down'],
check=True, capture_output=True, timeout=5)
subprocess.run(['iw', 'dev', iface, 'set', 'type', 'monitor'],
check=True, capture_output=True, timeout=5)
subprocess.run(['ip', 'link', 'set', iface, 'up'],
check=True, capture_output=True, timeout=5)
return True
except Exception as e:
logging.warning(f"Monitor mode setup failed on {iface}: {e}")
return False
def _parse_80211_frame(frame: bytes) -> None:
"""Parse a radiotap+802.11 frame and call on_arrival for transmitting personal devices.
addr2 is always the transmitter. We care about:
- Management frames from stations (assoc req subtype=0, auth subtype=11)
- Data frames with ToDS=1 (station → AP)
Probe requests (subtype=4) are excluded — modern iOS uses random MACs for those.
"""
try:
if len(frame) < 4 or frame[0] != 0: # radiotap version must be 0
return
rt_len = struct.unpack_from('<H', frame, 2)[0]
if rt_len > len(frame) - 24:
return
dot11 = frame[rt_len:]
if len(dot11) < 24:
return
fc = struct.unpack_from('<H', dot11, 0)[0]
frame_type = (fc >> 2) & 0x3
frame_subtype = (fc >> 4) & 0xF
# addr2 = transmitter (bytes 10-15)
src_mac = ':'.join(f'{b:02x}' for b in dot11[10:16])
if src_mac in ('00:00:00:00:00:00', 'ff:ff:ff:ff:ff:ff'):
return
is_station_tx = False
if frame_type == 0 and frame_subtype in (0, 11): # assoc req, auth
is_station_tx = True
elif frame_type == 2: # data
to_ds = (fc >> 8) & 0x1
from_ds = (fc >> 9) & 0x1
if to_ds and not from_ds: # STA → AP
is_station_tx = True
if not is_station_tx:
return
_update_last_seen(src_mac)
if not is_personal_device(src_mac):
return
with known_lock:
already_known = src_mac in known_devices
if not already_known:
on_arrival(src_mac, 'unknown')
except Exception:
pass
def monitor_sniffer(iface: str) -> None:
"""802.11 monitor mode sniffer thread.
Reads raw radiotap+802.11 frames and calls on_arrival for personal devices
seen transmitting. Bypasses AP entirely — works even when AP proxy hides
all ARP/DHCP traffic.
"""
if not _enable_monitor_mode(iface):
logging.warning(f"Monitor mode unavailable on {iface}, monitor sniffer disabled")
return
logging.info(f"Monitor mode sniffer started on {iface}")
try:
sock = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(0x0003))
sock.bind((iface, 0))
sock.settimeout(1.0)
except Exception as e:
logging.error(f"Monitor mode socket failed on {iface}: {e}")
return
while True:
try:
frame = sock.recv(2048)
_parse_80211_frame(frame)
except socket.timeout:
continue
except Exception as e:
logging.debug(f"Monitor sniffer recv error: {e}")
time.sleep(0.1)
def ble_sniffer() -> None:
"""
BLE sniffer thread.
@@ -1965,12 +2159,18 @@ def main() -> None:
threading.Thread(target=netlink_watcher, daemon=True, name='netlink-watcher'),
threading.Thread(target=personal_watchdog, daemon=True, name='personal-watchdog'),
threading.Thread(target=ble_sniffer, daemon=True, name='ble-sniffer'),
threading.Thread(target=arp_scanner, args=(iface,), daemon=True, name='arp-scanner'),
]
monitor_iface = _find_monitor_iface(iface)
if monitor_iface:
logging.info(f"Monitor mode interface found: {monitor_iface}")
threads.append(threading.Thread(target=monitor_sniffer, args=(monitor_iface,), daemon=True, name='monitor-sniffer'))
for t in threads:
t.start()
logging.info("Net alerter running — DHCP sniffer + Netlink neighbor watcher + personal device watchdog + BLE sniffer active")
logging.info("Net alerter running — DHCP sniffer + Netlink neighbor watcher + personal watchdog + BLE sniffer + ARP scanner active")
# Keep main thread alive
try: