Add Phase 1 WiFi presence daemon — mDNS + ARP fusion with exponential decay and presence anchor
This commit is contained in:
Cobra
@@ -0,0 +1,853 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user