From ab0688e1b3342c9c1c8446aa5d282e8614ce217c Mon Sep 17 00:00:00 2001 From: Cobra Date: Thu, 9 Apr 2026 13:05:18 -0400 Subject: [PATCH] Add SQLite OUI database lookup to net_alerter - lookup_oui() now tries SQLite first (/opt/net_alerter/oui.db) - Falls back to inline OUI_DICT if DB missing or query fails - No network calls, local SQLite reads only - Debug-level logging for failures (no output noise) --- net_alerter/net_alerter.py | 23 ++++++++++++++++++++--- 1 file changed, 20 insertions(+), 3 deletions(-) diff --git a/net_alerter/net_alerter.py b/net_alerter/net_alerter.py index cc5da06..6723265 100644 --- a/net_alerter/net_alerter.py +++ b/net_alerter/net_alerter.py @@ -20,6 +20,7 @@ import threading import time import urllib.parse import urllib.request +import sqlite3 from pathlib import Path # --- Config (from .env or env vars) --- @@ -27,6 +28,7 @@ MATRIX_HOMESERVER = "" MATRIX_ACCESS_TOKEN = "" MATRIX_ROOM_ID = "" LOG_FILE = "/opt/net_alerter/net_alerter.log" +OUI_DB_PATH = "/opt/net_alerter/oui.db" known_devices = {} # {mac: {ip, hostname, vendor, first_seen, last_seen, infrastructure (opt)}} known_lock = threading.Lock() @@ -279,14 +281,29 @@ def load_env(): def lookup_oui(mac: str) -> str: """ Look up OUI vendor from MAC address (first 3 octets). - Uses inline dictionary of top 200 common vendors. + Tries SQLite database first, falls back to inline dictionary. Returns vendor name or 'unknown' if not found. """ if not mac or mac == "00:00:00:00:00:00": return "unknown" - oui_prefix = mac[:8].replace(":", "").upper() - return OUI_DICT.get(oui_prefix, "unknown") + # Try database first + oui_prefix = mac[:8].lower() + try: + conn = sqlite3.connect(OUI_DB_PATH, timeout=1) + conn.row_factory = sqlite3.Row + cursor = conn.cursor() + result = cursor.execute("SELECT vendor FROM oui WHERE prefix = ?", (oui_prefix,)).fetchone() + conn.close() + if result: + return result['vendor'] + except Exception: + # Log at DEBUG only, no output noise + logging.debug(f"OUI database lookup failed for {oui_prefix}", exc_info=False) + + # Fall back to inline dictionary + oui_prefix_upper = oui_prefix.replace(":", "").upper() + return OUI_DICT.get(oui_prefix_upper, "unknown") def seed_from_arp_cache():