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.
This commit is contained in:
Cobra
2026-04-09 14:34:46 -04:00
parent 67cf907f74
commit 6447a462c2
2 changed files with 87 additions and 1 deletions
+65
View File
@@ -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!")