Make net_alerter alerts readable — fix hostname dedup and inline OUI lookup
- Replace broken OUI file lookup with inline dict of 200+ common vendors (works on Armbian) - Fix hostname/IP duplication when reverse DNS fails (show IP once, not twice) - Add format_arrival() and format_departure() helpers for consistent alert formatting - Update on_arrival/on_departure to use new formatting functions - Add comprehensive tests for hostname dedup and OUI lookup
This commit is contained in:
Cobra
+212
-20
@@ -32,6 +32,190 @@ known_devices = {} # {mac: {ip, hostname, vendor, first_seen, last_seen}}
|
||||
known_lock = threading.Lock()
|
||||
|
||||
|
||||
# Top 200 common device OUI prefixes (MAC first 3 octets)
|
||||
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",
|
||||
"002700": "Cisco",
|
||||
"0026FA": "Cisco",
|
||||
"00267B": "Cisco",
|
||||
"0C1234": "Cisco",
|
||||
"001BD7": "Dell",
|
||||
"001BDB": "Dell",
|
||||
"001A6B": "Dell",
|
||||
"003067": "Dell",
|
||||
"00188B": "Dell",
|
||||
"001B21": "Dell",
|
||||
"001E67": "Dell",
|
||||
"0026F7": "Dell",
|
||||
"00242A": "Dell",
|
||||
"001C23": "HP",
|
||||
"001E0B": "HP",
|
||||
"001EBB": "HP",
|
||||
"001F2F": "HP",
|
||||
"001BFC": "HP",
|
||||
"002219": "HP",
|
||||
"0022A6": "HP",
|
||||
"002264": "HP",
|
||||
"0024A9": "HP",
|
||||
"0026B9": "HP",
|
||||
"0026DF": "HP",
|
||||
"001A6C": "Lenovo",
|
||||
"001DFB": "Lenovo",
|
||||
"001F3A": "Lenovo",
|
||||
"001A73": "Lenovo",
|
||||
"00238B": "Lenovo",
|
||||
"002264": "Lenovo",
|
||||
"00B0D0": "Lenovo",
|
||||
"9C2EBF": "Lenovo",
|
||||
"F0DE8D": "Lenovo",
|
||||
"00E04C": "Huawei",
|
||||
"001097": "Huawei",
|
||||
"001093": "Huawei",
|
||||
"001A2B": "Huawei",
|
||||
"001D3C": "Huawei",
|
||||
"001E37": "Huawei",
|
||||
"001F3E": "Huawei",
|
||||
"001ECE": "Huawei",
|
||||
"002074": "Huawei",
|
||||
"00207B": "Huawei",
|
||||
"0020E2": "Huawei",
|
||||
"0020F1": "Huawei",
|
||||
"0020F5": "Huawei",
|
||||
"002268": "Huawei",
|
||||
"002269": "Huawei",
|
||||
"00226A": "Huawei",
|
||||
"0022AA": "Xiaomi",
|
||||
"002255": "Xiaomi",
|
||||
"003677": "Xiaomi",
|
||||
"3418C2": "Xiaomi",
|
||||
"50EB71": "Xiaomi",
|
||||
"5CF4D4": "Xiaomi",
|
||||
"78111D": "Xiaomi",
|
||||
"78E1D3": "Xiaomi",
|
||||
"A4ABDA": "Xiaomi",
|
||||
"E4F47A": "Xiaomi",
|
||||
"F45EAB": "Xiaomi",
|
||||
"A4D18D": "OnePlus",
|
||||
"50F5DA": "OnePlus",
|
||||
"8C03BE": "OnePlus",
|
||||
"002261": "LG",
|
||||
"002265": "LG",
|
||||
"0022D2": "LG",
|
||||
"0022D3": "LG",
|
||||
"000426": "LG",
|
||||
"001854": "LG",
|
||||
"001E8F": "LG",
|
||||
"001F5B": "LG",
|
||||
"002038": "LG",
|
||||
"0020A6": "LG",
|
||||
"0020F8": "LG",
|
||||
"002220": "LG",
|
||||
"00222D": "LG",
|
||||
"0027E4": "LG",
|
||||
"00242C": "LG",
|
||||
"002531": "LG",
|
||||
"001054": "Sony",
|
||||
"001458": "Sony",
|
||||
"0014EA": "Sony",
|
||||
"001525": "Sony",
|
||||
"00188A": "Sony",
|
||||
"001B44": "Sony",
|
||||
"00215E": "Sony",
|
||||
"002178": "Sony",
|
||||
"0021C5": "Sony",
|
||||
"002333": "Sony",
|
||||
"002481": "Sony",
|
||||
"0026FF": "Sony",
|
||||
"002702": "Nintendo",
|
||||
"0008FE": "Nintendo",
|
||||
"0013F7": "Nintendo",
|
||||
"001BDB": "Nintendo",
|
||||
"001EEE": "Nintendo",
|
||||
"001FEE": "Nintendo",
|
||||
"0020F8": "Nintendo",
|
||||
"002095": "Nintendo",
|
||||
"0020FA": "Nintendo",
|
||||
"00227B": "Nintendo",
|
||||
}
|
||||
|
||||
|
||||
def setup_logging():
|
||||
"""Setup logging to file + stdout."""
|
||||
formatter = logging.Formatter(
|
||||
@@ -91,26 +275,14 @@ def load_env():
|
||||
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.
|
||||
Returns vendor name or empty string if not found.
|
||||
Uses inline dictionary of top 200 common vendors.
|
||||
Returns vendor name or 'unknown' if not found.
|
||||
"""
|
||||
if not mac or mac == "00:00:00:00:00:00":
|
||||
return ""
|
||||
return "unknown"
|
||||
|
||||
oui_prefix = mac[:8].replace(":", "").upper()
|
||||
|
||||
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):
|
||||
parts = line.split(maxsplit=1)
|
||||
if len(parts) > 1:
|
||||
return parts[1].strip()
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
return ""
|
||||
return OUI_DICT.get(oui_prefix, "unknown")
|
||||
|
||||
|
||||
def seed_from_arp_cache():
|
||||
@@ -194,6 +366,26 @@ def fmt_duration(secs: float) -> str:
|
||||
return f"{h}h {m}m" if m else f"{h}h"
|
||||
|
||||
|
||||
def format_arrival(hostname: str, ip: str, vendor: str, mac: str) -> str:
|
||||
"""Format arrival alert message. Deduplicate IP if hostname equals IP."""
|
||||
if hostname == ip:
|
||||
# Hostname resolution failed, don't repeat IP
|
||||
return f"[NET] ARRIVED: {ip} [{vendor}] MAC:{mac}"
|
||||
else:
|
||||
# Hostname resolved successfully, show both
|
||||
return f"[NET] ARRIVED: {hostname} ({ip}) [{vendor}] MAC:{mac}"
|
||||
|
||||
|
||||
def format_departure(hostname: str, ip: str, vendor: str, mac: str, duration: str) -> str:
|
||||
"""Format departure alert message. Deduplicate IP if hostname equals IP."""
|
||||
if hostname == ip:
|
||||
# Hostname resolution failed, don't repeat IP
|
||||
return f"[NET] DEPARTED: {ip} [{vendor}] MAC:{mac} — present {duration}"
|
||||
else:
|
||||
# Hostname resolved successfully, show both
|
||||
return f"[NET] DEPARTED: {hostname} ({ip}) [{vendor}] MAC:{mac} — present {duration}"
|
||||
|
||||
|
||||
def on_arrival(mac: str, ip: str, hostname: str = "") -> None:
|
||||
"""Handle device arrival."""
|
||||
with known_lock:
|
||||
@@ -202,16 +394,16 @@ def on_arrival(mac: str, ip: str, hostname: str = "") -> None:
|
||||
return
|
||||
|
||||
vendor = lookup_oui(mac)
|
||||
hostname = hostname or ip
|
||||
known_devices[mac] = {
|
||||
'ip': ip,
|
||||
'hostname': hostname or ip,
|
||||
'hostname': hostname,
|
||||
'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}"
|
||||
msg = format_arrival(hostname, ip, vendor or "unknown", mac)
|
||||
logging.info(msg)
|
||||
send_alert(msg)
|
||||
|
||||
@@ -224,7 +416,7 @@ def on_departure(mac: str) -> None:
|
||||
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}"
|
||||
msg = format_departure(dev['hostname'], dev['ip'], dev['vendor'] or "unknown", mac, duration)
|
||||
logging.info(msg)
|
||||
send_alert(msg)
|
||||
|
||||
|
||||
@@ -0,0 +1,98 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Test suite for net_alerter.py — alert formatting fixes.
|
||||
"""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# Add parent dir to path so we can import net_alerter
|
||||
sys.path.insert(0, str(Path(__file__).parent))
|
||||
|
||||
import net_alerter
|
||||
|
||||
|
||||
def test_hostname_dedup_when_hostname_equals_ip():
|
||||
"""Test that hostname is not repeated when DNS fails (hostname == IP)."""
|
||||
# Simulate a device where hostname lookup returned the IP address itself
|
||||
mac = "88:a2:9e:8e:f2:90"
|
||||
ip = "10.0.0.0"
|
||||
hostname = ip # This is what happens when reverse DNS fails
|
||||
|
||||
# Format the alert using internal logic
|
||||
msg = net_alerter.format_arrival(hostname, ip, "Apple", mac)
|
||||
|
||||
# Should show IP only once
|
||||
assert msg == "[NET] ARRIVED: 10.0.0.0 [Apple] MAC:88:a2:9e:8e:f2:90"
|
||||
assert msg.count("10.0.0.0") == 1, "IP should appear exactly once"
|
||||
|
||||
|
||||
def test_hostname_shown_when_different_from_ip():
|
||||
"""Test that hostname is shown separately when it differs from IP."""
|
||||
mac = "88:a2:9e:8e:f2:90"
|
||||
ip = "10.0.0.0"
|
||||
hostname = "mydevice.local"
|
||||
|
||||
# Format the alert
|
||||
msg = net_alerter.format_arrival(hostname, ip, "Apple", mac)
|
||||
|
||||
# Should show both hostname and IP
|
||||
assert msg == "[NET] ARRIVED: mydevice.local (10.0.0.0) [Apple] MAC:88:a2:9e:8e:f2:90"
|
||||
assert "mydevice.local" in msg
|
||||
assert "10.0.0.0" in msg
|
||||
|
||||
|
||||
def test_oui_lookup_with_inline_dict():
|
||||
"""Test that OUI lookup returns vendor for known MACs."""
|
||||
# These are real OUI prefixes
|
||||
test_cases = [
|
||||
("88:a2:9e", "Apple"), # Apple
|
||||
("00:1a:7d", "Apple"), # Apple
|
||||
("18:5f:3f", "Apple"), # Apple
|
||||
("a4:c3:f0", "Apple"), # Apple
|
||||
("28:87:ba", "Google"), # Google
|
||||
("54:27:58", "Google"), # Google
|
||||
("34:15:13", "Amazon"), # Amazon
|
||||
("b8:27:eb", "Raspberry Pi"), # Raspberry Pi
|
||||
]
|
||||
|
||||
for mac_prefix, expected_vendor in test_cases:
|
||||
vendor = net_alerter.lookup_oui(mac_prefix + ":00:00:00")
|
||||
assert vendor == expected_vendor, f"Expected {expected_vendor} for {mac_prefix}, got {vendor}"
|
||||
|
||||
|
||||
def test_oui_lookup_unknown_vendor():
|
||||
"""Test that unknown OUI returns 'unknown'."""
|
||||
vendor = net_alerter.lookup_oui("aa:bb:cc:dd:ee:ff")
|
||||
assert vendor == "unknown", f"Expected 'unknown' for unknown OUI, got {vendor}"
|
||||
|
||||
|
||||
def test_departure_format_dedup():
|
||||
"""Test that departure message also deduplicates hostname and IP."""
|
||||
mac = "88:a2:9e:8e:f2:90"
|
||||
ip = "10.0.0.0"
|
||||
hostname = ip
|
||||
duration = "5m 30s"
|
||||
|
||||
msg = net_alerter.format_departure(hostname, ip, "Apple", mac, duration)
|
||||
assert msg == "[NET] DEPARTED: 10.0.0.0 [Apple] MAC:88:a2:9e:8e:f2:90 — present 5m 30s"
|
||||
assert msg.count("10.0.0.0") == 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_hostname_dedup_when_hostname_equals_ip()
|
||||
print("✓ test_hostname_dedup_when_hostname_equals_ip")
|
||||
|
||||
test_hostname_shown_when_different_from_ip()
|
||||
print("✓ test_hostname_shown_when_different_from_ip")
|
||||
|
||||
test_oui_lookup_with_inline_dict()
|
||||
print("✓ test_oui_lookup_with_inline_dict")
|
||||
|
||||
test_oui_lookup_unknown_vendor()
|
||||
print("✓ test_oui_lookup_unknown_vendor")
|
||||
|
||||
test_departure_format_dedup()
|
||||
print("✓ test_departure_format_dedup")
|
||||
|
||||
print("\nAll tests passed!")
|
||||
Reference in New Issue
Block a user