Replace nmap ARP scan with passive /proc/net/arp read — no active probing (#449)

- Removed get_local_subnets() and scan_subnet() — no longer needed
- Added read_arp_cache() to read kernel ARP table from /proc/net/arp
- Added fallback to 'ip neigh show' if /proc/net/arp unavailable
- Added lookup_oui() to query system /usr/share/hwdata/oui.txt for vendor names
- Removed subprocess nmap call and re import (no longer needed)
- run() now directly reads ARP cache instead of scanning subnets
- Zero active network traffic — purely passive monitoring
- Removed sudo requirement

net_alerter/net_alerter.py

@@ -6,7 +6,6 @@ Baseline on first run (silent). Alerts on join and rejoin.
 
 import json
 import os
-import re
 import socket
 import sqlite3
 import subprocess
@@ -140,74 +139,116 @@ def send_matrix_alert(msg: str) -> None:
         print(f"[error] Matrix send failed with HTTP {code}", file=sys.stderr)
 
 
-def get_local_subnets() -> list[str]:
+def read_arp_cache() -> list[dict]:
-    try:
+    """
-        out = subprocess.check_output(["ip", "-o", "-4", "addr", "show"], text=True)
+    Read kernel ARP cache from /proc/net/arp (passive, no active probing).
-    except Exception:
+    Falls back to 'ip neigh show' if /proc/net/arp unavailable.
-        return []
+    OUI vendor lookup from system hwdata if available.
-
+    Returns list of dicts: {"ip": str, "mac": str, "name": str}.
-    subnets = []
+    """
-    for line in out.splitlines():
-        parts = line.split()
-        if len(parts) < 4:
-            continue
-        iface = parts[1]
-        cidr = parts[3]
-        if any(iface.startswith(p) for p in ("lo", "tailscale", "wg", "tun", "docker", "br-")):
-            continue
-        subnets.append(cidr)
-    return subnets
-
-
-def scan_subnet(cidr: str) -> list[dict]:
-    try:
-        out = subprocess.check_output(
-            ["sudo", "nmap", "-sn", "-PR", "--host-timeout", "5s", cidr],
-            text=True, stderr=subprocess.DEVNULL
-        )
-    except Exception:
-        return []
-
     devices = []
-    current: dict = {}
-    for line in out.splitlines():
-        if line.startswith("Nmap scan report for"):
-            if current.get("mac"):
-                devices.append(current)
-            m = re.search(r"for (.+?)(?:\s+\((.+?)\))?$", line)
-            if m:
-                if m.group(2):
-                    current = {"name": m.group(1), "ip": m.group(2), "mac": ""}
-                else:
-                    current = {"name": "", "ip": m.group(1), "mac": ""}
-        elif "MAC Address:" in line:
-            m = re.search(r"MAC Address: ([0-9A-F:]{17})(?:\s+\((.+?)\))?", line)
-            if m:
-                current["mac"] = m.group(1).lower()
-                if m.group(2) and m.group(2) != "Unknown":
-                    current["vendor"] = m.group(2)
 
-    if current.get("mac"):
+    # Try /proc/net/arp first
-        devices.append(current)
+    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 out.splitlines():
+            fields = line.split()
+            if len(fields) < 5:
+                continue
+            # Format: <IP> dev <IFACE> lladdr <MAC> ...
+            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:
+        pass
 
     return devices
 
 
+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.
+    Returns vendor name or empty string if not found.
+    """
+    if not mac or mac == "00:00:00:00:00:00":
+        return ""
+
+    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()
+        except Exception:
+            continue
+
+    return ""
+
+
 def run() -> None:
     conn = get_db()
     ts = now_utc()
     hostname = socket.gethostname()
 
-    subnets = get_local_subnets()
+    # Read kernel ARP cache (passive, no active probing)
-    if not subnets:
+    arp_devices = read_arp_cache()
-        print("[warn] No local subnets found", file=sys.stderr)
-        conn.close()
-        return
-
-    # Collect all MACs visible right now across all subnets
     scan_results: dict[str, dict] = {}
-    for cidr in subnets:
+    for d in arp_devices:
-        for d in scan_subnet(cidr):
         scan_results[d["mac"]] = d
 
     # Is this the first run? (empty DB = baseline mode, no alerts)