#!/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 re 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 # Interface flap detection — suppresses departure storms _flap_window = [] # [(timestamp, mac)] _flap_lock = threading.Lock() _interface_recovering = False _recovery_timer = None # Network equipment OUI prefixes (first 3 octets, uppercase no colons) # Auto-exclusion for known infrastructure devices NETWORK_EQUIPMENT_OUIS = { # Ubiquiti "0418D6", "24A43C", "44D9E7", "68729C", "788A20", "802AA8", "9C05D6", "DC9FDB", "E063DA", "F09FC2", "6C63F8", "B4FBE4", "00156D", "002722", "04186F", # Cisco "000C29", "001A2F", "001BB1", "001E13", "001F26", "002155", "0060B0", "00902B", "008024", "0090AB", "00E014", "001011", # Aruba "000B86", "00FC8B", "040E3C", "1C28AF", "40E3D6", "6C29D1", "84D47E", "884774", "94B40F", "AC3744", # TP-Link "000AEB", "001777", "1C61B4", "5C63BF", "6046D9", "A84E3F", "B4B024", "C46AB7", "EC086B", "F81A67", # Netgear "001B2F", "00146C", "002196", "00A0CC", "081102", "1C1E29", "206A8A", "30469A", "4F2AFF", "6CB0CE", "8C3BAD", # MikroTik "4C5E0C", "B8069F", "C4AD34", "CC2DE0", "D4CA6D", "E48D8C", # Ruckus "000C67", "00249B", "04BD88", "14778B", "2C5D93", "34A84E", } # Churn tracking — auto-suppresses devices that cycle 3+ times in 30min _churn_tracker = {} # {mac: [timestamps of departure events]} _churn_lock = threading.Lock() CHURN_THRESHOLD = 3 # cycles in window before suppression CHURN_WINDOW_SEC = 1800 # 30 minutes # Mobile device OUI prefixes for personal device detection MOBILE_DEVICE_OUIS = { # Apple "28A02E", "A4C3F0", "00A0C6", "38C086", "5C4950", "7061D0", "88A29E", "A81D6A", "A8BBCF", "AC3744", "B0EE7B", "B82DFE", "BC9674", "C8DFC0", "D4996F", "D8BB85", "E0D55E", "F4CE46", "F8FF9B", "FCFC48", # Google Pixel "28876D", "34E5FB", "4C80F3", "70EBF2", "AC16E5", "D0A05F", "E4C547", # Samsung "00166B", "001843", "001D67", "001FAF", "0060B0", "084ECE", "0825F3", "088038", "0A49EF", "0C1432", "0EBB1B", "1851BF", "205C85", "20C18B", "240A64", "2C5A7B", "32A639", "360DFE", "3A590D", "3CC06C", "441735", "44B16C", "48D76E", "58D55C", "5C2EE4", "68C903", "6AC4C6", "6EAA0A", "7045AC", "7499EA", "74D4DD", "781F02", "789098", "7C59B9", "7E0A6D", # OnePlus "00166B", "001BF3", "001C47", "002000", "0026F2", "3C28CB", "38DB23", "42BAFC", "5C55F9", "681E7D", "A8178E", "AA6B1F", "BC1E1A", "EA96D7", # Xiaomi "0022AA", "002255", "003677", "3418C2", "48EE0C", "54EF25", "7418DB", "7C8B5C", "7E4A5D", "868A5A", "8A6FB8", "8C8E14", "8EBE59", "A0871B", "B8328D", "B832D6", "BA5294", "BC7C47", "C8816D", "E4AF17", "E899C3", # Huawei "00E04C", "001097", "001093", "001A2B", "001D3C", "001E37", "001F3E", "001ECE", "002074", "00207B", "0020E2", "0020F1", "0020F5", "002268", "002269", "00226A", # LG "001E4B", "001E8D", "00218B", "002215", "0022FB", "0023B2", "0023F8", "002655", "00266C", "00AB8B", "30F9ED", "D8D0DF", # HTC "001072", "001058", "0010C6", "001153", "001167", "0011BD", "00121F", "00121C", "001212", "001236", "00124B", "00131D", "0013D4", "0013E8", # Motorola "001555", "001620", "001645", "001693", "0016B8", "001730", "0017F6", "001822", "001888", "0018F8", "001913", "001985", "001A45", "001A92", "001AFC", } # Personal device tracking personal_devices = set() # {mac: ...} of detected personal device MACs personal_lock = threading.Lock() location_occupancy = "UNKNOWN" # Current location state: VACANT, OCCUPIED, UNKNOWN occupancy_lock = threading.Lock() def _check_flap(mac: str) -> bool: """Return True if this departure is part of an interface flap (suppress it).""" global _interface_recovering, _recovery_timer now = time.time() with _flap_lock: _flap_window[:] = [(t, m) for t, m in _flap_window if now - t < 3.0] _flap_window.append((now, mac)) if len(_flap_window) >= 3 and not _interface_recovering: _interface_recovering = True if _recovery_timer: _recovery_timer.cancel() _recovery_timer = threading.Timer(60.0, _clear_recovery) _recovery_timer.daemon = True _recovery_timer.start() logging.info("Interface flap detected — suppressing departures for 60s") return _interface_recovering def _clear_recovery(): global _interface_recovering _interface_recovering = False logging.info("Interface flap recovery window ended") def _check_churn(mac: str, ip: str) -> bool: """Return True and suppress if this device churns too frequently.""" now = time.time() with _churn_lock: events = _churn_tracker.get(mac, []) events = [t for t in events if now - t < CHURN_WINDOW_SEC] events.append(now) _churn_tracker[mac] = events if len(events) >= CHURN_THRESHOLD: with known_lock: infrastructure_ips.add(ip) logging.info(f"Churn suppression: {ip} (MAC:{mac}) cycled {len(events)}x in {CHURN_WINDOW_SEC//60}min — auto-added to infrastructure") return True return False # 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 # MAC address validation regex: exactly 12 hex digits (after normalization) _MAC_RE = re.compile(r'^[0-9A-F]{12}$') def _normalize_mac(mac: str) -> str: """Normalize MAC address: remove colons/hyphens, uppercase. Returns empty string if invalid.""" normalized = mac.replace(":", "").replace("-", "").upper() if not _MAC_RE.match(normalized): return "" return normalized def load_personal_macs_from_env() -> None: """Parse NET_ALERTER_PERSONAL_MACS env var and add to personal_devices set.""" macs_str = os.getenv("NET_ALERTER_PERSONAL_MACS", "") if not macs_str.strip(): return macs = [m.strip() for m in macs_str.split(",") if m.strip()] with personal_lock: for mac in macs: normalized = _normalize_mac(mac) if not normalized: # Skip invalid MACs logging.warning(f"Skipping invalid MAC from NET_ALERTER_PERSONAL_MACS: {mac}") continue personal_devices.add(normalized) logging.info(f"Added personal device MAC: {normalized}") 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", "") load_personal_macs_from_env() 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}") load_personal_macs_from_env() except Exception as e: logging.error(f"Failed to load .env: {e}") def is_personal_device(mac: str) -> bool: """ Check if MAC is a personal device (mobile OUI or manually configured). Returns True if device OUI is in MOBILE_DEVICE_OUIS OR mac is in personal_devices set. """ # Normalize MAC mac_normalized = _normalize_mac(mac) if not mac_normalized: # Skip invalid MACs return False oui = mac_normalized[:6] # Check personal_devices set first (manually configured) with personal_lock: if mac_normalized in personal_devices: return True # Check against MOBILE_DEVICE_OUIS if oui in MOBILE_DEVICE_OUIS: return True return False 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 _update_occupancy_state_unlocked() -> None: """ Internal: Update location occupancy state. Assumes known_lock AND occupancy_lock are already held by caller. """ global location_occupancy # Count active personal devices (in known_devices, not marked departing) active_personal_count = 0 for mac, dev in known_devices.items(): if not dev.get('departing', False) and is_personal_device(mac): active_personal_count += 1 # Determine new occupancy state new_occupancy = "OCCUPIED" if active_personal_count > 0 else "VACANT" # Fire alert on transitions (but not UNKNOWN → OCCUPIED on startup) if new_occupancy != location_occupancy: if not (location_occupancy == "UNKNOWN" and new_occupancy == "OCCUPIED"): msg = f"[NET] Location: {new_occupancy}" logging.info(msg) send_alert(msg) else: logging.debug(f"Suppressing startup occupancy alert: UNKNOWN → OCCUPIED") location_occupancy = new_occupancy logging.info(f"Location occupancy updated to {location_occupancy} ({active_personal_count} personal devices active)") def update_occupancy_state() -> None: """ Update location occupancy state based on presence of personal devices. Fires Matrix alerts on VACANT/OCCUPIED transitions (not on UNKNOWN transitions). Thread-safe: acquires locks in order: known_lock, then occupancy_lock. """ with known_lock: with occupancy_lock: _update_occupancy_state_unlocked() 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 # OUI-based auto-exclusion for network equipment oui = mac.replace(":", "").upper()[:6] if oui in NETWORK_EQUIPMENT_OUIS: infrastructure_ips.add(ip) logging.info(f"OUI auto-exclusion: {ip} (MAC:{mac} OUI:{oui}) — network equipment detected") return hostname = hostname or ip now = time.time() # Lock order: known_lock, then departure_lock, then occupancy_lock # Extract timers to cancel OUTSIDE all locks to avoid deadlock timers_to_cancel = [] with departure_lock: if mac in departure_timers: timers_to_cancel.append(departure_timers.pop(mac)) # Cancel all timers BEFORE acquiring any other locks for timer in timers_to_cancel: timer.cancel() with known_lock: with departure_lock: # Check if device marked departing (rapid flap recovery) if mac in known_devices and known_devices[mac].get('departing'): # Device returned during departure window — cancel timer, clear flag, no alert known_devices[mac]['departing'] = False with occupancy_lock: _update_occupancy_state_unlocked() last_departed_time.pop(mac, None) logging.debug(f"Device {mac} re-arrived during departure window — suppressed alert") return # Log if we cancelled a timer if timers_to_cancel: 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 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 with occupancy_lock: _update_occupancy_state_unlocked() return # Still inside known_lock, but not inside departure_lock now 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)") with occupancy_lock: _update_occupancy_state_unlocked() return # Not a re-arrival, update last_seen and don't alert (already known) with occupancy_lock: _update_occupancy_state_unlocked() 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) update_occupancy_state() 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 # Check if this is part of an interface flap (suppress simultaneous mass-departure) if _check_flap(mac): logging.debug(f"Flap suppression: ignoring departure for {mac}") return # Churn-based auto-suppression: if device cycles 3+ times in 30min, suppress if _check_churn(mac, dev['ip']): 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 # Extract timer to cancel outside of lock to avoid deadlock timer_to_cancel = None with departure_lock: # Cancel any existing timer for this MAC if mac in departure_timers: timer_to_cancel = departure_timers.pop(mac) # Cancel outside lock if timer_to_cancel: timer_to_cancel.cancel() with departure_lock: def send_departure_alert(): """Callback to send alert after debounce expires.""" # Lock order: known_lock, then departure_lock, then occupancy_lock 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) with departure_lock: last_departed_time[mac] = time.time() if mac in departure_timers: del departure_timers[mac] # Update occupancy state after device removal (still inside known_lock) with occupancy_lock: _update_occupancy_state_unlocked() 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) 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. Seeding steps: 1. Self IP (this device's interface IP) 2. Gateway IP (from 'ip route show default') 3. Broadcast IP (x.x.x.255) 4. Network equipment by OUI (Ubiquiti, Cisco, Aruba, TP-Link, Netgear, MikroTik, Ruckus) 5. Operator-provided IPs via NET_ALERTER_INFRA_IPS env var (comma-separated) NET_ALERTER_INFRA_IPS is a post-deploy configuration for operators to add infrastructure IPs known after deployment (access points, switches, etc). """ 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 # Auto-exclude known network equipment by OUI with known_lock: for mac, dev in list(known_devices.items()): oui = mac.replace(":", "").upper()[:6] if oui in NETWORK_EQUIPMENT_OUIS: infrastructure_ips.add(dev['ip']) logging.info(f"OUI auto-exclusion: {dev['ip']} (MAC:{mac} OUI:{oui}) — network equipment detected") logging.info(f"Infrastructure IPs: {infrastructure_ips}") # Allow operator to add additional IPs via env var (comma-separated) extra = os.environ.get("NET_ALERTER_INFRA_IPS", "") for ip in (x.strip() for x in extra.split(",") if x.strip()): infrastructure_ips.add(ip) logging.info(f"Seeded extra infrastructure IP {ip} from env") 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()