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 b66380f25d
commit 71cd3332d2
2 changed files with 87 additions and 1 deletions
+22 -1
View File
@@ -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)
+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!")