n0mad1k/bigbrother
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
814085c0b34b0604991504f369528d933565e14b
bigbrother/net_alerter/net_alerter.py
T
Cobra 814085c0b3 Add iOS departure detection via passive ARP/mDNS capture and watchdog 
Implement passive observation tracking for iOS devices:
- Add last_seen dict and lock for tracking passive frame observations
- Parse ARP frames (ethertype 0x0806) for MAC observation
- Parse mDNS frames (UDP port 5353) for device name auto-discovery
- Rename parse_dhcp to parse_frame to reflect broader scope
- Add personal_names set for device name matching
- Parse NET_ALERTER_PERSONAL_NAMES env var (comma-separated device names)
- Add mDNS DNS message parsing to extract PTR/SRV/A record names
- Auto-add MACs to personal_devices when mDNS name matches personal_names
- Call _update_last_seen on all frame types (ARP, mDNS, DHCP)
- Add personal_watchdog thread to fire on_departure after WATCHDOG_TIMEOUT_SEC
- Add NET_ALERTER_WATCHDOG_TIMEOUT env config (default 900s)
- Fix load_personal_macs_from_env logging to show counts instead of MACs

Changes are minimal and focused:
- No socket changes (existing AF_PACKET socket already sees all frames)
- No new dependencies (stdlib only)
- Deadlock prevention (watchdog does not hold locks when calling on_departure)
- Backward compatible (personal_names optional, watchdog is passive)
2026-04-13 20:27:00 -04:00

