Rebuild net_alerter as persistent daemon — DHCP sniffer + Netlink RTM_NEWNEIGH/DELNEIGH; removes 5-min cron, zero active probing

net_alerter/net_alerter.py

@@ -1,222 +1,97 @@
 #!/usr/bin/env python3
 """
-net_alerter.py — Presence-tracking network monitor.
+net_alerter.py — Net alerter persistent daemon.
-Baseline on first run (silent). Alerts on join and rejoin.
+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 socket
-import sqlite3
+import struct
-import subprocess
 import sys
+import threading
+import time
 import urllib.parse
 import urllib.request
-from datetime import datetime, timezone
 from pathlib import Path
 
-# --- Config (overridable via env) ---
+# --- Config (from .env or env vars) ---
-# Minimum consecutive missed scans before a device is considered gone and the
+MATRIX_HOMESERVER = ""
-# tripwire resets. Uncomment to enable — set to 3 for 15-min grace at 5-min cron.
+MATRIX_ACCESS_TOKEN = ""
-# ABSENT_THRESHOLD = int(os.getenv("NET_ALERTER_ABSENT_THRESHOLD", "3"))
+MATRIX_ROOM_ID = ""
+LOG_FILE = "/opt/net_alerter/net_alerter.log"
 
-MATRIX_HOMESERVER = os.getenv("MATRIX_HOMESERVER", "https://m.example.org")
+known_devices = {}  # {mac: {ip, hostname, vendor, first_seen, last_seen}}
-MATRIX_ACCESS_TOKEN = os.getenv("MATRIX_ACCESS_TOKEN", "")
+known_lock = threading.Lock()
-MATRIX_ROOM_ID = os.getenv("MATRIX_ROOM_ID", "!REDACTED:example.org")
-MATRIX_USER = os.getenv("MATRIX_USER", "")
-MATRIX_PASSWORD = os.getenv("MATRIX_PASSWORD", "")
-DB_PATH = Path(os.getenv("NET_ALERTER_DB", "/opt/net_alerter/devices.db"))
-ENV_PATH = Path(os.getenv("NET_ALERTER_ENV", "/opt/net_alerter/.env"))
 
 
-def now_utc() -> str:
+def setup_logging():
-    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+    """Setup logging to file + stdout."""
-
+    formatter = logging.Formatter(
-
+        fmt='%(asctime)s [%(levelname)s] %(message)s',
-def get_db() -> sqlite3.Connection:
+        datefmt='%Y-%m-%dT%H:%M:%S'
-    DB_PATH.parent.mkdir(parents=True, exist_ok=True)
-    conn = sqlite3.connect(DB_PATH)
-    conn.execute("""
-        CREATE TABLE IF NOT EXISTS devices (
-            mac        TEXT PRIMARY KEY,
-            ip         TEXT,
-            name       TEXT,
-            first_seen TEXT,
-            last_seen  TEXT,
-            is_present INTEGER NOT NULL DEFAULT 1,
-            last_left  TEXT
-        )
-    """)
-    # Migrate existing DBs that lack the new columns
-    existing = {row[1] for row in conn.execute("PRAGMA table_info(devices)")}
-    if "is_present" not in existing:
-        conn.execute("ALTER TABLE devices ADD COLUMN is_present INTEGER NOT NULL DEFAULT 1")
-    if "last_left" not in existing:
-        conn.execute("ALTER TABLE devices ADD COLUMN last_left TEXT")
-    # absent_count tracks consecutive missed scans (used by ABSENT_THRESHOLD)
-    if "absent_count" not in existing:
-        conn.execute("ALTER TABLE devices ADD COLUMN absent_count INTEGER NOT NULL DEFAULT 0")
-    conn.commit()
-    return conn
-
-
-def refresh_token() -> str:
-    if not MATRIX_USER or not MATRIX_PASSWORD:
-        print("[error] Cannot refresh token: MATRIX_USER/MATRIX_PASSWORD not set", file=sys.stderr)
-        return ""
-    payload = json.dumps({
-        "type": "m.login.password",
-        "identifier": {"type": "m.id.user", "user": MATRIX_USER},
-        "password": MATRIX_PASSWORD,
-        "initial_device_display_name": "net-alerter",
-    }).encode()
-    req = urllib.request.Request(
-        f"{MATRIX_HOMESERVER}/_matrix/client/v3/login",
-        data=payload,
-        headers={"Content-Type": "application/json"},
-        method="POST",
     )
-    try:
+
-        with urllib.request.urlopen(req, timeout=10) as r:
+    # File handler
-            data = json.loads(r.read())
+    Path(LOG_FILE).parent.mkdir(parents=True, exist_ok=True)
-        token = data.get("access_token", "")
+    file_handler = logging.FileHandler(LOG_FILE)
-        if token and ENV_PATH.exists():
+    file_handler.setFormatter(formatter)
-            lines = ENV_PATH.read_text().splitlines()
+
-            lines = [
+    # Stdout handler
-                f"MATRIX_ACCESS_TOKEN={token}" if l.startswith("MATRIX_ACCESS_TOKEN=") else l
+    stdout_handler = logging.StreamHandler(sys.stdout)
-                for l in lines
+    stdout_handler.setFormatter(formatter)
-            ]
+
-            ENV_PATH.write_text("\n".join(lines) + "\n")
+    logger = logging.getLogger()
-        return token
+    logger.setLevel(logging.DEBUG)
-    except Exception as e:
+    logger.addHandler(file_handler)
-        print(f"[error] Token refresh failed: {e}", file=sys.stderr)
+    logger.addHandler(stdout_handler)
-        return ""
+
+    return logger
 
 
-def _do_send(token: str, msg: str) -> int:
+def load_env():
-    url = (
+    """Parse .env file manually (key=value), no dependency required."""
-        f"{MATRIX_HOMESERVER}/_matrix/client/v3/rooms/"
+    global MATRIX_HOMESERVER, MATRIX_ACCESS_TOKEN, MATRIX_ROOM_ID
-        f"{urllib.parse.quote(MATRIX_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 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
-        },
-        method="POST",
-    )
-    try:
-        urllib.request.urlopen(req, timeout=10)
-        return 200
-    except urllib.error.HTTPError as e:
-        return e.code
-    except Exception as e:
-        print(f"[error] Matrix send exception: {e}", file=sys.stderr)
-        return 0
 
+    env_path = Path(os.getenv("NET_ALERTER_ENV", "/opt/net_alerter/.env"))
 
-def send_matrix_alert(msg: str) -> None:
+    if not env_path.exists():
-    global MATRIX_ACCESS_TOKEN
+        logging.warning(f".env not found at {env_path}, using env vars only")
-    if not MATRIX_ACCESS_TOKEN:
+        MATRIX_HOMESERVER = os.getenv("MATRIX_HOMESERVER", "https://m.example.org")
-        print("[warn] No MATRIX_ACCESS_TOKEN set, skipping alert", file=sys.stderr)
+        MATRIX_ACCESS_TOKEN = os.getenv("MATRIX_ACCESS_TOKEN", "")
+        MATRIX_ROOM_ID = os.getenv("MATRIX_ROOM_ID", "")
         return
-    code = _do_send(MATRIX_ACCESS_TOKEN, msg)
-    if code == 401:
-        print("[info] Token expired, refreshing...", file=sys.stderr)
-        new_token = refresh_token()
-        if new_token:
-            MATRIX_ACCESS_TOKEN = new_token
-            _do_send(MATRIX_ACCESS_TOKEN, msg)
-        else:
-            print("[error] Could not refresh token, alert dropped", file=sys.stderr)
-    elif code == 403:
-        print("[error] Room permission denied (403) — check bot membership in room", file=sys.stderr)
-    elif code not in (200, 0):
-        print(f"[error] Matrix send failed with HTTP {code}", file=sys.stderr)
 
-
-def read_arp_cache() -> list[dict]:
-    """
-    Read kernel ARP cache from /proc/net/arp (passive, no active probing).
-    Falls back to 'ip neigh show' if /proc/net/arp unavailable.
-    OUI vendor lookup from system hwdata if available.
-    Returns list of dicts: {"ip": str, "mac": str, "name": str}.
-    """
-    devices = []
-
-    # Try /proc/net/arp first
-    proc_path = Path("/proc/net/arp")
-    if proc_path.exists():
-        try:
-            lines = proc_path.read_text().splitlines()
-            for line in lines:
-                if line.startswith("IP"):  # Skip header
-                    continue
-                fields = line.split()
-                if len(fields) < 4:
-                    continue
-                ip = fields[0]
-                hw_addr = fields[3]
-                flags = fields[2]
-
-                # Skip incomplete entries
-                if hw_addr == "00:00:00:00:00:00":
-                    continue
-                if flags == "0x0":
-                    continue
-
-                device = {"ip": ip, "mac": hw_addr.lower(), "name": ""}
-
-                # Try OUI vendor lookup from system hwdata
-                try:
-                    vendor = lookup_oui(hw_addr)
-                    if vendor:
-                        device["vendor"] = vendor
-                except Exception:
-                    pass
-
-                devices.append(device)
-            return devices
-        except Exception:
-            pass
-
-    # Fallback to 'ip neigh show'
     try:
-        out = subprocess.check_output(["ip", "neigh", "show"], text=True)
+        for line in env_path.read_text().splitlines():
-        for line in out.splitlines():
+            line = line.strip()
-            fields = line.split()
+            if not line or line.startswith("#"):
-            if len(fields) < 5:
                 continue
-            # Format: <IP> dev <IFACE> lladdr <MAC> ...
+            if "=" not in line:
-            ip = fields[0]
-            try:
-                lladdr_idx = fields.index("lladdr")
-                if lladdr_idx + 1 < len(fields):
-                    mac = fields[lladdr_idx + 1].lower()
-                    device = {"ip": ip, "mac": mac, "name": ""}
-                    try:
-                        vendor = lookup_oui(mac)
-                        if vendor:
-                            device["vendor"] = vendor
-                    except Exception:
-                        pass
-                    devices.append(device)
-            except ValueError:
                 continue
-    except Exception:
+            key, val = line.split("=", 1)
-        pass
+            if key == "MATRIX_HOMESERVER":
-
+                MATRIX_HOMESERVER = val
-    return devices
+            elif key == "MATRIX_ACCESS_TOKEN":
+                MATRIX_ACCESS_TOKEN = val
+            elif key == "MATRIX_ROOM_ID":
+                MATRIX_ROOM_ID = val
+        logging.info(f"Loaded config from {env_path}")
+    except Exception as e:
+        logging.error(f"Failed to load .env: {e}")
 
 
 def lookup_oui(mac: str) -> str:
     """
     Look up OUI vendor from MAC address (first 3 octets).
