diff --git a/net_alerter/net_alerter.py b/net_alerter/net_alerter.py index c328fab..ef5a5d2 100644 --- a/net_alerter/net_alerter.py +++ b/net_alerter/net_alerter.py @@ -421,6 +421,19 @@ def on_arrival(mac: str, ip: str, hostname: str = "") -> None: 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 @@ -493,7 +506,8 @@ def on_departure(mac: str) -> None: return dev_copy = dev.copy() - known_devices.pop(mac) + # 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: @@ -503,6 +517,13 @@ def on_departure(mac: str) -> None: 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) diff --git a/net_alerter/test_net_alerter.py b/net_alerter/test_net_alerter.py index 8f6df10..ee88560 100644 --- a/net_alerter/test_net_alerter.py +++ b/net_alerter/test_net_alerter.py @@ -282,6 +282,68 @@ def test_re_arrival_cancels_departure_timer(): net_alerter.send_alert = original_send_alert +def test_rapid_flap_no_duplicate_arrived_alerts(): + """ + Test the specific bug: rapid RTM_NEWNEIGH events on wlan0 flap. + - Device departs (pop removed it from known_devices) + - Kernel rebuilds ARP cache → multiple RTM_NEWNEIGH events + - Each on_arrival() call finds MAC absent → treats as new → sends ARRIVED + Result: 5-10+ duplicate ARRIVED alerts + + With the fix: + - on_departure marks device with departing=True instead of popping + - on_arrival() checks for departing flag and silently cancels timer + - No duplicate ARRIVED alerts + """ + 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 and is already known (initial arrival) + now = time.time() + net_alerter.known_devices[mac] = { + "ip": ip, + "hostname": "testhost", + "vendor": "Test", + "first_seen": now, + "last_seen": now + } + + # Simulate initial arrival alert (already happened) + alert_calls.clear() # Reset to track only departure/re-arrival alerts + + # Device departs (wlan0 goes down) + net_alerter.on_departure(mac) + + # No alert yet (15 min debounce) + assert len([a for a in alert_calls if "DEPARTED" in a]) == 0 + + # wlan0 recovers, kernel rebuilds ARP cache → RTM_NEWNEIGH fires 5 times + # This should NOT generate 5 ARRIVED alerts + for i in range(5): + net_alerter.on_arrival(mac, ip, "testhost") + + # Count ARRIVED alerts + arrived_alerts = [a for a in alert_calls if "ARRIVED" in a] + assert len(arrived_alerts) == 0, \ + f"Rapid re-arrivals should not generate duplicate ARRIVED alerts, but got {len(arrived_alerts)}: {arrived_alerts}" + + finally: + # Clean up + for timer in net_alerter.departure_timers.values(): + timer.cancel() + net_alerter.departure_timers.clear() + 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") @@ -313,4 +375,7 @@ if __name__ == "__main__": test_re_arrival_cancels_departure_timer() print("✓ test_re_arrival_cancels_departure_timer") + test_rapid_flap_no_duplicate_arrived_alerts() + print("✓ test_rapid_flap_no_duplicate_arrived_alerts") + print("\nAll tests passed!")