n0mad1k/bigbrother
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
0b5c7f3907d295e5351172f48ce63fb6e836aa45
bigbrother/net_alerter/net_alerter.py
T
Cobra 71cd3332d2 Fix ARRIVED spam — mark departing instead of popping from known_devices 
When on_departure popped the device from known_devices, rapid RTM_NEWNEIGH
events during ARP cache rebuild each triggered on_arrival as if the device
was brand new, flooding the Matrix room with 5-10+ duplicate ARRIVED alerts.

Fix: keep device in known_devices with departing=True flag. on_arrival
silently cancels pending departure timers and returns without alerting when
it sees this flag. Timer callback checks flag before sending the alert and
only pops the device when departure is confirmed after 15 minutes.
2026-04-09 14:34:46 -04:00

869 lines
26 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""
net_alerter.py — Net alerter persistent daemon.
Monitors device presence via DHCP sniffing + Netlink neighbor events.
Zero active probing. No traffic generated.
Three concurrent threads:
1. DHCP sniffer — AF_PACKET raw socket, captures DHCP traffic passively
2. RTM_NEWNEIGH watcher — Netlink socket, kernel pushes new ARP neighbor events
3. RTM_DELNEIGH watcher — Netlink socket, kernel pushes deleted ARP neighbor events
"""
import json
import logging
import os
import socket
import struct
import sys
import threading
import time
import urllib.parse
import urllib.request
import sqlite3
from pathlib import Path
# --- Config (from .env or env vars) ---
MATRIX_HOMESERVER = ""
MATRIX_ACCESS_TOKEN = ""
MATRIX_ROOM_ID = ""
LOG_FILE = "/opt/net_alerter/net_alerter.log"
OUI_DB_PATH = "/opt/net_alerter/oui.db"
known_devices = {} # {mac: {ip, hostname, vendor, first_seen, last_seen, infrastructure (opt)}}
known_lock = threading.Lock()
infrastructure_ips = set() # IPs that should never trigger alerts (gateway, broadcast, self)
departure_timers = {} # {mac: threading.Timer} for 15-min departure debounce
departure_lock = threading.Lock() # Protects departure_timers and last_departed_time
last_departed_time = {} # {mac: timestamp} when device was actually alerted as departed
# Top 200 common device OUI prefixes (MAC first 3 octets)
OUI_DICT = {
"88A29E": "Apple",
"001A7D": "Apple",
"185F3F": "Apple",
"A4C3F0": "Apple",
"0899D8": "Apple",
"004096": "Apple",
"00219B": "Apple",
"0021E9": "Apple",
"005973": "Apple",
"006377": "Apple",
"0064B9": "Apple",
"0084F3": "Apple",
"00A04D": "Apple",
"00D04B": "Apple",
"28879F": "Google",
"2887BA": "Google",
"5427EB": "Google",
"542758": "Google",
"341513": "Amazon",
"0C47C2": "Amazon",
"B827EB": "Raspberry Pi",
"2C56DC": "Raspberry Pi",
"E45F01": "Raspberry Pi",
"000D82": "Intel",
"001025": "Intel",
"001ABA": "Intel",
"001F3B": "Intel",
"001F3C": "Intel",
"001BA9": "Intel",
"001BFB": "Intel",
"001D7A": "Intel",
"001DB8": "Intel",
"001E67": "Intel",
"001F29": "Intel",
"F44DA2": "Intel",
"B03C1C": "Intel",
"00166B": "Qualcomm",
"001BF3": "Qualcomm",
"001C47": "Qualcomm",
"002000": "Qualcomm",
"0026F2": "Qualcomm",
"0026F3": "Qualcomm",
"0026F4": "Qualcomm",
"002618": "Qualcomm",
"FFFFFF": "Espressif",
"687DA8": "Espressif",
"A01D48": "Espressif",
"30AEA4": "Espressif",
"B4E63B": "Espressif",
"80E386": "TP-Link",
"8CBF26": "TP-Link",
"90843F": "TP-Link",
"A838B4": "TP-Link",
"C81F66": "TP-Link",
"DCFEEE": "TP-Link",
"F81A67": "TP-Link",
"689B5A": "Netgear",
"6C46CB": "Netgear",
"78A21B": "Netgear",
"94B3D5": "Netgear",
"B0EE7B": "Netgear",
"B4FBE4": "Netgear",
"C8D7B4": "Netgear",
"D8BB85": "Netgear",
"F8AB05": "Netgear",
"001D2F": "Cisco",
"001930": "Cisco",
"001A2F": "Cisco",
"001C0E": "Cisco",
"001CED": "Cisco",
"001EBD": "Cisco",
"001F6C": "Cisco",
"0021A0": "Cisco",
"002678": "Cisco",
"0026CB": "Cisco",
"00266E": "Cisco",
"00269F": "Cisco",
"0026A6": "Cisco",
"002700": "Cisco",
"0026FA": "Cisco",
"00267B": "Cisco",
"0C1234": "Cisco",
"001BD7": "Dell",
"001BDB": "Dell",
"001A6B": "Dell",
"003067": "Dell",
"00188B": "Dell",
"001B21": "Dell",
"001E67": "Dell",
"0026F7": "Dell",
"00242A": "Dell",
"001C23": "HP",
"001E0B": "HP",
"001EBB": "HP",
"001F2F": "HP",
"001BFC": "HP",
"002219": "HP",
"0022A6": "HP",
"002264": "HP",
"0024A9": "HP",
"0026B9": "HP",
"0026DF": "HP",
"001A6C": "Lenovo",
"001DFB": "Lenovo",
"001F3A": "Lenovo",
"001A73": "Lenovo",
"00238B": "Lenovo",
"002264": "Lenovo",
"00B0D0": "Lenovo",
"9C2EBF": "Lenovo",
"F0DE8D": "Lenovo",
"00E04C": "Huawei",
"001097": "Huawei",
"001093": "Huawei",
"001A2B": "Huawei",
"001D3C": "Huawei",
"001E37": "Huawei",
"001F3E": "Huawei",
"001ECE": "Huawei",
"002074": "Huawei",
"00207B": "Huawei",
"0020E2": "Huawei",
"0020F1": "Huawei",
"0020F5": "Huawei",
"002268": "Huawei",
"002269": "Huawei",
"00226A": "Huawei",
"0022AA": "Xiaomi",
"002255": "Xiaomi",
"003677": "Xiaomi",
"3418C2": "Xiaomi",
"50EB71": "Xiaomi",
"5CF4D4": "Xiaomi",
"78111D": "Xiaomi",
"78E1D3": "Xiaomi",
"A4ABDA": "Xiaomi",
"E4F47A": "Xiaomi",
"F45EAB": "Xiaomi",
"A4D18D": "OnePlus",
"50F5DA": "OnePlus",
"8C03BE": "OnePlus",
"002261": "LG",
"002265": "LG",
"0022D2": "LG",
"0022D3": "LG",
"000426": "LG",
"001854": "LG",
"001E8F": "LG",
"001F5B": "LG",
"002038": "LG",
"0020A6": "LG",
"0020F8": "LG",
"002220": "LG",
"00222D": "LG",
"0027E4": "LG",
"00242C": "LG",
"002531": "LG",
"001054": "Sony",
"001458": "Sony",
"0014EA": "Sony",
"001525": "Sony",
"00188A": "Sony",
"001B44": "Sony",
"00215E": "Sony",
"002178": "Sony",
"0021C5": "Sony",
"002333": "Sony",
"002481": "Sony",
"0026FF": "Sony",
"002702": "Nintendo",
"0008FE": "Nintendo",
"0013F7": "Nintendo",
"001BDB": "Nintendo",
"001EEE": "Nintendo",
"001FEE": "Nintendo",
"0020F8": "Nintendo",
"002095": "Nintendo",
"0020FA": "Nintendo",
"00227B": "Nintendo",
}
def setup_logging():
"""Setup logging to file + stdout."""
formatter = logging.Formatter(
fmt='%(asctime)s [%(levelname)s] %(message)s',
datefmt='%Y-%m-%dT%H:%M:%S'
)
# File handler
Path(LOG_FILE).parent.mkdir(parents=True, exist_ok=True)
file_handler = logging.FileHandler(LOG_FILE)
file_handler.setFormatter(formatter)
# Stdout handler
stdout_handler = logging.StreamHandler(sys.stdout)
stdout_handler.setFormatter(formatter)
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
logger.addHandler(file_handler)
logger.addHandler(stdout_handler)
return logger
def load_env():
"""Parse .env file manually (key=value), no dependency required."""
global MATRIX_HOMESERVER, MATRIX_ACCESS_TOKEN, MATRIX_ROOM_ID
env_path = Path(os.getenv("NET_ALERTER_ENV", "/opt/net_alerter/.env"))
if not env_path.exists():
logging.warning(f".env not found at {env_path}, using env vars only")
MATRIX_HOMESERVER = os.getenv("MATRIX_HOMESERVER", "https://m.example.org")
MATRIX_ACCESS_TOKEN = os.getenv("MATRIX_ACCESS_TOKEN", "")
MATRIX_ROOM_ID = os.getenv("MATRIX_ROOM_ID", "")
return
try:
for line in env_path.read_text().splitlines():
line = line.strip()
if not line or line.startswith("#"):
continue
if "=" not in line:
continue
key, val = line.split("=", 1)
if key == "MATRIX_HOMESERVER":
MATRIX_HOMESERVER = val
elif key == "MATRIX_ACCESS_TOKEN":
MATRIX_ACCESS_TOKEN = val
elif key == "MATRIX_ROOM_ID":
MATRIX_ROOM_ID = val
logging.info(f"Loaded config from {env_path}")
except Exception as e:
logging.error(f"Failed to load .env: {e}")
def lookup_oui(mac: str) -> str:
"""
Look up OUI vendor from MAC address (first 3 octets).
Tries SQLite database first, falls back to inline dictionary.
Returns vendor name or 'unknown' if not found.
"""
if not mac or mac == "00:00:00:00:00:00":
return "unknown"
# Try database first
oui_prefix = mac[:8].lower()
try:
conn = sqlite3.connect(OUI_DB_PATH, timeout=1)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
result = cursor.execute("SELECT vendor FROM oui WHERE prefix = ?", (oui_prefix,)).fetchone()
conn.close()
if result:
return result['vendor']
except Exception:
# Log at DEBUG only, no output noise
logging.debug(f"OUI database lookup failed for {oui_prefix}", exc_info=False)
# Fall back to inline dictionary
oui_prefix_upper = oui_prefix.replace(":", "").upper()
return OUI_DICT.get(oui_prefix_upper, "unknown")
def seed_from_arp_cache():
"""Read /proc/net/arp, populate known_devices without alerting."""
global known_devices
try:
lines = Path("/proc/net/arp").read_text().splitlines()
for line in lines:
if line.startswith("IP"):
continue
fields = line.split()
if len(fields) < 4:
continue
ip = fields[0]
mac = fields[3].lower()
flags = fields[2]
if mac == "00:00:00:00:00:00" or flags == "0x0":
continue
vendor = lookup_oui(mac)
with known_lock:
if mac not in known_devices:
dev = {
'ip': ip,
'hostname': ip,
'vendor': vendor,
'first_seen': time.time(),
'last_seen': time.time()
}
# Mark infrastructure IPs so they never alert
if ip in infrastructure_ips:
dev['infrastructure'] = True
known_devices[mac] = dev
logging.info(f"Seeded {len(known_devices)} devices from ARP cache")
except Exception as e:
logging.error(f"Failed to seed from ARP cache: {e}")
def send_alert(msg: str) -> None:
"""Send alert to Matrix room."""
if not MATRIX_ACCESS_TOKEN:
logging.warning("No MATRIX_ACCESS_TOKEN set, skipping alert")
return
url = (
f"{MATRIX_HOMESERVER}/_matrix/client/v3/rooms/"
f"{urllib.parse.quote(MATRIX_ROOM_ID, safe='')}/send/m.room.message"
)
payload = json.dumps({"msgtype": "m.text", "body": msg}).encode()
req = urllib.request.Request(
url,
data=payload,
headers={
"Authorization": f"Bearer {MATRIX_ACCESS_TOKEN}",
"Content-Type": "application/json",
"User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36",
},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=10) as r:
pass
logging.debug(f"Alert sent to Matrix")
except urllib.error.HTTPError as e:
logging.error(f"Matrix send failed with HTTP {e.code}")
except Exception as e:
logging.error(f"Matrix send exception: {e}")
def fmt_duration(secs: float) -> str:
"""Format duration in seconds to human-readable format."""
if secs < 60:
return f"{int(secs)}s"
elif secs < 3600:
m = int(secs // 60)
s = int(secs % 60)
return f"{m}m {s}s" if s else f"{m}m"
else:
h = int(secs // 3600)
m = int((secs % 3600) // 60)
return f"{h}h {m}m" if m else f"{h}h"
def format_arrival(hostname: str, ip: str, vendor: str, mac: str) -> str:
"""Format arrival alert message. Deduplicate IP if hostname equals IP."""
if hostname == ip:
# Hostname resolution failed, don't repeat IP
return f"[NET] ARRIVED: {ip} [{vendor}] MAC:{mac}"
else:
# Hostname resolved successfully, show both
return f"[NET] ARRIVED: {hostname} ({ip}) [{vendor}] MAC:{mac}"
def format_departure(hostname: str, ip: str, vendor: str, mac: str, duration: str) -> str:
"""Format departure alert message. Deduplicate IP if hostname equals IP."""
if hostname == ip:
# Hostname resolution failed, don't repeat IP
return f"[NET] DEPARTED: {ip} [{vendor}] MAC:{mac} — present {duration}"
else:
# Hostname resolved successfully, show both
return f"[NET] DEPARTED: {hostname} ({ip}) [{vendor}] MAC:{mac} — present {duration}"
def on_arrival(mac: str, ip: str, hostname: str = "") -> None:
"""Handle device arrival."""
# Infrastructure IPs never trigger alerts
if ip in infrastructure_ips:
logging.debug(f"Ignoring infrastructure IP {ip} (MAC:{mac})")
return
hostname = hostname or ip
now = time.time()
# Silent re-arrival if device marked departing (rapid flap recovery)
with departure_lock:
if mac in known_devices and known_devices[mac].get('departing'):
# Device returned during departure window — cancel timer, clear flag, no alert
if mac in departure_timers:
departure_timers[mac].cancel()
del departure_timers[mac]
with known_lock:
known_devices[mac]['departing'] = False
last_departed_time.pop(mac, None)
logging.debug(f"Device {mac} re-arrived during departure window — suppressed alert")
return
# Check re-arrival suppression BEFORE acquiring lock (avoid nested locks)
with departure_lock:
# ALWAYS cancel any pending departure timer when device re-arrives
if mac in departure_timers:
departure_timers[mac].cancel()
del departure_timers[mac]
logging.debug(f"Cancelled departure timer for {mac} — device re-arrived before debounce expired")
if mac in last_departed_time:
time_since_departure = now - last_departed_time[mac]
if time_since_departure < 300: # 5 min
logging.debug(f"Suppressing re-arrival alert for {mac} (departed {int(time_since_departure)}s ago)")
# Re-add to known_devices to track presence
with known_lock:
if mac not in known_devices:
vendor = lookup_oui(mac)
known_devices[mac] = {
'ip': ip,
'hostname': hostname,
'vendor': vendor,
'first_seen': now,
'last_seen': now
}
else:
known_devices[mac]['last_seen'] = now
return
with known_lock:
if mac in known_devices:
# Device is already known
dev = known_devices[mac]
dev['last_seen'] = now
# DHCP renewal dedup: if last_seen < 30 min, skip alert
if now - dev['first_seen'] < 1800: # 30 min
logging.debug(f"Skipping DHCP renewal alert for {mac} (seen {int(now - dev['first_seen'])}s ago)")
return
# Not a re-arrival, update last_seen and don't alert (already known)
return
# New device
vendor = lookup_oui(mac)
known_devices[mac] = {
'ip': ip,
'hostname': hostname,
'vendor': vendor,
'first_seen': now,
'last_seen': now
}
msg = format_arrival(hostname, ip, vendor or "unknown", mac)
logging.info(msg)
send_alert(msg)
def on_departure(mac: str) -> None:
"""
Handle device departure with 15-minute debounce.
Only sends alert if device still gone after 15 min.
"""
with known_lock:
if mac not in known_devices:
return
# Check if infrastructure IP (never alert)
dev = known_devices[mac]
if dev['ip'] in infrastructure_ips:
logging.debug(f"Ignoring departure for infrastructure IP {dev['ip']} (MAC:{mac})")
return
dev_copy = dev.copy()
# Mark departing instead of popping — allows re-arrival to detect and cancel timer
known_devices[mac]['departing'] = True
# Start 15-minute debounce timer
with departure_lock:
# Cancel any existing timer for this MAC
if mac in departure_timers:
departure_timers[mac].cancel()
def send_departure_alert():
"""Callback to send alert after debounce expires."""
# Check if device re-arrived (departing flag cleared by on_arrival)
with known_lock:
if not known_devices.get(mac, {}).get('departing'):
return # device re-arrived, departure was cancelled
# Device still departed — send alert and remove from tracking
known_devices.pop(mac, None)
duration = fmt_duration(time.time() - dev_copy['first_seen'])
msg = format_departure(dev_copy['hostname'], dev_copy['ip'], dev_copy['vendor'] or "unknown", mac, duration)
logging.info(msg)
send_alert(msg)
# Record departure time for re-arrival suppression
with departure_lock:
last_departed_time[mac] = time.time()
if mac in departure_timers:
del departure_timers[mac]
timer = threading.Timer(900.0, send_departure_alert) # 15 min = 900 sec
timer.daemon = True
timer.name = f'departure-debounce-{mac}'
departure_timers[mac] = timer
timer.start()
logging.debug(f"Departure debounce started for {mac}, will alert in 15 min if still gone")
def get_primary_interface() -> str:
"""Get primary network interface (not lo)."""
try:
with open('/proc/net/route') as f:
for line in f.readlines()[1:]:
parts = line.split()
if len(parts) > 1 and parts[0] != 'lo' and parts[1] == '00000000':
return parts[0]
with open('/proc/net/route') as f:
for line in f.readlines()[1:]:
parts = line.split()
if parts[0] != 'lo':
return parts[0]
except Exception:
pass
return 'eth0'
def get_local_ip(iface: str) -> str:
"""Get the local IP address for the given interface."""
try:
import subprocess
result = subprocess.run(['ip', 'addr', 'show', iface], capture_output=True, text=True, timeout=5)
if result.returncode == 0:
for line in result.stdout.split('\n'):
if 'inet ' in line:
parts = line.strip().split()
if len(parts) >= 2:
ip_with_mask = parts[1]
ip = ip_with_mask.split('/')[0]
return ip
except Exception as e:
logging.warning(f"Failed to get local IP for {iface}: {e}")
return ""
def seed_infrastructure_ips(iface: str) -> None:
"""
Seed infrastructure IPs that should never trigger alerts.
Reads gateway from 'ip route show default', adds broadcast and self IP.
"""
global infrastructure_ips
infrastructure_ips.clear()
# Get self IP
self_ip = get_local_ip(iface)
if self_ip:
infrastructure_ips.add(self_ip)
logging.info(f"Seeded self IP {self_ip} as infrastructure")
# Read gateway from 'ip route show default'
try:
import subprocess
result = subprocess.run(['ip', 'route', 'show', 'default'], capture_output=True, text=True, timeout=5)
if result.returncode == 0:
# Parse: default via 10.0.0.0 dev eth0 proto dhcp metric 100
for line in result.stdout.strip().split('\n'):
if 'via' in line:
parts = line.split()
for i, part in enumerate(parts):
if part == 'via' and i + 1 < len(parts):
gateway_ip = parts[i + 1]
infrastructure_ips.add(gateway_ip)
logging.info(f"Seeded gateway {gateway_ip} as infrastructure")
break
except Exception as e:
logging.warning(f"Failed to read gateway: {e}")
# Add broadcast (x.x.x.255 for most networks)
if self_ip:
try:
parts = self_ip.rsplit('.', 1)
if len(parts) == 2:
broadcast_ip = parts[0] + ".255"
infrastructure_ips.add(broadcast_ip)
logging.info(f"Seeded broadcast {broadcast_ip} as infrastructure")
except Exception:
pass
logging.info(f"Infrastructure IPs: {infrastructure_ips}")
def dhcp_sniffer(iface: str) -> None:
"""
DHCP sniffer thread.
AF_PACKET raw socket, captures DHCP traffic passively.
"""
try:
s = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(0x0003))
s.bind((iface, 0))
logging.info(f"DHCP sniffer started on {iface}")
while True:
try:
data, _ = s.recvfrom(65535)
parse_dhcp(data)
except Exception as e:
logging.error(f"DHCP sniffer error: {e}")
time.sleep(1)
except Exception as e:
logging.error(f"Failed to start DHCP sniffer: {e}")
def parse_dhcp(data: bytes) -> None:
"""Parse DHCP packet and trigger arrival/departure events."""
try:
if len(data) < 14:
return
# Ethernet frame
eth_type = struct.unpack('!H', data[12:14])[0]
if eth_type != 0x0800: # not IPv4
return
# IP header
if len(data) < 23:
return
proto = data[23]
if proto != 17: # not UDP
return
# UDP ports
if len(data) < 38:
return
src_port = struct.unpack('!H', data[34:36])[0]
dst_port = struct.unpack('!H', data[36:38])[0]
if not (src_port in (67, 68) and dst_port in (67, 68)):
return
# DHCP payload (starts at byte 42)
if len(data) < 42:
return
dhcp = data[42:]
if len(dhcp) < 236:
return
# MAC address (chaddr at offset 28, 6 bytes)
mac_bytes = dhcp[28:34]
mac = ':'.join(f'{b:02x}' for b in mac_bytes)
if mac in ('00:00:00:00:00:00', 'ff:ff:ff:ff:ff:ff'):
return
# Parse DHCP options (start at byte 240)
if len(dhcp) < 240:
return
msg_type = None
hostname = ""
requested_ip = ""
ciaddr = socket.inet_ntoa(dhcp[12:16])
i = 240
while i < len(dhcp):
opt = dhcp[i]
if opt == 255:
break
if opt == 0:
i += 1
continue
if i + 1 >= len(dhcp):
break
length = dhcp[i + 1]
if i + 2 + length > len(dhcp):
break
val = dhcp[i + 2:i + 2 + length]
if opt == 53 and length == 1: # DHCP message type
msg_type = val[0]
elif opt == 12: # Hostname
hostname = val.decode('utf-8', errors='replace').strip('\x00')
elif opt == 50 and length == 4: # Requested IP
requested_ip = socket.inet_ntoa(val)
i += 2 + length
# Process based on message type
if msg_type in (1, 3): # Discover or Request (arrival)
ip = requested_ip or ciaddr
if ip and ip != '0.0.0.0':
on_arrival(mac, ip, hostname)
elif msg_type == 7: # Release (departure)
on_departure(mac)
except Exception as e:
logging.debug(f"DHCP parse error: {e}")
# Netlink constants
NETLINK_ROUTE = 0
RTMGRP_NEIGH = 0x4
RTM_NEWNEIGH = 28
RTM_DELNEIGH = 29
NUD_REACHABLE = 0x02
NUD_STALE = 0x04
NUD_DELAY = 0x08
NUD_PROBE = 0x10
NDA_DST = 1
NDA_LLADDR = 2
def netlink_watcher() -> None:
"""
Netlink watcher thread.
Kernel pushes RTM_NEWNEIGH and RTM_DELNEIGH events.
"""
try:
s = socket.socket(socket.AF_NETLINK, socket.SOCK_RAW, NETLINK_ROUTE)
s.bind((os.getpid(), RTMGRP_NEIGH))
logging.info("Netlink neighbor watcher started")
while True:
try:
data = s.recv(65535)
parse_netlink(data)
except Exception as e:
logging.error(f"Netlink watcher error: {e}")
time.sleep(1)
except Exception as e:
logging.error(f"Failed to start Netlink watcher: {e}")
def parse_netlink(data: bytes) -> None:
"""Parse Netlink neighbor messages."""
try:
offset = 0
while offset < len(data):
if offset + 16 > len(data):
break
nlmsg_len, nlmsg_type, _, _, _ = struct.unpack_from('=IHHII', data, offset)
if nlmsg_len < 16:
break
payload = data[offset + 16:offset + nlmsg_len]
if nlmsg_type in (RTM_NEWNEIGH, RTM_DELNEIGH):
parse_ndmsg(payload, nlmsg_type)
offset += (nlmsg_len + 3) & ~3
except Exception as e:
logging.debug(f"Netlink parse error: {e}")
def parse_ndmsg(data: bytes, msg_type: int) -> None:
"""Parse ndmsg (neighbor discovery message)."""
try:
if len(data) < 12:
return
state = struct.unpack_from('=H', data, 8)[0]
mac = None
ip = None
offset = 12
while offset + 4 <= len(data):
rta_len, rta_type = struct.unpack_from('=HH', data, offset)
if rta_len < 4:
break
val = data[offset + 4:offset + rta_len]
if rta_type == NDA_LLADDR and len(val) == 6:
mac = ':'.join(f'{b:02x}' for b in val)
elif rta_type == NDA_DST:
if len(val) == 4:
ip = socket.inet_ntoa(val)
offset += (rta_len + 3) & ~3
if not mac or mac in ('00:00:00:00:00:00', 'ff:ff:ff:ff:ff:ff'):
return
if msg_type == RTM_NEWNEIGH and (state & (NUD_REACHABLE | NUD_STALE | NUD_DELAY | NUD_PROBE)):
on_arrival(mac, ip or 'unknown')
elif msg_type == RTM_DELNEIGH:
on_departure(mac)
except Exception as e:
logging.debug(f"ndmsg parse error: {e}")
def main() -> None:
"""Main entry point."""
logger = setup_logging()
load_env()
iface = get_primary_interface()
logging.info(f"Net alerter starting on interface {iface}")
seed_infrastructure_ips(iface)
seed_from_arp_cache()
# Start monitor threads
threads = [
threading.Thread(target=dhcp_sniffer, args=(iface,), daemon=True, name='dhcp-sniffer'),
threading.Thread(target=netlink_watcher, daemon=True, name='netlink-watcher'),
]
for t in threads:
t.start()
logging.info("Net alerter running — DHCP sniffer + Netlink neighbor watcher active")
# Keep main thread alive
try:
while True:
time.sleep(60)
with known_lock:
logging.debug(f"Tracking {len(known_devices)} devices")
except KeyboardInterrupt:
logging.info("Shutting down")
sys.exit(0)
if __name__ == '__main__':
main()
Reference in New Issue View Git Blame Copy Permalink