From 3ddb854685ec321775623a53d398fc03f81826cf Mon Sep 17 00:00:00 2001 From: Cobra Date: Fri, 10 Apr 2026 14:26:27 -0400 Subject: [PATCH] =?UTF-8?q?Revert=20"Add=20Phase=201=20WiFi=20presence=20d?= =?UTF-8?q?aemon=20=E2=80=94=20mDNS=20+=20ARP=20fusion=20with=20exponentia?= =?UTF-8?q?l=20decay=20and=20presence=20anchor"?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This reverts commit b9adf498106c52f1c7508a791a7aac30d5ca1955. --- net_alerter/presence_daemon.py | 853 --------------------------------- 1 file changed, 853 deletions(-) delete mode 100644 net_alerter/presence_daemon.py diff --git a/net_alerter/presence_daemon.py b/net_alerter/presence_daemon.py deleted file mode 100644 index 02dcb26..0000000 --- a/net_alerter/presence_daemon.py +++ /dev/null @@ -1,853 +0,0 @@ -#!/usr/bin/env python3 -""" -presence_daemon.py — WiFi person presence intelligence daemon (Phase 1). -Fuses mDNS + ARP signals with exponential decay and presence anchor. - -State machine per MAC: ABSENT → PRESENT → TRANSITIONING_OUT → ABSENT -Certainty weights: mDNS=0.85, ARP=0.65, decay_k=0.08/min -Presence anchor: 45 min from last signal >= 0.40 decayed -Transition window: 15 min in TRANSITIONING_OUT -""" - -import argparse -import json -import logging -import math -import os -import signal -import socket -import sqlite3 -import struct -import subprocess -import sys -import threading -import time -import urllib.parse -import urllib.request -from pathlib import Path - -# Signal model -MDNS_CERTAINTY = 0.85 -ARP_CERTAINTY = 0.65 -DECAY_K = 0.08 -PRESENCE_ANCHOR_MINUTES = 45 -TRANSITION_MINUTES = 15 -CERTAINTY_THRESHOLD = 0.50 -ANCHOR_THRESHOLD = 0.40 -STATE_TICK_INTERVAL = 30 # seconds -DB_BATCH_INTERVAL = 30 # seconds -MDNS_MULTICAST_ADDR = "224.0.0.251" -MDNS_PORT = 5353 - -# Netlink constants -NETLINK_ROUTE = 0 -RTMGRP_NEIGH = 4 -RTM_NEWNEIGH = 16 -RTM_DELNEIGH = 17 - -# OUI dictionary (common vendors, matches net_alerter.py) -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", -} - -# Global state -persons = {} # {mac: {hostname, vendor, first_seen, last_seen, state, certainty, signals, anchor_time}} -persons_lock = threading.Lock() -infrastructure_ips = set() -signals_queue = [] -signals_queue_lock = threading.Lock() -events_queue = [] -events_queue_lock = threading.Lock() - -# Interface flap detection -_flap_window = [] -_flap_lock = threading.Lock() -_interface_recovering = False -_recovery_timer = None - -# Global shutdown -shutdown_event = threading.Event() - - -def setup_logging(log_file, log_level): - """Configure logging.""" - level = getattr(logging, log_level, logging.INFO) - fmt = "%(asctime)s [%(levelname)s] %(message)s" - - if log_file: - handler = logging.FileHandler(log_file) - else: - handler = logging.StreamHandler(sys.stderr) - - handler.setFormatter(logging.Formatter(fmt)) - logging.root.handlers.clear() - logging.root.addHandler(handler) - logging.root.setLevel(level) - - -def decay(certainty: float, age_minutes: float) -> float: - """Apply exponential decay to certainty over time.""" - return certainty * math.exp(-DECAY_K * age_minutes) - - -def wifi_fusion(mdns_c: float, arp_c: float) -> float: - """Corroboration bonus when both signals present.""" - a, b = max(mdns_c, arp_c), min(mdns_c, arp_c) - return a + 0.08 * b - - -def get_primary_interface() -> str: - """Get primary network interface (not lo). Matches net_alerter.py pattern.""" - 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: - 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. Matches net_alerter.py.""" - global infrastructure_ips - infrastructure_ips.clear() - - self_ip = get_local_ip(iface) - if self_ip: - infrastructure_ips.add(self_ip) - logging.info(f"Seeded self IP {self_ip} as infrastructure") - - try: - result = subprocess.run(['ip', 'route', 'show', 'default'], - capture_output=True, text=True, timeout=5) - if result.returncode == 0: - 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}") - - 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 lookup_oui(mac: str) -> str: - """Look up OUI vendor from MAC address (first 3 octets).""" - if not mac or mac == "00:00:00:00:00:00": - return "unknown" - - oui_prefix = mac[:8].lower() - oui_prefix_upper = oui_prefix.replace(":", "").upper() - return OUI_DICT.get(oui_prefix_upper, "unknown") - - -def read_arp_cache() -> dict: - """Read /proc/net/arp and return {ip: mac} mapping.""" - arp_map = {} - try: - with open('/proc/net/arp') as f: - for line in f.readlines()[1:]: - fields = line.split() - if len(fields) < 4: - continue - ip = fields[0] - mac = fields[3].lower() - if mac != "00:00:00:00:00:00": - arp_map[ip] = mac - except Exception as e: - logging.debug(f"Failed to read ARP cache: {e}") - return arp_map - - -def _check_flap(mac: str) -> bool: - """Check if interface is flapping; skip RTM_DELNEIGH if true.""" - 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.warning(f"Interface flap detected, entering recovery mode for 60s") - return _interface_recovering - - -def _clear_recovery(): - """Clear recovery flag after timeout.""" - global _interface_recovering - _interface_recovering = False - logging.info("Interface recovery timeout expired") - - -def record_signal(mac: str, hostname: str = "", signal_type: str = "arp", - certainty: float = ARP_CERTAINTY) -> None: - """Record a signal and update person state.""" - if not mac or mac == "00:00:00:00:00:00" or mac in ("ff:ff:ff:ff:ff:ff", "FF:FF:FF:FF:FF:FF"): - return - - mac = mac.lower() - - # Check infrastructure exclusion - if signal_type == "arp": - arp_map = read_arp_cache() - if mac in arp_map: - ip = arp_map[mac] - if ip in infrastructure_ips: - return - - now = time.time() - with persons_lock: - if mac not in persons: - vendor = lookup_oui(mac) - persons[mac] = { - 'hostname': hostname or "", - 'vendor': vendor, - 'first_seen': now, - 'last_seen': now, - 'state': 'ABSENT', - 'certainty': 0.0, - 'signals': [], - 'anchor_time': None, - } - - person = persons[mac] - person['last_seen'] = now - if hostname and not person['hostname']: - person['hostname'] = hostname - - # Add signal to history (keep last 100) - person['signals'].append((signal_type, certainty, now)) - person['signals'] = person['signals'][-100:] - - # Queue signal for DB - with signals_queue_lock: - signals_queue.append((mac, signal_type, certainty, now)) - - -def compute_person_certainty(mac: str) -> float: - """Compute current certainty for a person based on decayed signals.""" - with persons_lock: - if mac not in persons: - return 0.0 - - person = persons[mac] - now = time.time() - signals = person['signals'] - - if not signals: - return 0.0 - - # Separate by type and get most recent - mdns_signals = [(c, t) for st, c, t in signals if st == "mdns"] - arp_signals = [(c, t) for st, c, t in signals if st == "arp"] - - mdns_c = 0.0 - arp_c = 0.0 - - if mdns_signals: - cert, timestamp = mdns_signals[-1] - age_min = (now - timestamp) / 60.0 - mdns_c = decay(cert, age_min) - - if arp_signals: - cert, timestamp = arp_signals[-1] - age_min = (now - timestamp) / 60.0 - arp_c = decay(cert, age_min) - - # Fusion - if mdns_c > 0 and arp_c > 0: - return wifi_fusion(mdns_c, arp_c) - else: - return max(mdns_c, arp_c) - - -def emit_event(mac: str, event_type: str, person: dict) -> None: - """Emit a state transition event.""" - now = time.time() - event = { - 'mac': mac, - 'hostname': person.get('hostname', ''), - 'vendor': person.get('vendor', 'unknown'), - 'certainty': person.get('certainty', 0.0), - 'state': person.get('state', 'ABSENT'), - 'event': event_type, - 'first_seen': person.get('first_seen', now), - 'last_seen': person.get('last_seen', now), - 'duration_min': (person.get('last_seen', now) - person.get('first_seen', now)) / 60.0, - } - with events_queue_lock: - events_queue.append((mac, event_type, person.get('certainty', 0.0), now, event)) - - -def state_machine_tick() -> None: - """Run state machine: decay signals, compute certainty, check transitions.""" - now = time.time() - - with persons_lock: - for mac, person in list(persons.items()): - old_state = person['state'] - certainty = compute_person_certainty(mac) - person['certainty'] = certainty - - if person['state'] == 'ABSENT': - # ABSENT → PRESENT - if certainty >= CERTAINTY_THRESHOLD: - person['state'] = 'PRESENT' - person['anchor_time'] = now - logging.info(f"{mac} ({person['hostname']}): ABSENT → PRESENT (cert={certainty:.2f})") - emit_event(mac, 'arrived', person) - - elif person['state'] == 'PRESENT': - # Update anchor if recent strong signal - if certainty >= ANCHOR_THRESHOLD: - person['anchor_time'] = now - - # PRESENT → TRANSITIONING_OUT (anchor expired) - if person['anchor_time'] and (now - person['anchor_time']) > PRESENCE_ANCHOR_MINUTES * 60: - person['state'] = 'TRANSITIONING_OUT' - person['transitioning_start'] = now - logging.info(f"{mac} ({person['hostname']}): PRESENT → TRANSITIONING_OUT (anchor expired)") - - elif person['state'] == 'TRANSITIONING_OUT': - # Check for re-entry - if certainty >= ANCHOR_THRESHOLD: - person['state'] = 'PRESENT' - person['anchor_time'] = now - logging.info(f"{mac} ({person['hostname']}): TRANSITIONING_OUT → PRESENT (re-entry)") - - # TRANSITIONING_OUT → ABSENT (timeout) - if (now - person.get('transitioning_start', now)) > TRANSITION_MINUTES * 60: - person['state'] = 'ABSENT' - person['session_count'] = person.get('session_count', 1) + 1 - person['total_minutes'] += (person['last_seen'] - person['first_seen']) / 60.0 - logging.info(f"{mac} ({person['hostname']}): TRANSITIONING_OUT → ABSENT (timeout)") - emit_event(mac, 'departed', person) - - -def state_machine_thread(interval: float) -> None: - """Periodic state machine tick.""" - while not shutdown_event.is_set(): - try: - state_machine_tick() - shutdown_event.wait(interval) - except Exception as e: - logging.error(f"State machine error: {e}") - - -def parse_mdns_packet(data: bytes, src_ip: str) -> list: - """ - Parse raw mDNS packet and extract (hostname, signal_type, certainty). - Returns list of (hostname, signal_type) tuples. - """ - signals = [] - try: - if len(data) < 12: - return signals - - # DNS header: ID(2), Flags(2), QD(2), AN(2), NS(2), AR(2) = 12 bytes - qd_count = struct.unpack('!H', data[4:6])[0] - an_count = struct.unpack('!H', data[6:8])[0] - - # Skip questions - offset = 12 - for _ in range(qd_count): - while offset < len(data) and data[offset] != 0: - offset += data[offset] + 1 - if offset < len(data): - offset += 5 # null + type(2) + class(2) - - # Parse answers (RRs) - for _ in range(an_count): - if offset >= len(data): - break - - # Parse name (skip) - while offset < len(data) and data[offset] != 0: - if data[offset] & 0xc0: # Compression pointer - offset += 2 - break - else: - offset += data[offset] + 1 - if offset < len(data): - offset += 1 # null terminator - - if offset + 10 > len(data): - break - - rrtype = struct.unpack('!H', data[offset:offset+2])[0] - rrclass = struct.unpack('!H', data[offset+2:offset+4])[0] - rdlen = struct.unpack('!H', data[offset+8:offset+10])[0] - rdata_offset = offset + 10 - - if rdata_offset + rdlen > len(data): - break - - # Extract signals from PTR (12), A (1), SRV (33) - if rrtype == 12: # PTR - # PTR points to a hostname - try: - name_bytes = data[rdata_offset:rdata_offset+rdlen] - hostname = parse_dns_name(name_bytes, data) - if hostname and '_services._dns-sd._udp.local' in hostname.lower(): - signals.append(('service_ptr', 'mdns')) - except Exception: - pass - - elif rrtype == 1: # A record - if rdlen >= 4: - signals.append(('a_record', 'mdns')) - - elif rrtype == 33: # SRV - signals.append(('srv_record', 'mdns')) - - offset = rdata_offset + rdlen - - return signals - - except Exception as e: - logging.debug(f"mDNS parse error: {e}") - return signals - - -def parse_dns_name(data: bytes, full_packet: bytes) -> str: - """Parse DNS name from packet data with pointer support.""" - try: - parts = [] - offset = 0 - while offset < len(data): - length = data[offset] - if length == 0: - break - if length & 0xc0: # Compression pointer - break - offset += 1 - if offset + length > len(data): - break - parts.append(data[offset:offset+length].decode('utf-8', errors='ignore')) - offset += length - return '.'.join(parts) - except Exception: - return "" - - -def mdns_listener_thread(iface: str) -> None: - """Listen for mDNS packets on 224.0.0.251:5353.""" - try: - s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) - s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) - - # Bind to MDNS port - s.bind(('', MDNS_PORT)) - - # Join multicast group - mreq = socket.inet_aton(MDNS_MULTICAST_ADDR) + socket.inet_aton('0.0.0.0') - s.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq) - - logging.info(f"mDNS listener started on {MDNS_MULTICAST_ADDR}:{MDNS_PORT}") - - while not shutdown_event.is_set(): - try: - s.settimeout(5.0) - data, (src_ip, src_port) = s.recvfrom(4096) - - # Parse packet - signals = parse_mdns_packet(data, src_ip) - - if signals: - # Look up MAC from ARP cache - arp_map = read_arp_cache() - mac = arp_map.get(src_ip) - - if mac: - # Extract hostname if possible - hostname = f"mDNS_{src_ip}" - record_signal(mac, hostname, 'mdns', MDNS_CERTAINTY) - logging.debug(f"mDNS signal: {mac} from {src_ip}") - - except socket.timeout: - pass - except Exception as e: - logging.error(f"mDNS listener error: {e}") - time.sleep(1) - - except Exception as e: - logging.error(f"Failed to start mDNS listener: {e}") - - -def netlink_watcher_thread(iface: str) -> None: - """Watch Netlink 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 not shutdown_event.is_set(): - try: - s.settimeout(5.0) - data = s.recv(65535) - parse_netlink(data) - except socket.timeout: - pass - 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 messages and extract ndmsg events.""" - 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) and extract MAC/IP.""" - try: - if len(data) < 12: - return - - # ndmsg: family(1), pad(1), ifindex(2), state(2), flags(1), type(1) - family = data[0] - if family not in (2, 10): # AF_INET (2) or AF_INET6 (10) - return - - # Parse attributes (RTA_*) - offset = 12 - mac = None - ip = None - - while offset < len(data): - if offset + 4 > len(data): - break - - rta_len = struct.unpack('!H', data[offset:offset+2])[0] - rta_type = struct.unpack('!H', data[offset+2:offset+4])[0] - - if rta_len < 4: - break - - payload_len = rta_len - 4 - payload = data[offset+4:offset+4+payload_len] - - # RTA_DST (1) = IP, RTA_LLADDR (2) = MAC - if rta_type == 1 and family == 2: # IPv4 - if payload_len == 4: - ip = '.'.join(str(b) for b in payload) - elif rta_type == 2 and payload_len == 6: # MAC - mac = ':'.join(f'{b:02x}' for b in payload) - - offset += (rta_len + 3) & ~3 - - if msg_type == RTM_NEWNEIGH and mac and ip: - record_signal(mac, "", 'arp', ARP_CERTAINTY) - logging.debug(f"ARP signal: {mac} ({ip})") - - elif msg_type == RTM_DELNEIGH and mac: - if _check_flap(mac): - logging.debug(f"Skipping RTM_DELNEIGH for {mac} (flap detected)") - else: - logging.debug(f"ARP departure candidate: {mac}") - - except Exception as e: - logging.debug(f"ndmsg parse error: {e}") - - -def init_db(db_path: str) -> None: - """Initialize SQLite database with schema.""" - try: - conn = sqlite3.connect(db_path, timeout=10) - cursor = conn.cursor() - cursor.execute("PRAGMA journal_mode=WAL") - - # Load schema from file if it exists - schema_file = Path(__file__).parent / "presence_schema.sql" - if schema_file.exists(): - schema = schema_file.read_text() - cursor.executescript(schema) - else: - # Create inline - cursor.execute(""" - CREATE TABLE IF NOT EXISTS persons ( - mac TEXT PRIMARY KEY, - hostname TEXT, - vendor TEXT, - first_seen REAL NOT NULL, - last_seen REAL NOT NULL, - state TEXT NOT NULL DEFAULT 'ABSENT', - certainty REAL NOT NULL DEFAULT 0.0, - session_count INTEGER NOT NULL DEFAULT 1, - total_minutes REAL NOT NULL DEFAULT 0.0 - ) - """) - cursor.execute(""" - CREATE TABLE IF NOT EXISTS signals ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - mac TEXT NOT NULL, - signal_type TEXT NOT NULL, - certainty REAL NOT NULL, - observed_at REAL NOT NULL - ) - """) - cursor.execute(""" - CREATE TABLE IF NOT EXISTS events ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - mac TEXT NOT NULL, - hostname TEXT, - event_type TEXT NOT NULL, - certainty REAL NOT NULL, - timestamp REAL NOT NULL - ) - """) - cursor.execute("CREATE INDEX IF NOT EXISTS idx_signals_mac ON signals(mac)") - cursor.execute("CREATE INDEX IF NOT EXISTS idx_signals_mac_time ON signals(mac, observed_at)") - cursor.execute("CREATE INDEX IF NOT EXISTS idx_events_mac ON events(mac)") - cursor.execute("CREATE INDEX IF NOT EXISTS idx_events_timestamp ON events(timestamp)") - - conn.commit() - conn.close() - logging.info(f"Database initialized: {db_path}") - - except Exception as e: - logging.error(f"Failed to initialize database: {e}") - - -def db_batch_writer_thread(db_path: str, interval: float) -> None: - """Periodically write batched signals and events to SQLite.""" - while not shutdown_event.is_set(): - try: - shutdown_event.wait(interval) - - # Flush signals - with signals_queue_lock: - if signals_queue: - conn = sqlite3.connect(db_path, timeout=10) - cursor = conn.cursor() - cursor.executemany( - "INSERT INTO signals (mac, signal_type, certainty, observed_at) VALUES (?, ?, ?, ?)", - signals_queue - ) - conn.commit() - conn.close() - logging.debug(f"Wrote {len(signals_queue)} signals to DB") - signals_queue.clear() - - # Flush events - with events_queue_lock: - if events_queue: - conn = sqlite3.connect(db_path, timeout=10) - cursor = conn.cursor() - for mac, event_type, certainty, timestamp, _ in events_queue: - cursor.execute( - """INSERT INTO events (mac, hostname, event_type, certainty, timestamp) - VALUES (?, ?, ?, ?, ?)""", - (mac, "", event_type, certainty, timestamp) - ) - conn.commit() - conn.close() - logging.debug(f"Wrote {len(events_queue)} events to DB") - events_queue.clear() - - except Exception as e: - logging.error(f"DB batch writer error: {e}") - - -def send_output(mode: str, output_file: str, matrix_webhook: str, event: dict) -> None: - """Emit event in specified output mode.""" - try: - json_str = json.dumps(event) - - if mode == 'stdout': - print(json_str) - - elif mode == 'file' and output_file: - with open(output_file, 'a') as f: - f.write(json_str + '\n') - - elif mode == 'matrix' and matrix_webhook: - payload = json.dumps({'text': json_str}).encode('utf-8') - req = urllib.request.Request(matrix_webhook, data=payload) - req.add_header('Content-Type', 'application/json') - urllib.request.urlopen(req, timeout=5) - - except Exception as e: - logging.error(f"Output error ({mode}): {e}") - - -def output_thread(mode: str, output_file: str, matrix_webhook: str) -> None: - """Emit queued events.""" - while not shutdown_event.is_set(): - try: - with events_queue_lock: - if events_queue: - for mac, event_type, certainty, timestamp, event in list(events_queue): - send_output(mode, output_file, matrix_webhook, event) - - time.sleep(0.5) - - except Exception as e: - logging.error(f"Output thread error: {e}") - - -def handle_shutdown(signum, frame): - """Handle SIGTERM/SIGINT.""" - logging.info(f"Received signal {signum}, shutting down...") - shutdown_event.set() - - -def main(): - parser = argparse.ArgumentParser( - description='WiFi person presence intelligence daemon (Phase 1)' - ) - parser.add_argument('--iface', default=None, help='Network interface (auto-detect if not set)') - parser.add_argument('--db', default='/tmp/presence.db', help='SQLite database path') - parser.add_argument('--output', choices=['stdout', 'file', 'matrix'], default='stdout', - help='Output mode') - parser.add_argument('--output-file', help='File path for file output mode') - parser.add_argument('--matrix-webhook', help='Matrix webhook URL for matrix output mode') - parser.add_argument('--log-level', choices=['DEBUG', 'INFO', 'WARNING'], default='INFO', - help='Log level') - parser.add_argument('--log-file', help='Log file path (default: stderr)') - - args = parser.parse_args() - - setup_logging(args.log_file, args.log_level) - logging.info("presence_daemon starting...") - - # Determine interface - iface = args.iface or get_primary_interface() - logging.info(f"Using interface: {iface}") - - # Initialize database - init_db(args.db) - - # Seed infrastructure IPs - seed_infrastructure_ips(iface) - - # Setup signal handlers - signal.signal(signal.SIGTERM, handle_shutdown) - signal.signal(signal.SIGINT, handle_shutdown) - - # Start threads - threads = [] - - t_mdns = threading.Thread(target=mdns_listener_thread, args=(iface,), daemon=True) - threads.append(t_mdns) - t_mdns.start() - - t_netlink = threading.Thread(target=netlink_watcher_thread, args=(iface,), daemon=True) - threads.append(t_netlink) - t_netlink.start() - - t_state = threading.Thread(target=state_machine_thread, args=(STATE_TICK_INTERVAL,), daemon=True) - threads.append(t_state) - t_state.start() - - t_db = threading.Thread(target=db_batch_writer_thread, args=(args.db, DB_BATCH_INTERVAL), daemon=True) - threads.append(t_db) - t_db.start() - - t_output = threading.Thread(target=output_thread, args=(args.output, args.output_file, args.matrix_webhook), - daemon=True) - threads.append(t_output) - t_output.start() - - logging.info("All threads started. Monitoring presence...") - - # Block until shutdown - try: - while not shutdown_event.is_set(): - time.sleep(1) - except KeyboardInterrupt: - pass - - logging.info("Flushing final batch writes...") - time.sleep(DB_BATCH_INTERVAL + 1) - - logging.info("presence_daemon stopped") - - -if __name__ == '__main__': - main()