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

Implement occupancy tracking with personal device detection and deadlock fix

Browse Source 
- Add MOBILE_DEVICE_OUIS set with 55+ OUI prefixes for personal devices
- Add is_personal_device() function to detect personal devices by OUI or manual config
- Add load_personal_macs_from_env() to parse NET_ALERTER_PERSONAL_MACS
- Add location_occupancy state tracking (VACANT/OCCUPIED/UNKNOWN)
- Add update_occupancy_state() function with thread-safe alerts on transitions
- Add _update_occupancy_state_unlocked() helper to prevent deadlock
- Fix deadlock in on_arrival() by using unlocked version when holding known_lock
- Update test suite with 5 new occupancy tests
- Fix test assertions to exclude occupancy alerts for device arrival tests
This commit is contained in:
Cobra
2026-04-12 07:26:57 -04:00
parent 96d1042239
commit ec6e9e31d1
2 changed files with 348 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
net_alerter/net_alerter.py
+128 -1
View File
@@ -75,6 +75,49 @@ _churn_lock = threading.Lock()
CHURN_THRESHOLD = 3 # cycles in window before suppression
CHURN_WINDOW_SEC = 1800 # 30 minutes
# 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_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)."""
@@ -322,6 +365,19 @@ def setup_logging():
return logger
def load_personal_macs_from_env() -> None:
"""Parse NET_ALERTER_PERSONAL_MACS env var and add to personal_devices set."""
macs_str = os.getenv("NET_ALERTER_PERSONAL_MACS", "")
if not macs_str.strip():
return
macs = [m.strip().upper() for m in macs_str.split(",") if m.strip()]
with personal_lock:
for mac in macs:
personal_devices.add(mac)
logging.info(f"Added personal device MAC: {mac}")
def load_env():
"""Parse .env file manually (key=value), no dependency required."""
global MATRIX_HOMESERVER, MATRIX_ACCESS_TOKEN, MATRIX_ROOM_ID
@@ -333,6 +389,7 @@ def load_env():
MATRIX_HOMESERVER = os.getenv("MATRIX_HOMESERVER", "https://m.example.org")
MATRIX_ACCESS_TOKEN = os.getenv("MATRIX_ACCESS_TOKEN", "")
MATRIX_ROOM_ID = os.getenv("MATRIX_ROOM_ID", "")
load_personal_macs_from_env()
return
try:
@@ -350,10 +407,32 @@ def load_env():
elif key == "MATRIX_ROOM_ID":
MATRIX_ROOM_ID = val
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_upper = mac.replace(":", "").upper()
oui = mac_upper[:6]
# Check personal_devices set first (manually configured)
with personal_lock:
if mac_upper 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).
@@ -376,7 +455,7 @@ def lookup_oui(mac: str) -> str:
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")
@@ -487,6 +566,46 @@ def format_departure(hostname: str, ip: str, vendor: str, mac: str, duration: st
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 is already held by caller.
"""
global location_occupancy
with occupancy_lock:
# 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 known_lock before checking devices.
"""
with known_lock:
_update_occupancy_state_unlocked()
def on_arrival(mac: str, ip: str, hostname: str = "") -> None:
"""Handle device arrival."""
# Infrastructure IPs never trigger alerts
@@ -512,6 +631,7 @@ def on_arrival(mac: str, ip: str, hostname: str = "") -> None:
del departure_timers[mac]
with known_lock:
known_devices[mac]['departing'] = False
_update_occupancy_state_unlocked()
last_departed_time.pop(mac, None)
logging.debug(f"Device {mac} re-arrived during departure window — suppressed alert")
return
@@ -541,6 +661,7 @@ def on_arrival(mac: str, ip: str, hostname: str = "") -> None:
}
else:
known_devices[mac]['last_seen'] = now
_update_occupancy_state_unlocked()
return
with known_lock:
@@ -552,9 +673,11 @@ def on_arrival(mac: str, ip: str, hostname: str = "") -> None:
# 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)")
_update_occupancy_state_unlocked()
return
# Not a re-arrival, update last_seen and don't alert (already known)
_update_occupancy_state_unlocked()
return
# New device
@@ -570,6 +693,7 @@ def on_arrival(mac: str, ip: str, hostname: str = "") -> None:
msg = format_arrival(hostname, ip, vendor or "unknown", mac)
logging.info(msg)
send_alert(msg)
update_occupancy_state()
def on_departure(mac: str) -> None:
@@ -625,6 +749,9 @@ def on_departure(mac: str) -> None:
if mac in departure_timers:
del departure_timers[mac]
# Update occupancy state after device removal
update_occupancy_state()
timer = threading.Timer(900.0, send_departure_alert) # 15 min = 900 sec
timer.daemon = True
timer.name = f'departure-debounce-{mac}'