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
This commit is contained in:
Cobra
+99
-58
@@ -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: <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()
|
||||
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]
|
||||
|
||||
Reference in New Issue
Block a user