diff --git a/net_alerter/net_alerter.py b/net_alerter/net_alerter.py index b786cc2..cc5da06 100644 --- a/net_alerter/net_alerter.py +++ b/net_alerter/net_alerter.py @@ -28,8 +28,12 @@ MATRIX_ACCESS_TOKEN = "" MATRIX_ROOM_ID = "" LOG_FILE = "/opt/net_alerter/net_alerter.log" -known_devices = {} # {mac: {ip, hostname, vendor, first_seen, last_seen}} +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) @@ -308,13 +312,17 @@ def seed_from_arp_cache(): vendor = lookup_oui(mac) with known_lock: if mac not in known_devices: - known_devices[mac] = { + 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: @@ -388,19 +396,61 @@ def format_departure(hostname: str, ip: str, vendor: str, mac: str, duration: st 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() + + # Check re-arrival suppression BEFORE acquiring lock (avoid nested locks) + with departure_lock: + 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)") + # Clean up departure timer if still pending + if mac in departure_timers: + departure_timers[mac].cancel() + del departure_timers[mac] + # 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: - known_devices[mac]['last_seen'] = time.time() + # 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) - hostname = hostname or ip known_devices[mac] = { 'ip': ip, 'hostname': hostname, 'vendor': vendor, - 'first_seen': time.time(), - 'last_seen': time.time() + 'first_seen': now, + 'last_seen': now } msg = format_arrival(hostname, ip, vendor or "unknown", mac) @@ -409,16 +459,49 @@ def on_arrival(mac: str, ip: str, hostname: str = "") -> None: def on_departure(mac: str) -> None: - """Handle device departure.""" + """ + 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 - dev = known_devices.pop(mac) - duration = fmt_duration(time.time() - dev['first_seen']) - msg = format_departure(dev['hostname'], dev['ip'], dev['vendor'] or "unknown", mac, duration) - logging.info(msg) - send_alert(msg) + # 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() + known_devices.pop(mac) + + # 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.""" + 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: @@ -441,6 +524,70 @@ def get_primary_interface() -> str: 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. @@ -652,6 +799,7 @@ def main() -> None: iface = get_primary_interface() logging.info(f"Net alerter starting on interface {iface}") + seed_infrastructure_ips(iface) seed_from_arp_cache() # Start monitor threads diff --git a/net_alerter/test_net_alerter.py b/net_alerter/test_net_alerter.py index d4bb43b..4bdc224 100644 --- a/net_alerter/test_net_alerter.py +++ b/net_alerter/test_net_alerter.py @@ -1,10 +1,13 @@ #!/usr/bin/env python3 """ -Test suite for net_alerter.py — alert formatting fixes. +Test suite for net_alerter.py — alert formatting fixes and false positive suppression. """ import sys +import threading +import time from pathlib import Path +from unittest.mock import patch, MagicMock # Add parent dir to path so we can import net_alerter sys.path.insert(0, str(Path(__file__).parent)) @@ -79,6 +82,145 @@ def test_departure_format_dedup(): assert msg.count("10.0.0.0") == 1 +def test_infrastructure_ips_never_alert(): + """Test that infrastructure IPs (gateway, broadcast, self) don't trigger alerts.""" + # Reset known_devices + net_alerter.known_devices.clear() + + # Mock send_alert to track calls + alert_calls = [] + original_send_alert = net_alerter.send_alert + net_alerter.send_alert = lambda msg: alert_calls.append(msg) + + try: + # Seed infrastructure IPs with infrastructure flag + net_alerter.known_devices = { + "ff:ff:ff:ff:ff:ff": {"ip": "10.0.0.0", "hostname": "255", "vendor": "Broadcast", "first_seen": time.time(), "last_seen": time.time(), "infrastructure": True}, + "00:00:00:00:00:00": {"ip": "10.0.0.0", "hostname": "1", "vendor": "Gateway", "first_seen": time.time(), "last_seen": time.time(), "infrastructure": True}, + "12:34:56:78:90:ab": {"ip": "10.0.0.0", "hostname": "10", "vendor": "Self", "first_seen": time.time(), "last_seen": time.time(), "infrastructure": True}, + } + + # Try to trigger departure for infrastructure IPs - should be suppressed + net_alerter.on_departure("ff:ff:ff:ff:ff:ff") + net_alerter.on_departure("00:00:00:00:00:00") + net_alerter.on_departure("12:34:56:78:90:ab") + + # No alerts should be sent + assert len(alert_calls) == 0, f"Infrastructure IPs should not trigger alerts, but got: {alert_calls}" + finally: + net_alerter.send_alert = original_send_alert + + +def test_departure_debounce_15min(): + """Test that departure alerts are debounced for 15 minutes.""" + net_alerter.known_devices.clear() + net_alerter.departure_timers.clear() + + alert_calls = [] + original_send_alert = net_alerter.send_alert + net_alerter.send_alert = lambda msg: alert_calls.append(msg) + + try: + mac = "aa:bb:cc:dd:ee:ff" + + # Add device to known_devices + net_alerter.known_devices[mac] = { + "ip": "10.0.0.0", + "hostname": "testhost", + "vendor": "Test", + "first_seen": time.time(), + "last_seen": time.time() + } + + # Trigger departure + net_alerter.on_departure(mac) + + # Should NOT have sent alert yet (still in debounce period) + assert len(alert_calls) == 0, f"Departure should be debounced, but got alert: {alert_calls}" + + # Timer should be created + assert mac in net_alerter.departure_timers, "Departure timer should be created" + + # Cancel the timer (cleanup) + net_alerter.departure_timers[mac].cancel() + del net_alerter.departure_timers[mac] + finally: + net_alerter.send_alert = original_send_alert + + +def test_re_arrival_within_5min_suppresses_alert(): + """Test that device re-arrival within 5 min of departure suppresses ARRIVED alert.""" + net_alerter.known_devices.clear() + net_alerter.departure_timers.clear() + net_alerter.last_departed_time.clear() + + alert_calls = [] + original_send_alert = net_alerter.send_alert + net_alerter.send_alert = lambda msg: alert_calls.append(msg) + + try: + mac = "aa:bb:cc:dd:ee:ff" + ip = "10.0.0.0" + + # Device exists + net_alerter.known_devices[mac] = { + "ip": ip, + "hostname": "testhost", + "vendor": "Test", + "first_seen": time.time(), + "last_seen": time.time() + } + + # Trigger departure (but don't let the timer fire) + net_alerter.on_departure(mac) + + # Record departure time (simulating alert being sent after debounce) + net_alerter.last_departed_time[mac] = time.time() + + # Now device re-arrives within 5 minutes + net_alerter.on_arrival(mac, ip, "testhost") + + # Should NOT have sent ARRIVED alert (suppressed) + assert len(alert_calls) == 0, f"Re-arrival within 5 min should be suppressed, but got: {alert_calls}" + + # Cleanup + if mac in net_alerter.departure_timers: + net_alerter.departure_timers[mac].cancel() + finally: + net_alerter.send_alert = original_send_alert + + +def test_dhcp_renewal_dedup_30min(): + """Test that DHCP renewal from known device (< 30 min old) doesn't send ARRIVED alert.""" + net_alerter.known_devices.clear() + + alert_calls = [] + original_send_alert = net_alerter.send_alert + net_alerter.send_alert = lambda msg: alert_calls.append(msg) + + try: + mac = "aa:bb:cc:dd:ee:ff" + ip = "10.0.0.0" + + # Device already known and seen < 30 min ago + now = time.time() + net_alerter.known_devices[mac] = { + "ip": ip, + "hostname": "testhost", + "vendor": "Test", + "first_seen": now - 600, # 10 min ago + "last_seen": now - 100, # 100 sec ago + } + + # DHCP renewal (arrival) + net_alerter.on_arrival(mac, ip, "testhost") + + # Should NOT have sent alert (last_seen within 30 min) + assert len(alert_calls) == 0, f"DHCP renewal < 30 min should be deduped, but got: {alert_calls}" + finally: + net_alerter.send_alert = original_send_alert + + if __name__ == "__main__": test_hostname_dedup_when_hostname_equals_ip() print("✓ test_hostname_dedup_when_hostname_equals_ip") @@ -95,4 +237,16 @@ if __name__ == "__main__": test_departure_format_dedup() print("✓ test_departure_format_dedup") + test_infrastructure_ips_never_alert() + print("✓ test_infrastructure_ips_never_alert") + + test_departure_debounce_15min() + print("✓ test_departure_debounce_15min") + + test_re_arrival_within_5min_suppresses_alert() + print("✓ test_re_arrival_within_5min_suppresses_alert") + + test_dhcp_renewal_dedup_30min() + print("✓ test_dhcp_renewal_dedup_30min") + print("\nAll tests passed!")