1365 lines
46 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""
net_alerter.py — Net alerter persistent daemon.
Monitors device presence via DHCP sniffing + Netlink neighbor events.
Zero active probing. No traffic generated.
Three concurrent threads:
1. DHCP sniffer — AF_PACKET raw socket, captures DHCP traffic passively
2. RTM_NEWNEIGH watcher — Netlink socket, kernel pushes new ARP neighbor events
3. RTM_DELNEIGH watcher — Netlink socket, kernel pushes deleted ARP neighbor events
"""
import json
import logging
import os
import re
import socket
import struct
import sys
import threading
import time
import urllib.parse
import urllib.request
import sqlite3
from pathlib import Path
# --- Config (from .env or env vars) ---
LOG_FILE = "/opt/net_alerter/net_alerter.log"
OUI_DB_PATH = "/opt/net_alerter/oui.db"
ENV_PATH = Path(os.getenv("NET_ALERTER_ENV", "/opt/net_alerter/.env"))
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
# Passive observation tracking for iOS departure detection
last_seen = {} # {mac: float} — updated by passive ARP/mDNS capture
last_seen_lock = threading.Lock()
# Interface flap detection — suppresses departure storms
_flap_window = [] # [(timestamp, mac)]
_flap_lock = threading.Lock()
_interface_recovering = False
_recovery_timer = None
# Network equipment OUI prefixes (first 3 octets, uppercase no colons)
# Auto-exclusion for known infrastructure devices
NETWORK_EQUIPMENT_OUIS = {
# Ubiquiti
"0418D6", "24A43C", "44D9E7", "68729C", "788A20", "802AA8",
"9C05D6", "DC9FDB", "E063DA", "F09FC2", "6C63F8", "B4FBE4",
"00156D", "002722", "04186F",
# Cisco
"000C29", "001A2F", "001BB1", "001E13", "001F26", "002155",
"0060B0", "00902B", "008024", "0090AB", "00E014", "001011",
# Aruba
"000B86", "00FC8B", "040E3C", "1C28AF", "40E3D6", "6C29D1",
"84D47E", "884774", "94B40F", "AC3744",
# TP-Link
"000AEB", "001777", "1C61B4", "5C63BF", "6046D9", "A84E3F",
"B4B024", "C46AB7", "EC086B", "F81A67",
# Netgear
"001B2F", "00146C", "002196", "00A0CC", "081102", "1C1E29",
"206A8A", "30469A", "4F2AFF", "6CB0CE", "8C3BAD",
# MikroTik
"4C5E0C", "B8069F", "C4AD34", "CC2DE0", "D4CA6D", "E48D8C",
# Ruckus
"000C67", "00249B", "04BD88", "14778B", "2C5D93", "34A84E",
}
# Churn tracking — auto-suppresses devices that cycle 3+ times in 30min
_churn_tracker = {} # {mac: [timestamps of departure events]}
_churn_lock = threading.Lock()
CHURN_THRESHOLD = 3 # cycles in window before suppression
CHURN_WINDOW_SEC = 1800 # 30 minutes
WATCHDOG_TIMEOUT_SEC = int(os.getenv("NET_ALERTER_WATCHDOG_TIMEOUT", "900"))
# Mobile device OUI prefixes for personal device detection
MOBILE_DEVICE_OUIS = {
# Apple
"28A02E", "A4C3F0", "00A0C6", "38C086", "5C4950", "7061D0", "88A29E",
"A81D6A", "A8BBCF", "AC3744", "B0EE7B", "B82DFE", "BC9674", "C8DFC0",
"D4996F", "D8BB85", "E0D55E", "F4CE46", "F8FF9B", "FCFC48",
# Google Pixel
"28876D", "34E5FB", "4C80F3", "70EBF2", "AC16E5", "D0A05F", "E4C547",
# Samsung
"00166B", "001843", "001D67", "001FAF", "0060B0", "084ECE", "0825F3",
"088038", "0A49EF", "0C1432", "0EBB1B", "1851BF", "205C85", "20C18B",
"240A64", "2C5A7B", "32A639", "360DFE", "3A590D", "3CC06C", "441735",
"44B16C", "48D76E", "58D55C", "5C2EE4", "68C903", "6AC4C6", "6EAA0A",
"7045AC", "7499EA", "74D4DD", "781F02", "789098", "7C59B9", "7E0A6D",
# OnePlus
"00166B", "001BF3", "001C47", "002000", "0026F2", "3C28CB", "38DB23",
"42BAFC", "5C55F9", "681E7D", "A8178E", "AA6B1F", "BC1E1A", "EA96D7",
# Xiaomi
"0022AA", "002255", "003677", "3418C2", "48EE0C", "54EF25", "7418DB",
"7C8B5C", "7E4A5D", "868A5A", "8A6FB8", "8C8E14", "8EBE59", "A0871B",
"B8328D", "B832D6", "BA5294", "BC7C47", "C8816D", "E4AF17", "E899C3",
# Huawei
"00E04C", "001097", "001093", "001A2B", "001D3C", "001E37", "001F3E",
"001ECE", "002074", "00207B", "0020E2", "0020F1", "0020F5", "002268",
"002269", "00226A",
# LG
"001E4B", "001E8D", "00218B", "002215", "0022FB", "0023B2", "0023F8",
"002655", "00266C", "00AB8B", "30F9ED", "D8D0DF",
# HTC
"001072", "001058", "0010C6", "001153", "001167", "0011BD", "00121F",
"00121C", "001212", "001236", "00124B", "00131D", "0013D4", "0013E8",
# Motorola
"001555", "001620", "001645", "001693", "0016B8", "001730", "0017F6",
"001822", "001888", "0018F8", "001913", "001985", "001A45", "001A92",
"001AFC",
}
# Personal device tracking
personal_devices = set() # {mac: ...} of detected personal device MACs
personal_names = set() # {name: ...} device names for auto-discovery via mDNS
personal_lock = threading.Lock()
location_occupancy = "UNKNOWN" # Current location state: VACANT, OCCUPIED, UNKNOWN
occupancy_lock = threading.Lock()
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")
def _check_churn(mac: str, ip: str) -> bool:
"""Return True and suppress if this device churns too frequently."""
now = time.time()
with _churn_lock:
events = _churn_tracker.get(mac, [])
events = [t for t in events if now - t < CHURN_WINDOW_SEC]
events.append(now)
_churn_tracker[mac] = events
if len(events) >= CHURN_THRESHOLD:
infrastructure_ips.add(ip)
logging.info(f"Churn suppression: {ip} (MAC:{mac}) cycled {len(events)}x in {CHURN_WINDOW_SEC//60}min — auto-added to infrastructure")
return True
return False
# Top 200 common device OUI prefixes (MAC first 3 octets)
OUI_DICT = {
"88A29E": "Apple",
"001A7D": "Apple",
"185F3F": "Apple",
"A4C3F0": "Apple",
"0899D8": "Apple",
"004096": "Apple",
"00219B": "Apple",
"0021E9": "Apple",
"005973": "Apple",
"006377": "Apple",
"0064B9": "Apple",
"0084F3": "Apple",
"00A04D": "Apple",
"00D04B": "Apple",
"28879F": "Google",
"2887BA": "Google",
"5427EB": "Google",
"542758": "Google",
"341513": "Amazon",
"0C47C2": "Amazon",
"B827EB": "Raspberry Pi",
"2C56DC": "Raspberry Pi",
"E45F01": "Raspberry Pi",
"000D82": "Intel",
"001025": "Intel",
"001ABA": "Intel",
"001F3B": "Intel",
"001F3C": "Intel",
"001BA9": "Intel",
"001BFB": "Intel",
"001D7A": "Intel",
"001DB8": "Intel",
"001E67": "Intel",
"001F29": "Intel",
"F44DA2": "Intel",
"B03C1C": "Intel",
"00166B": "Qualcomm",
"001BF3": "Qualcomm",
"001C47": "Qualcomm",
"002000": "Qualcomm",
"0026F2": "Qualcomm",
"0026F3": "Qualcomm",
"0026F4": "Qualcomm",
"002618": "Qualcomm",
"FFFFFF": "Espressif",
"687DA8": "Espressif",
"A01D48": "Espressif",
"30AEA4": "Espressif",
"B4E63B": "Espressif",
"80E386": "TP-Link",
"8CBF26": "TP-Link",
"90843F": "TP-Link",
"A838B4": "TP-Link",
"C81F66": "TP-Link",
"DCFEEE": "TP-Link",
"F81A67": "TP-Link",
"689B5A": "Netgear",
"6C46CB": "Netgear",
"78A21B": "Netgear",
"94B3D5": "Netgear",
"B0EE7B": "Netgear",
"B4FBE4": "Netgear",
"C8D7B4": "Netgear",
"D8BB85": "Netgear",
"F8AB05": "Netgear",
"001D2F": "Cisco",
"001930": "Cisco",
"001A2F": "Cisco",
"001C0E": "Cisco",
"001CED": "Cisco",
"001EBD": "Cisco",
"001F6C": "Cisco",
"0021A0": "Cisco",
"002678": "Cisco",
"0026CB": "Cisco",
"00266E": "Cisco",
"00269F": "Cisco",
"0026A6": "Cisco",
"002700": "Cisco",
"0026FA": "Cisco",
"00267B": "Cisco",
"0C1234": "Cisco",
"001BD7": "Dell",
"001BDB": "Dell",
"001A6B": "Dell",
"003067": "Dell",
"00188B": "Dell",
"001B21": "Dell",
"001E67": "Dell",
"0026F7": "Dell",
"00242A": "Dell",
"001C23": "HP",
"001E0B": "HP",
"001EBB": "HP",
"001F2F": "HP",
"001BFC": "HP",
"002219": "HP",
"0022A6": "HP",
"002264": "HP",
"0024A9": "HP",
"0026B9": "HP",
"0026DF": "HP",
"001A6C": "Lenovo",
"001DFB": "Lenovo",
"001F3A": "Lenovo",
"001A73": "Lenovo",
"00238B": "Lenovo",
"002264": "Lenovo",
"00B0D0": "Lenovo",
"9C2EBF": "Lenovo",
"F0DE8D": "Lenovo",
"00E04C": "Huawei",
"001097": "Huawei",
"001093": "Huawei",
"001A2B": "Huawei",
"001D3C": "Huawei",
"001E37": "Huawei",
"001F3E": "Huawei",
"001ECE": "Huawei",
"002074": "Huawei",
"00207B": "Huawei",
"0020E2": "Huawei",
"0020F1": "Huawei",
"0020F5": "Huawei",
"002268": "Huawei",
"002269": "Huawei",
"00226A": "Huawei",
"0022AA": "Xiaomi",
"002255": "Xiaomi",
"003677": "Xiaomi",
"3418C2": "Xiaomi",
"50EB71": "Xiaomi",
"5CF4D4": "Xiaomi",
"78111D": "Xiaomi",
"78E1D3": "Xiaomi",
"A4ABDA": "Xiaomi",
"E4F47A": "Xiaomi",
"F45EAB": "Xiaomi",
"A4D18D": "OnePlus",
"50F5DA": "OnePlus",
"8C03BE": "OnePlus",
"002261": "LG",
"002265": "LG",
"0022D2": "LG",
"0022D3": "LG",
"000426": "LG",
"001854": "LG",
"001E8F": "LG",
"001F5B": "LG",
"002038": "LG",
"0020A6": "LG",
"0020F8": "LG",
"002220": "LG",
"00222D": "LG",
"0027E4": "LG",
"00242C": "LG",
"002531": "LG",
"001054": "Sony",
"001458": "Sony",
"0014EA": "Sony",
"001525": "Sony",
"00188A": "Sony",
"001B44": "Sony",
"00215E": "Sony",
"002178": "Sony",
"0021C5": "Sony",
"002333": "Sony",
"002481": "Sony",
"0026FF": "Sony",
"002702": "Nintendo",
"0008FE": "Nintendo",
"0013F7": "Nintendo",
"001BDB": "Nintendo",
"001EEE": "Nintendo",
"001FEE": "Nintendo",
"0020F8": "Nintendo",
"002095": "Nintendo",
"0020FA": "Nintendo",
"00227B": "Nintendo",
}
def setup_logging():
"""Setup logging to file + stdout."""
formatter = logging.Formatter(
fmt='%(asctime)s [%(levelname)s] %(message)s',
datefmt='%Y-%m-%dT%H:%M:%S'
)
# File handler
Path(LOG_FILE).parent.mkdir(parents=True, exist_ok=True)
file_handler = logging.FileHandler(LOG_FILE)
file_handler.setFormatter(formatter)
# Stdout handler
stdout_handler = logging.StreamHandler(sys.stdout)
stdout_handler.setFormatter(formatter)
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
logger.addHandler(file_handler)
logger.addHandler(stdout_handler)
return logger
# MAC address validation regex: exactly 12 hex digits (after normalization)
_MAC_RE = re.compile(r'^[0-9A-F]{12}$')
def _normalize_mac(mac: str) -> str:
"""Normalize MAC address: remove colons/hyphens, uppercase. Returns empty string if invalid."""
normalized = mac.replace(":", "").replace("-", "").upper()
if not _MAC_RE.match(normalized):
return ""
return normalized
def load_personal_macs_from_env() -> None:
"""Parse NET_ALERTER_PERSONAL_MACS and NET_ALERTER_PERSONAL_NAMES env vars.
Adds MACs from NET_ALERTER_PERSONAL_MACS to personal_devices set.
Adds device names from NET_ALERTER_PERSONAL_NAMES to personal_names set for mDNS discovery.
"""
macs_str = os.getenv("NET_ALERTER_PERSONAL_MACS", "")
names_str = os.getenv("NET_ALERTER_PERSONAL_NAMES", "")
with personal_lock:
# Load explicit MACs
if macs_str.strip():
macs = [m.strip() for m in macs_str.split(",") if m.strip()]
for mac in macs:
normalized = _normalize_mac(mac)
if not normalized: # Skip invalid MACs
logging.warning(f"Skipping invalid MAC from NET_ALERTER_PERSONAL_MACS: {mac}")
continue
personal_devices.add(normalized)
# Load device names for mDNS auto-discovery
if names_str.strip():
names = [n.strip() for n in names_str.split(",") if n.strip()]
for name in names:
personal_names.add(name)
# Log summary
if personal_devices or personal_names:
logging.info(f"Loaded {len(personal_devices)} personal device MAC(s) and {len(personal_names)} personal name(s)")
def _read_env_value(key: str) -> str:
"""Read a single environment variable from .env file or system env."""
if ENV_PATH.exists():
try:
for line in ENV_PATH.read_text().splitlines():
line = line.strip()
if not line or line.startswith("#"):
continue
if "=" not in line:
continue
k, v = line.split("=", 1)
if k == key:
return v
except Exception as e:
logging.error(f"Failed to read {key} from .env: {e}")
return os.getenv(key, "")
def load_env():
"""Parse .env file manually (key=value), no dependency required.
Note: Matrix credentials (MATRIX_HOMESERVER, MATRIX_ACCESS_TOKEN, MATRIX_ROOM_ID)
are NOT cached in globals. They are read at call time in send_alert() to minimize
credential lifetime in process memory.
"""
if not ENV_PATH.exists():
logging.warning(f".env not found at {ENV_PATH}, using env vars only")
load_personal_macs_from_env()
return
try:
for line in ENV_PATH.read_text().splitlines():
line = line.strip()
if not line or line.startswith("#"):
continue
if "=" not in line:
continue
# Only load non-credential config here
# Credentials are read at call time to minimize exposure
logging.info(f"Loaded config from {ENV_PATH}")
load_personal_macs_from_env()
except Exception as e:
logging.error(f"Failed to load .env: {e}")
def is_personal_device(mac: str) -> bool:
"""
Check if MAC is a personal device (mobile OUI or manually configured).
Returns True if device OUI is in MOBILE_DEVICE_OUIS OR mac is in personal_devices set.
"""
# Normalize MAC
mac_normalized = _normalize_mac(mac)
if not mac_normalized: # Skip invalid MACs
return False
oui = mac_normalized[:6]
# Check personal_devices set first (manually configured)
with personal_lock:
if mac_normalized in personal_devices:
return True
# Check against MOBILE_DEVICE_OUIS
if oui in MOBILE_DEVICE_OUIS:
return True
return False
def lookup_oui(mac: str) -> str:
"""
Look up OUI vendor from MAC address (first 3 octets).
Tries SQLite database first, falls back to inline dictionary.
Returns vendor name or 'unknown' if not found.
"""
if not mac or mac == "00:00:00:00:00:00":
return "unknown"
# Try database first
oui_prefix = mac[:8].lower()
try:
conn = sqlite3.connect(OUI_DB_PATH, timeout=1)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
result = cursor.execute("SELECT vendor FROM oui WHERE prefix = ?", (oui_prefix,)).fetchone()
conn.close()
if result:
return result['vendor']
except Exception:
# Log at DEBUG only, no output noise
logging.debug(f"OUI database lookup failed for {oui_prefix}", exc_info=False)
# Fall back to inline dictionary
oui_prefix_upper = oui_prefix.replace(":", "").upper()
return OUI_DICT.get(oui_prefix_upper, "unknown")
def seed_from_arp_cache():
"""Read /proc/net/arp, populate known_devices without alerting."""
global known_devices
try:
lines = Path("/proc/net/arp").read_text().splitlines()
for line in lines:
if line.startswith("IP"):
continue
fields = line.split()
if len(fields) < 4:
continue
ip = fields[0]
mac = fields[3].lower()
flags = fields[2]
if mac == "00:00:00:00:00:00" or flags == "0x0":
continue
vendor = lookup_oui(mac)
with known_lock:
if mac not in known_devices:
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:
logging.error(f"Failed to seed from ARP cache: {e}")
def send_alert(msg: str) -> None:
"""Send alert to Matrix room.
Credentials are read at call time from .env/env vars to minimize
their lifetime in process memory.
"""
token = _read_env_value("MATRIX_ACCESS_TOKEN")
if not token:
logging.warning("No MATRIX_ACCESS_TOKEN set, skipping alert")
return
homeserver = _read_env_value("MATRIX_HOMESERVER") or "https://m.example.org"
room_id = _read_env_value("MATRIX_ROOM_ID")
if not room_id:
logging.warning("No MATRIX_ROOM_ID set, skipping alert")
return
url = (
f"{homeserver}/_matrix/client/v3/rooms/"
f"{urllib.parse.quote(room_id, safe='')}/send/m.room.message"
)
payload = json.dumps({"msgtype": "m.text", "body": msg}).encode()
req = urllib.request.Request(
url,
data=payload,
headers={
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
"User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36",
},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=10) as r:
pass
logging.debug(f"Alert sent to Matrix")
except urllib.error.HTTPError as e:
logging.error(f"Matrix send failed with HTTP {e.code}")
except Exception as e:
logging.error(f"Matrix send exception: {e}")
def fmt_duration(secs: float) -> str:
"""Format duration in seconds to human-readable format."""
if secs < 60:
return f"{int(secs)}s"
elif secs < 3600:
m = int(secs // 60)
s = int(secs % 60)
return f"{m}m {s}s" if s else f"{m}m"
else:
h = int(secs // 3600)
m = int((secs % 3600) // 60)
return f"{h}h {m}m" if m else f"{h}h"
def format_arrival(hostname: str, ip: str, vendor: str, mac: str) -> str:
"""Format arrival alert message. Deduplicate IP if hostname equals IP."""
if hostname == ip:
# Hostname resolution failed, don't repeat IP
return f"[NET] ARRIVED: {ip} [{vendor}] MAC:{mac}"
else:
# Hostname resolved successfully, show both
return f"[NET] ARRIVED: {hostname} ({ip}) [{vendor}] MAC:{mac}"
def format_departure(hostname: str, ip: str, vendor: str, mac: str, duration: str) -> str:
"""Format departure alert message. Deduplicate IP if hostname equals IP."""
if hostname == ip:
# Hostname resolution failed, don't repeat IP
return f"[NET] DEPARTED: {ip} [{vendor}] MAC:{mac} — present {duration}"
else:
# Hostname resolved successfully, show both
return f"[NET] DEPARTED: {hostname} ({ip}) [{vendor}] MAC:{mac} — present {duration}"
def _update_occupancy_state_unlocked() -> None:
"""
Internal: Update location occupancy state.
Assumes known_lock AND occupancy_lock are already held by caller.
"""
global location_occupancy
# Count active personal devices (in known_devices, not marked departing)
active_personal_count = 0
for mac, dev in known_devices.items():
if not dev.get('departing', False) and is_personal_device(mac):
active_personal_count += 1
# Determine new occupancy state
new_occupancy = "OCCUPIED" if active_personal_count > 0 else "VACANT"
# Fire alert on transitions (but not UNKNOWN → OCCUPIED on startup)
if new_occupancy != location_occupancy:
if not (location_occupancy == "UNKNOWN" and new_occupancy == "OCCUPIED"):
msg = f"[NET] Location: {new_occupancy}"
logging.info(msg)
send_alert(msg)
else:
logging.debug(f"Suppressing startup occupancy alert: UNKNOWN → OCCUPIED")
location_occupancy = new_occupancy
logging.info(f"Location occupancy updated to {location_occupancy} ({active_personal_count} personal devices active)")
def update_occupancy_state() -> None:
"""
Update location occupancy state based on presence of personal devices.
Fires Matrix alerts on VACANT/OCCUPIED transitions (not on UNKNOWN transitions).
Thread-safe: acquires locks in order: known_lock, then occupancy_lock.
"""
with known_lock:
with occupancy_lock:
_update_occupancy_state_unlocked()
def on_arrival(mac: str, ip: str, hostname: str = "") -> None:
"""Handle device arrival."""
# Update passive observation tracking
_update_last_seen(mac)
# Infrastructure IPs never trigger alerts
if ip in infrastructure_ips:
logging.debug(f"Ignoring infrastructure IP {ip} (MAC:{mac})")
return
# OUI-based auto-exclusion for network equipment
oui = mac.replace(":", "").upper()[:6]
if oui in NETWORK_EQUIPMENT_OUIS:
infrastructure_ips.add(ip)
logging.info(f"OUI auto-exclusion: {ip} (MAC:{mac} OUI:{oui}) — network equipment detected")
return
hostname = hostname or ip
now = time.time()
# Lock order: known_lock, then departure_lock, then occupancy_lock
# Extract timers to cancel OUTSIDE all locks to avoid deadlock
timers_to_cancel = []
with departure_lock:
if mac in departure_timers:
timers_to_cancel.append(departure_timers.pop(mac))
# Cancel all timers BEFORE acquiring any other locks
for timer in timers_to_cancel:
timer.cancel()
with known_lock:
with departure_lock:
# Check if device marked departing (rapid flap recovery)
if mac in known_devices and known_devices[mac].get('departing'):
# Device returned during departure window — cancel timer, clear flag, no alert
known_devices[mac]['departing'] = False
with occupancy_lock:
_update_occupancy_state_unlocked()
last_departed_time.pop(mac, None)
logging.debug(f"Device {mac} re-arrived during departure window — suppressed alert")
return
# Log if we cancelled a timer
if timers_to_cancel:
logging.debug(f"Cancelled departure timer for {mac} — device re-arrived before debounce expired")
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)")
# Re-add to known_devices to track presence
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
with occupancy_lock:
_update_occupancy_state_unlocked()
return
# Still inside known_lock, but not inside departure_lock now
if mac in known_devices:
# 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)")
with occupancy_lock:
_update_occupancy_state_unlocked()
return
# Not a re-arrival, update last_seen and don't alert (already known)
with occupancy_lock:
_update_occupancy_state_unlocked()
return
# New device
vendor = lookup_oui(mac)
known_devices[mac] = {
'ip': ip,
'hostname': hostname,
'vendor': vendor,
'first_seen': now,
'last_seen': now
}
msg = format_arrival(hostname, ip, vendor or "unknown", mac)
logging.info(msg)
send_alert(msg)
update_occupancy_state()
def on_departure(mac: str) -> None:
"""
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
# 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
# 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
# Churn-based auto-suppression: if device cycles 3+ times in 30min, suppress
if _check_churn(mac, dev['ip']):
return
dev_copy = dev.copy()
# Mark departing instead of popping — allows re-arrival to detect and cancel timer
known_devices[mac]['departing'] = True
# Start 15-minute debounce timer
# Extract timer to cancel outside of lock to avoid deadlock
timer_to_cancel = None
with departure_lock:
# Cancel any existing timer for this MAC
if mac in departure_timers:
timer_to_cancel = departure_timers.pop(mac)
# Cancel outside lock
if timer_to_cancel:
timer_to_cancel.cancel()
with departure_lock:
def send_departure_alert():
"""Callback to send alert after debounce expires."""
# Lock order: known_lock, then departure_lock, then occupancy_lock
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)
with departure_lock:
last_departed_time[mac] = time.time()
if mac in departure_timers:
del departure_timers[mac]
# Update occupancy state after device removal (still inside known_lock)
with occupancy_lock:
_update_occupancy_state_unlocked()
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)
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:
"""Get primary network interface (not lo)."""
try:
with open('/proc/net/route') as f:
for line in f.readlines()[1:]:
parts = line.split()
if len(parts) > 1 and parts[0] != 'lo' and parts[1] == '00000000':
return parts[0]
with open('/proc/net/route') as f:
for line in f.readlines()[1:]:
parts = line.split()
if parts[0] != 'lo':
return parts[0]
except Exception:
pass
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.
Seeding steps:
1. Self IP (this device's interface IP)
2. Gateway IP (from 'ip route show default')
3. Broadcast IP (x.x.x.255)
4. Network equipment by OUI (Ubiquiti, Cisco, Aruba, TP-Link, Netgear, MikroTik, Ruckus)
5. Operator-provided IPs via NET_ALERTER_INFRA_IPS env var (comma-separated)
NET_ALERTER_INFRA_IPS is a post-deploy configuration for operators to add
infrastructure IPs known after deployment (access points, switches, etc).
"""
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
# Auto-exclude known network equipment by OUI
with known_lock:
for mac, dev in list(known_devices.items()):
oui = mac.replace(":", "").upper()[:6]
if oui in NETWORK_EQUIPMENT_OUIS:
infrastructure_ips.add(dev['ip'])
logging.info(f"OUI auto-exclusion: {dev['ip']} (MAC:{mac} OUI:{oui}) — network equipment detected")
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:
"""
DHCP sniffer thread.
AF_PACKET raw socket, captures DHCP traffic passively.
"""
try:
s = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(0x0003))
s.bind((iface, 0))
logging.info(f"DHCP sniffer started on {iface}")
while True:
try:
data, _ = s.recvfrom(65535)
parse_frame(data)
except Exception as e:
logging.error(f"DHCP sniffer error: {e}")
time.sleep(1)
except Exception as e:
logging.error(f"Failed to start DHCP sniffer: {e}")
def _update_last_seen(mac: str) -> None:
"""Update the last_seen timestamp for a MAC address (passive observation)."""
with last_seen_lock:
last_seen[mac] = time.time()
def _parse_mdns_for_names(payload: bytes, mac: str) -> None:
"""Parse mDNS DNS message to extract PTR/SRV/A record names and check against personal_names.
If a record name matches a name in personal_names, add the source MAC to personal_devices.
"""
try:
if len(payload) < 12: # DNS header minimum size
return
# DNS header: ID(2) Flags(2) QDCOUNT(2) ANCOUNT(2) NSCOUNT(2) ARCOUNT(2)
qdcount = struct.unpack('!H', payload[4:6])[0]
ancount = struct.unpack('!H', payload[6:8])[0]
# Skip questions section (we only care about answers/records)
offset = 12
for _ in range(qdcount):
offset = _skip_dns_name(payload, offset)
if offset is None or offset + 4 > len(payload):
return
offset += 4 # QTYPE + QCLASS
# Parse answers/authority/additional records
total_records = ancount + struct.unpack('!H', payload[8:10])[0] + struct.unpack('!H', payload[10:12])[0]
for _ in range(total_records):
if offset is None or offset + 10 > len(payload):
break
# Parse record name
name_offset = offset
offset = _skip_dns_name(payload, offset)
if offset is None:
break
# Parse TYPE(2) CLASS(2) TTL(4) RDLEN(2)
if offset + 10 > len(payload):
break
record_type, record_class, ttl, rdlen = struct.unpack('!HHIH', payload[offset:offset+10])
offset += 10
# Extract the name string from the record
name = _extract_dns_name(payload, name_offset)
if name and personal_names:
with personal_lock:
# Check if the record name matches any personal device names
for pname in personal_names:
if pname.lower() in name.lower():
# Found a match - add this MAC to personal_devices
mac_normalized = _normalize_mac(mac)
if mac_normalized:
personal_devices.add(mac_normalized)
logging.debug(f"mDNS auto-discovery: added {mac_normalized} for name match '{pname}' in '{name}'")
break
# Skip RDATA
if offset + rdlen > len(payload):
break
offset += rdlen
except Exception as e:
logging.debug(f"mDNS parse error: {e}")
def _skip_dns_name(payload: bytes, offset: int) -> int:
"""Skip a DNS name in the payload, handling compression pointers. Returns new offset or None on error."""
try:
while offset < len(payload):
length = payload[offset]
if length == 0:
return offset + 1
if (length & 0xC0) == 0xC0: # Pointer
return offset + 2
offset += 1 + length
return None
except Exception:
return None
def _extract_dns_name(payload: bytes, offset: int) -> str:
"""Extract a DNS name from payload at offset. Handles compression pointers."""
try:
parts = []
visited = set()
while offset < len(payload):
if offset in visited: # Prevent infinite loops
break
visited.add(offset)
length = payload[offset]
offset += 1
if length == 0:
break
elif (length & 0xC0) == 0xC0: # Pointer
if offset >= len(payload):
break
pointer_offset = struct.unpack('!H', bytes([length & 0x3F]) + bytes([payload[offset]]))[0]
offset = pointer_offset
continue
# Regular label
if offset + length > len(payload):
break
label = payload[offset:offset + length].decode('utf-8', errors='replace')
parts.append(label)
offset += length
return '.'.join(parts) if parts else ""
except Exception:
return ""
def parse_frame(data: bytes) -> None:
"""Parse Ethernet frame: ARP, mDNS, and DHCP packets for passive observation and event triggering."""
try:
if len(data) < 14:
return
# Ethernet frame — extract source MAC (bytes 6:12)
src_mac = ':'.join(f'{b:02x}' for b in data[6:12])
# Ethernet type field
eth_type = struct.unpack('!H', data[12:14])[0]
# Branch A: ARP frame (ethertype 0x0806)
if eth_type == 0x0806:
_update_last_seen(src_mac)
return
# For IP frames, extract IP header info
if eth_type != 0x0800: # not IPv4
return
# IP header
if len(data) < 23:
return
proto = data[23]
if proto != 17: # not UDP
return
# UDP header
if len(data) < 38:
return
src_port = struct.unpack('!H', data[34:36])[0]
dst_port = struct.unpack('!H', data[36:38])[0]
# Branch B: mDNS frame (IPv4 UDP dst port 5353)
if dst_port == 5353:
_update_last_seen(src_mac)
# Parse mDNS DNS message for device names
if len(data) >= 42:
_parse_mdns_for_names(data[42:], src_mac)
return
# Branch C: DHCP frame (UDP ports 67/68)
if not (src_port in (67, 68) and dst_port in (67, 68)):
return
# DHCP payload (starts at byte 42)
if len(data) < 42:
return
dhcp = data[42:]
if len(dhcp) < 236:
return
# MAC address (chaddr at offset 28, 6 bytes)
mac_bytes = dhcp[28:34]
mac = ':'.join(f'{b:02x}' for b in mac_bytes)
if mac in ('00:00:00:00:00:00', 'ff:ff:ff:ff:ff:ff'):
return
# Update last_seen for DHCP packets
_update_last_seen(mac)
# Parse DHCP options (start at byte 240)
if len(dhcp) < 240:
return
msg_type = None
hostname = ""
requested_ip = ""
ciaddr = socket.inet_ntoa(dhcp[12:16])
i = 240
while i < len(dhcp):
opt = dhcp[i]
if opt == 255:
break
if opt == 0:
i += 1
continue
if i + 1 >= len(dhcp):
break
length = dhcp[i + 1]
if i + 2 + length > len(dhcp):
break
val = dhcp[i + 2:i + 2 + length]
if opt == 53 and length == 1: # DHCP message type
msg_type = val[0]
elif opt == 12: # Hostname
hostname = val.decode('utf-8', errors='replace').strip('\x00')
elif opt == 50 and length == 4: # Requested IP
requested_ip = socket.inet_ntoa(val)
i += 2 + length
# Process based on message type
if msg_type in (1, 3): # Discover or Request (arrival)
ip = requested_ip or ciaddr
if ip and ip != '0.0.0.0':
on_arrival(mac, ip, hostname)
elif msg_type == 7: # Release (departure)
on_departure(mac)
except Exception as e:
logging.debug(f"Frame parse error: {e}")
# Netlink constants
NETLINK_ROUTE = 0
RTMGRP_NEIGH = 0x4
RTM_NEWNEIGH = 28
RTM_DELNEIGH = 29
NUD_REACHABLE = 0x02
NUD_STALE = 0x04
NUD_DELAY = 0x08
NUD_PROBE = 0x10
NDA_DST = 1
NDA_LLADDR = 2
def netlink_watcher() -> None:
"""
Netlink watcher thread.
Kernel pushes RTM_NEWNEIGH and RTM_DELNEIGH events.
"""
try:
s = socket.socket(socket.AF_NETLINK, socket.SOCK_RAW, NETLINK_ROUTE)
s.bind((os.getpid(), RTMGRP_NEIGH))
logging.info("Netlink neighbor watcher started")
while True:
try:
data = s.recv(65535)
parse_netlink(data)
except Exception as e:
logging.error(f"Netlink watcher error: {e}")
time.sleep(1)
except Exception as e:
logging.error(f"Failed to start Netlink watcher: {e}")
def parse_netlink(data: bytes) -> None:
"""Parse Netlink neighbor messages."""
try:
offset = 0
while offset < len(data):
if offset + 16 > len(data):
break
nlmsg_len, nlmsg_type, _, _, _ = struct.unpack_from('=IHHII', data, offset)
if nlmsg_len < 16:
break
payload = data[offset + 16:offset + nlmsg_len]
if nlmsg_type in (RTM_NEWNEIGH, RTM_DELNEIGH):
parse_ndmsg(payload, nlmsg_type)
offset += (nlmsg_len + 3) & ~3
except Exception as e:
logging.debug(f"Netlink parse error: {e}")
def parse_ndmsg(data: bytes, msg_type: int) -> None:
"""Parse ndmsg (neighbor discovery message)."""
try:
if len(data) < 12:
return
state = struct.unpack_from('=H', data, 8)[0]
mac = None
ip = None
offset = 12
while offset + 4 <= len(data):
rta_len, rta_type = struct.unpack_from('=HH', data, offset)
if rta_len < 4:
break
val = data[offset + 4:offset + rta_len]
if rta_type == NDA_LLADDR and len(val) == 6:
mac = ':'.join(f'{b:02x}' for b in val)
elif rta_type == NDA_DST:
if len(val) == 4:
ip = socket.inet_ntoa(val)
offset += (rta_len + 3) & ~3
if not mac or mac in ('00:00:00:00:00:00', 'ff:ff:ff:ff:ff:ff'):
return
if msg_type == RTM_NEWNEIGH and (state & (NUD_REACHABLE | NUD_STALE | NUD_DELAY | NUD_PROBE)):
on_arrival(mac, ip or 'unknown')
elif msg_type == RTM_DELNEIGH:
on_departure(mac)
except Exception as e:
logging.debug(f"ndmsg parse error: {e}")
def personal_watchdog() -> None:
"""Watchdog thread: fires on_departure for personal devices silent > WATCHDOG_TIMEOUT_SEC.
Monitors last_seen timestamps for personal devices and triggers departure alerts
if a device goes silent for longer than the configured timeout.
"""
while True:
try:
time.sleep(60) # Check every minute
now = time.time()
# Get current set of personal devices without holding lock for long
with personal_lock:
tracked = set(personal_devices)
for mac in tracked:
# Get last_seen timestamp
with last_seen_lock:
ts = last_seen.get(mac)
if ts is None:
continue
# Check if device is silent
if now - ts > WATCHDOG_TIMEOUT_SEC:
# Only fire if device is currently known and not already departing
# Don't hold lock when calling on_departure
with known_lock:
if mac in known_devices and not known_devices[mac].get('departing'):
# Copy the device info before releasing lock
dev_info = known_devices[mac].copy()
# Fire departure outside the lock
logging.info(f"Watchdog: personal device {mac} silent >{WATCHDOG_TIMEOUT_SEC}s, triggering departure")
on_departure(mac)
except Exception as e:
logging.error(f"Watchdog thread error: {e}")
def main() -> None:
"""Main entry point."""
logger = setup_logging()
load_env()
iface = get_primary_interface()
logging.info(f"Net alerter starting on interface {iface}")
seed_infrastructure_ips(iface)
seed_from_arp_cache()
# Start monitor threads
threads = [
threading.Thread(target=dhcp_sniffer, args=(iface,), daemon=True, name='dhcp-sniffer'),
threading.Thread(target=netlink_watcher, daemon=True, name='netlink-watcher'),
threading.Thread(target=personal_watchdog, daemon=True, name='personal-watchdog'),
]
for t in threads:
t.start()
logging.info("Net alerter running — DHCP sniffer + Netlink neighbor watcher + personal device watchdog active")
# Keep main thread alive
try:
while True:
time.sleep(60)
with known_lock:
logging.debug(f"Tracking {len(known_devices)} devices")
except KeyboardInterrupt:
logging.info("Shutting down")
sys.exit(0)
if __name__ == '__main__':
main()
Reference in New Issue View Git Blame Copy Permalink