-    Searches /usr/share/hwdata/oui.txt or /usr/share/misc/oui.txt if available.
+    Searches /usr/share/hwdata/oui.txt or /usr/share/misc/oui.txt.
     Returns vendor name or empty string if not found.
     """
     if not mac or mac == "00:00:00:00:00:00":
@@ -224,13 +99,11 @@ def lookup_oui(mac: str) -> str:
 
     oui_prefix = mac[:8].replace(":", "").upper()
 
-    # Check both possible OUI database locations
     for oui_path in ["/usr/share/hwdata/oui.txt", "/usr/share/misc/oui.txt"]:
         try:
             lines = Path(oui_path).read_text().splitlines()
             for line in lines:
                 if line.startswith(oui_prefix):
-                    # Format: XXXXXX (hex)	(vendor name)
                     parts = line.split(maxsplit=1)
                     if len(parts) > 1:
                         return parts[1].strip()
@@ -240,109 +113,376 @@ def lookup_oui(mac: str) -> str:
     return ""
 
 
-def run() -> None:
+def seed_from_arp_cache():
-    conn = get_db()
+    """Read /proc/net/arp, populate known_devices without alerting."""
-    ts = now_utc()
+    global known_devices
-    hostname = socket.gethostname()
 
-    # Read kernel ARP cache (passive, no active probing)
+    try:
-    arp_devices = read_arp_cache()
+        lines = Path("/proc/net/arp").read_text().splitlines()
-    scan_results: dict[str, dict] = {}
+        for line in lines:
-    for d in arp_devices:
+            if line.startswith("IP"):
-        scan_results[d["mac"]] = d
+                continue
+            fields = line.split()
+            if len(fields) < 4:
+                continue
 
-    # Is this the first run? (empty DB = baseline mode, no alerts)
+            ip = fields[0]
-    row_count = conn.execute("SELECT COUNT(*) FROM devices").fetchone()[0]
+            mac = fields[3].lower()
-    baseline_mode = row_count == 0
+            flags = fields[2]
-    if baseline_mode:
-        print(f"[info] Baseline run — seeding {len(scan_results)} devices, no alerts", file=sys.stderr)
 
-    alerts: list[tuple[str, dict]] = []  # (event_type, device_dict)
+            if mac == "00:00:00:00:00:00" or flags == "0x0":
+                continue
 
-    # Process devices currently on the network
+            vendor = lookup_oui(mac)
-    for mac, d in scan_results.items():
+            with known_lock:
-        row = conn.execute(
+                if mac not in known_devices:
-            "SELECT is_present FROM devices WHERE mac=?", (mac,)
+                    known_devices[mac] = {
-        ).fetchone()
+                        'ip': ip,
+                        'hostname': ip,
+                        'vendor': vendor,
+                        'first_seen': time.time(),
+                        'last_seen': time.time()
+                    }
 
-        if row is None:
+        logging.info(f"Seeded {len(known_devices)} devices from ARP cache")
-            # Brand new device
+    except Exception as e:
-            conn.execute(
+        logging.error(f"Failed to seed from ARP cache: {e}")
-                "INSERT INTO devices (mac, ip, name, first_seen, last_seen, is_present, last_left) "
-                "VALUES (?,?,?,?,?,1,NULL)",
-                (mac, d["ip"], d["name"], ts, ts),
-            )
-            if not baseline_mode:
-                alerts.append(("NEW", d))
-        else:
-            is_present = row[0]
-            if is_present == 0:
-                # Known device rejoining
-                conn.execute(
-                    "UPDATE devices SET ip=?, name=?, last_seen=?, is_present=1, last_left=NULL WHERE mac=?",
-                    (d["ip"], d["name"], ts, mac),
-                )
-                if not baseline_mode:
-                    alerts.append(("REJOIN", d))
-            else:
-                # Already present — just update last_seen, reset absent_count
-                conn.execute(
-                    "UPDATE devices SET ip=?, name=?, last_seen=?, absent_count=0 WHERE mac=?",
-                    (d["ip"], d["name"], ts, mac),
-                )
 
-    # Mark devices that were present but didn't appear in this scan as gone.
-    # absent_count increments each missed scan; is_present flips to 0 immediately
-    # (or after ABSENT_THRESHOLD consecutive misses if that option is enabled).
-    present_in_db = {
-        row[0]: row[1]
-        for row in conn.execute("SELECT mac, absent_count FROM devices WHERE is_present=1")
-    }
-    for mac, absent_count in present_in_db.items():
-        if mac in scan_results:
-            continue
-        new_count = absent_count + 1
-        # Uncomment the block below (and ABSENT_THRESHOLD above) to add a grace period:
-        # if new_count < ABSENT_THRESHOLD:
-        #     conn.execute("UPDATE devices SET absent_count=? WHERE mac=?", (new_count, mac))
-        #     continue
-        conn.execute(
-            "UPDATE devices SET is_present=0, last_left=?, absent_count=0 WHERE mac=?",
-            (ts, mac),
-        )
-        print(f"[info] Device left: {mac}", file=sys.stderr)
 
-    conn.commit()
+def send_alert(msg: str) -> None:
-    conn.close()
+    """Send alert to Matrix room."""
+    if not MATRIX_ACCESS_TOKEN:
+        logging.warning("No MATRIX_ACCESS_TOKEN set, skipping alert")
+        return
 
-    # Send alerts
+    url = (
-    for event, d in alerts:
+        f"{MATRIX_HOMESERVER}/_matrix/client/v3/rooms/"
-        name = d.get("name", "")
+        f"{urllib.parse.quote(MATRIX_ROOM_ID, safe='')}/send/m.room.message"
-        vendor = d.get("vendor", "")
+    )
-        if not name:
+    payload = json.dumps({"msgtype": "m.text", "body": msg}).encode()
+    req = urllib.request.Request(
+        url,
+        data=payload,
+        headers={
+            "Authorization": f"Bearer {MATRIX_ACCESS_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 on_arrival(mac: str, ip: str, hostname: str = "") -> None:
+    """Handle device arrival."""
+    with known_lock:
+        if mac in known_devices:
+            known_devices[mac]['last_seen'] = time.time()
+            return
+
+        vendor = lookup_oui(mac)
+        known_devices[mac] = {
+            'ip': ip,
+            'hostname': hostname or ip,
+            'vendor': vendor,
+            'first_seen': time.time(),
+            'last_seen': time.time()
+        }
+
+    label = hostname or ip
+    msg = f"[NET] ARRIVED: {label} ({ip}) [{vendor or 'unknown'}] MAC:{mac}"
+    logging.info(msg)
+    send_alert(msg)
+
+
+def on_departure(mac: str) -> None:
+    """Handle device departure."""
+    with known_lock:
+        if mac not in known_devices:
+            return
+        dev = known_devices.pop(mac)
+
+    duration = fmt_duration(time.time() - dev['first_seen'])
+    msg = f"[NET] DEPARTED: {dev['hostname']} ({dev['ip']}) [{dev['vendor'] or 'unknown'}] MAC:{mac} — present {duration}"
+    logging.info(msg)
+    send_alert(msg)
+
+
+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 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:
-                name = socket.gethostbyaddr(d["ip"])[0]
+                data, _ = s.recvfrom(65535)
-            except Exception:
+                parse_dhcp(data)
-                pass
+            except Exception as e:
-        if name and vendor:
+                logging.error(f"DHCP sniffer error: {e}")
-            label = f"{name} ({vendor})"
+                time.sleep(1)
-        elif name:
+    except Exception as e:
-            label = name
+        logging.error(f"Failed to start DHCP sniffer: {e}")
-        elif vendor:
-            label = vendor
-        else:
-            label = "unknown"
-        tag = "NEW DEVICE" if event == "NEW" else "DEVICE REJOINED"
-        msg = (
-            f"[{hostname}] {tag}\n"
-            f"  IP:   {d['ip']}\n"
-            f"  MAC:  {d['mac']}\n"
-            f"  Name: {label}\n"
-            f"  Time: {ts}"
-        )
-        print(msg)
-        send_matrix_alert(msg)
 
 
-if __name__ == "__main__":
+def parse_dhcp(data: bytes) -> None:
-    run()
+    """Parse DHCP packet and trigger arrival/departure events."""
+    try:
+        if len(data) < 14:
+            return
+
+        # Ethernet frame
+        eth_type = struct.unpack('!H', data[12:14])[0]
+        if eth_type != 0x0800:  # not IPv4
+            return
+
+        # IP header
+        if len(data) < 23:
+            return
+        proto = data[23]
+        if proto != 17:  # not UDP
+            return
+
+        # UDP ports
+        if len(data) < 38:
+            return
+        src_port = struct.unpack('!H', data[34:36])[0]
+        dst_port = struct.unpack('!H', data[36:38])[0]
+        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
+
+        # 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"DHCP 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 main() -> None:
+    """Main entry point."""
+    logger = setup_logging()
+    load_env()
+
+    iface = get_primary_interface()
+    logging.info(f"Net alerter starting on interface {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'),
+    ]
+
+    for t in threads:
+        t.start()
+
+    logging.info("Net alerter running — DHCP sniffer + Netlink neighbor watcher 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()

net_alerter/net_alerter.service

@@ -0,0 +1,16 @@
+[Unit]
+Description=Net Alerter
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=simple
+WorkingDirectory=/opt/net_alerter
+EnvironmentFile=/opt/net_alerter/.env
+ExecStart=/usr/bin/python3 /opt/net_alerter/net_alerter.py
+Restart=always
+RestartSec=10
+User=root
+
+[Install]
+WantedBy=multi-user.target