diff --git a/net_alerter/net_alerter.py b/net_alerter/net_alerter.py index d55eaa2..ef2a587 100644 --- a/net_alerter/net_alerter.py +++ b/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,75 +139,117 @@ 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]: - try: - out = subprocess.check_output(["ip", "-o", "-4", "addr", "show"], text=True) - except Exception: - return [] - - 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 [] - +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 = [] - 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"): - devices.append(current) + # 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 out.splitlines(): + fields = line.split() + if len(fields) < 5: + continue + # Format: dev lladdr ... + 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() - if not subnets: - print("[warn] No local subnets found", file=sys.stderr) - conn.close() - return - - # Collect all MACs visible right now across all subnets + # Read kernel ARP cache (passive, no active probing) + arp_devices = read_arp_cache() scan_results: dict[str, dict] = {} - for cidr in subnets: - for d in scan_subnet(cidr): - scan_results[d["mac"]] = d + for d in arp_devices: + scan_results[d["mac"]] = d # Is this the first run? (empty DB = baseline mode, no alerts) row_count = conn.execute("SELECT COUNT(*) FROM devices").fetchone()[0]