n0mad1k/bigbrother
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add false positive suppression to net_alerter — fix for #467

Browse Source 
- 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
2026-04-09 09:44:37 -04:00
parent 82228d79b3
commit a3c9353d3a
2 changed files with 315 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
net_alerter/net_alerter.py
+160 -12
View File
@@ -28,8 +28,12 @@ MATRIX_ACCESS_TOKEN = ""
MATRIX_ROOM_ID = ""
LOG_FILE = "/opt/net_alerter/net_alerter.log"
known_devices = {} # {mac: {ip, hostname, vendor, first_seen, last_seen}}
known_devices = {} # {mac: {ip, hostname, vendor, first_seen, last_seen, infrastructure (opt)}}
known_lock = threading.Lock()
infrastructure_ips = set() # IPs that should never trigger alerts (gateway, broadcast, self)
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
# Top 200 common device OUI prefixes (MAC first 3 octets)
@@ -308,13 +312,17 @@ def seed_from_arp_cache():
vendor = lookup_oui(mac)
with known_lock:
if mac not in known_devices:
known_devices[mac] = {
dev = {
'ip': ip,
'hostname': ip,
'vendor': vendor,
'first_seen': time.time(),
'last_seen': time.time()
}
# Mark infrastructure IPs so they never alert
if ip in infrastructure_ips:
dev['infrastructure'] = True
known_devices[mac] = dev
logging.info(f"Seeded {len(known_devices)} devices from ARP cache")
except Exception as e:
@@ -388,19 +396,61 @@ def format_departure(hostname: str, ip: str, vendor: str, mac: str, duration: st
def on_arrival(mac: str, ip: str, hostname: str = "") -> None:
"""Handle device arrival."""
# Infrastructure IPs never trigger alerts
if ip in infrastructure_ips:
logging.debug(f"Ignoring infrastructure IP {ip} (MAC:{mac})")
return
hostname = hostname or ip
now = time.time()
# Check re-arrival suppression BEFORE acquiring lock (avoid nested locks)
with departure_lock:
if mac in last_departed_time:
time_since_departure = now - last_departed_time[mac]
if time_since_departure < 300: # 5 min
logging.debug(f"Suppressing re-arrival alert for {mac} (departed {int(time_since_departure)}s ago)")
# Clean up departure timer if still pending
if mac in departure_timers:
departure_timers[mac].cancel()
del departure_timers[mac]
# Re-add to known_devices to track presence
with known_lock:
if mac not in known_devices:
vendor = lookup_oui(mac)
known_devices[mac] = {
'ip': ip,
'hostname': hostname,
'vendor': vendor,
'first_seen': now,
'last_seen': now
}
else:
known_devices[mac]['last_seen'] = now
return
with known_lock:
if mac in known_devices:
known_devices[mac]['last_seen'] = time.time()
# Device is already known
dev = known_devices[mac]
dev['last_seen'] = now
# DHCP renewal dedup: if last_seen < 30 min, skip alert
if now - dev['first_seen'] < 1800: # 30 min
logging.debug(f"Skipping DHCP renewal alert for {mac} (seen {int(now - dev['first_seen'])}s ago)")
return
# Not a re-arrival, update last_seen and don't alert (already known)
return
# New device
vendor = lookup_oui(mac)
hostname = hostname or ip
known_devices[mac] = {
'ip': ip,
'hostname': hostname,
'vendor': vendor,
'first_seen': time.time(),
'last_seen': time.time()
'first_seen': now,
'last_seen': now
}
msg = format_arrival(hostname, ip, vendor or "unknown", mac)
@@ -409,16 +459,49 @@ def on_arrival(mac: str, ip: str, hostname: str = "") -> None:
def on_departure(mac: str) -> None:
"""Handle device departure."""
"""
Handle device departure with 15-minute debounce.
Only sends alert if device still gone after 15 min.
"""
with known_lock:
if mac not in known_devices:
return
dev = known_devices.pop(mac)
duration = fmt_duration(time.time() - dev['first_seen'])
msg = format_departure(dev['hostname'], dev['ip'], dev['vendor'] or "unknown", mac, duration)
logging.info(msg)
send_alert(msg)
# Check if infrastructure IP (never alert)
dev = known_devices[mac]
if dev['ip'] in infrastructure_ips:
logging.debug(f"Ignoring departure for infrastructure IP {dev['ip']} (MAC:{mac})")
return
dev_copy = dev.copy()
known_devices.pop(mac)
# Start 15-minute debounce timer
with departure_lock:
# Cancel any existing timer for this MAC
if mac in departure_timers:
departure_timers[mac].cancel()
def send_departure_alert():
"""Callback to send alert after debounce expires."""
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)
send_alert(msg)
# Record departure time for re-arrival suppression
with departure_lock:
last_departed_time[mac] = time.time()
if mac in departure_timers:
del departure_timers[mac]
timer = threading.Timer(900.0, send_departure_alert) # 15 min = 900 sec
timer.daemon = True
timer.name = f'departure-debounce-{mac}'
departure_timers[mac] = timer
timer.start()
logging.debug(f"Departure debounce started for {mac}, will alert in 15 min if still gone")
def get_primary_interface() -> str:
@@ -441,6 +524,70 @@ def get_primary_interface() -> str:
return 'eth0'
def get_local_ip(iface: str) -> str:
"""Get the local IP address for the given interface."""
try:
import subprocess
result = subprocess.run(['ip', 'addr', 'show', iface], capture_output=True, text=True, timeout=5)
if result.returncode == 0:
for line in result.stdout.split('\n'):
if 'inet ' in line:
parts = line.strip().split()
if len(parts) >= 2:
ip_with_mask = parts[1]
ip = ip_with_mask.split('/')[0]
return ip
except Exception as e:
logging.warning(f"Failed to get local IP for {iface}: {e}")
return ""
def seed_infrastructure_ips(iface: str) -> None:
"""
Seed infrastructure IPs that should never trigger alerts.
Reads gateway from 'ip route show default', adds broadcast and self IP.
"""
global infrastructure_ips
infrastructure_ips.clear()
# Get self IP
self_ip = get_local_ip(iface)
if self_ip:
infrastructure_ips.add(self_ip)
logging.info(f"Seeded self IP {self_ip} as infrastructure")
# Read gateway from 'ip route show default'
try:
import subprocess
result = subprocess.run(['ip', 'route', 'show', 'default'], capture_output=True, text=True, timeout=5)
if result.returncode == 0:
# Parse: default via 10.0.0.0 dev eth0 proto dhcp metric 100
for line in result.stdout.strip().split('\n'):
if 'via' in line:
parts = line.split()
for i, part in enumerate(parts):
if part == 'via' and i + 1 < len(parts):
gateway_ip = parts[i + 1]
infrastructure_ips.add(gateway_ip)
logging.info(f"Seeded gateway {gateway_ip} as infrastructure")
break
except Exception as e:
logging.warning(f"Failed to read gateway: {e}")
# Add broadcast (x.x.x.255 for most networks)
if self_ip:
try:
parts = self_ip.rsplit('.', 1)
if len(parts) == 2:
broadcast_ip = parts[0] + ".255"
infrastructure_ips.add(broadcast_ip)
logging.info(f"Seeded broadcast {broadcast_ip} as infrastructure")
except Exception:
pass
logging.info(f"Infrastructure IPs: {infrastructure_ips}")
def dhcp_sniffer(iface: str) -> None:
"""
DHCP sniffer thread.
@@ -652,6 +799,7 @@ def main() -> None:
iface = get_primary_interface()
logging.info(f"Net alerter starting on interface {iface}")
seed_infrastructure_ips(iface)
seed_from_arp_cache()
# Start monitor threads
net_alerter/test_net_alerter.py
+155 -1
View File
@@ -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!")