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()
|
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():
|
def setup_logging():
|
||||||
"""Setup logging to file + stdout."""
|
"""Setup logging to file + stdout."""
|
||||||
formatter = logging.Formatter(
|
formatter = logging.Formatter(
|
||||||
@@ -91,26 +275,14 @@ def load_env():
|
|||||||
def lookup_oui(mac: str) -> str:
|
def lookup_oui(mac: str) -> str:
|
||||||
"""
|
"""
|
||||||
Look up OUI vendor from MAC address (first 3 octets).
|
Look up OUI vendor from MAC address (first 3 octets).
|
||||||
Searches /usr/share/hwdata/oui.txt or /usr/share/misc/oui.txt.
|
Uses inline dictionary of top 200 common vendors.
|
||||||
Returns vendor name or empty string if not found.
|
Returns vendor name or 'unknown' if not found.
|
||||||
"""
|
"""
|
||||||
if not mac or mac == "00:00:00:00:00:00":
|
if not mac or mac == "00:00:00:00:00:00":
|
||||||
return ""
|
return "unknown"
|
||||||
|
|
||||||
oui_prefix = mac[:8].replace(":", "").upper()
|
oui_prefix = mac[:8].replace(":", "").upper()
|
||||||
|
return OUI_DICT.get(oui_prefix, "unknown")
|
||||||
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 ""
|
|
||||||
|
|
||||||
|
|
||||||
def seed_from_arp_cache():
|
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"
|
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:
|
def on_arrival(mac: str, ip: str, hostname: str = "") -> None:
|
||||||
"""Handle device arrival."""
|
"""Handle device arrival."""
|
||||||
with known_lock:
|
with known_lock:
|
||||||
@@ -202,16 +394,16 @@ def on_arrival(mac: str, ip: str, hostname: str = "") -> None:
|
|||||||
return
|
return
|
||||||
|
|
||||||
vendor = lookup_oui(mac)
|
vendor = lookup_oui(mac)
|
||||||
|
hostname = hostname or ip
|
||||||
known_devices[mac] = {
|
known_devices[mac] = {
|
||||||
'ip': ip,
|
'ip': ip,
|
||||||
'hostname': hostname or ip,
|
'hostname': hostname,
|
||||||
'vendor': vendor,
|
'vendor': vendor,
|
||||||
'first_seen': time.time(),
|
'first_seen': time.time(),
|
||||||
'last_seen': time.time()
|
'last_seen': time.time()
|
||||||
}
|
}
|
||||||
|
|
||||||
label = hostname or ip
|
msg = format_arrival(hostname, ip, vendor or "unknown", mac)
|
||||||
msg = f"[NET] ARRIVED: {label} ({ip}) [{vendor or 'unknown'}] MAC:{mac}"
|
|
||||||
logging.info(msg)
|
logging.info(msg)
|
||||||
send_alert(msg)
|
send_alert(msg)
|
||||||
|
|
||||||
@@ -224,7 +416,7 @@ def on_departure(mac: str) -> None:
|
|||||||
dev = known_devices.pop(mac)
|
dev = known_devices.pop(mac)
|
||||||
|
|
||||||
duration = fmt_duration(time.time() - dev['first_seen'])
|
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)
|
logging.info(msg)
|
||||||
send_alert(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