Add false positive suppression to net_alerter — fix for #467
- Infrastructure IPs (gateway, self, broadcast) silently ignored in on_arrival() and on_departure(); seeded at startup from 'ip route' - 15-minute departure debounce: alert only fires if device is still absent after 900s, eliminating ARP cache flush false positives - 5-minute re-arrival suppression: ARRIVED alert skipped when device returns within 5 min of a recorded departure (covers interface flap) - DHCP renewal dedup: known device re-announcing within 30 min skips ARRIVED alert - Tests added for all four suppression paths (9 total, all passing)
This commit is contained in:
Cobra
@@ -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!")
|
||||
|
||||
Reference in New Issue
Block a user