Add interface flap detection and env var for infrastructure IPs in net_alerter

- Fix 1: Interface flap detection suppresses departure storms when wlan0 briefly loses association and triggers simultaneous RTM_DELNEIGH events for all cached neighbors
  - Tracks 3-second window of departure MACs
  - Locks recovery state for 60 seconds if 3+ departures occur in rapid succession
  - Prevents false positive alerts during interface recovery

- Fix 2: Additional infrastructure IPs via NET_ALERTER_INFRA_IPS env var
  - Allows operator to suppress alerts for devices like access points that cycle regularly
  - Comma-separated list of IP addresses to be seeded as infrastructure at startup
  - Example: NET_ALERTER_INFRA_IPS=10.0.0.0,10.0.0.0

Addresses DevTrack #467 (net_alerter false positive alerts)
This commit is contained in:
Cobra
2026-04-10 22:45:15 -04:00
parent 06c84e1800
commit 91a84b9d6f
+42
View File
@@ -37,6 +37,37 @@ 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
# Interface flap detection — suppresses departure storms
_flap_window = [] # [(timestamp, mac)]
_flap_lock = threading.Lock()
_interface_recovering = False
_recovery_timer = None
def _check_flap(mac: str) -> bool:
"""Return True if this departure is part of an interface flap (suppress it)."""
global _interface_recovering, _recovery_timer
now = time.time()
with _flap_lock:
_flap_window[:] = [(t, m) for t, m in _flap_window if now - t < 3.0]
_flap_window.append((now, mac))
if len(_flap_window) >= 3 and not _interface_recovering:
_interface_recovering = True
if _recovery_timer:
_recovery_timer.cancel()
_recovery_timer = threading.Timer(60.0, _clear_recovery)
_recovery_timer.daemon = True
_recovery_timer.start()
logging.info("Interface flap detected — suppressing departures for 60s")
return _interface_recovering
def _clear_recovery():
global _interface_recovering
_interface_recovering = False
logging.info("Interface flap recovery window ended")
# Top 200 common device OUI prefixes (MAC first 3 octets)
OUI_DICT = {
@@ -505,6 +536,11 @@ def on_departure(mac: str) -> None:
logging.debug(f"Ignoring departure for infrastructure IP {dev['ip']} (MAC:{mac})")
return
# Check if this is part of an interface flap (suppress simultaneous mass-departure)
if _check_flap(mac):
logging.debug(f"Flap suppression: ignoring departure for {mac}")
return
dev_copy = dev.copy()
# Mark departing instead of popping — allows re-arrival to detect and cancel timer
known_devices[mac]['departing'] = True
@@ -627,6 +663,12 @@ def seed_infrastructure_ips(iface: str) -> None:
logging.info(f"Infrastructure IPs: {infrastructure_ips}")
# Allow operator to add additional IPs via env var (comma-separated)
extra = os.environ.get("NET_ALERTER_INFRA_IPS", "")
for ip in (x.strip() for x in extra.split(",") if x.strip()):
infrastructure_ips.add(ip)
logging.info(f"Seeded extra infrastructure IP {ip} from env")
def dhcp_sniffer(iface: str) -> None:
"""