diff --git a/files/nats_registration_listener.py b/files/nats_registration_listener.py index d75178f..064c77a 100644 --- a/files/nats_registration_listener.py +++ b/files/nats_registration_listener.py @@ -1,17 +1,22 @@ #!/usr/bin/env python3 """ -NATS Registration Listener (compact one-line output + strict read-only dedupe) ------------------------------------------------------------------------------- -Print exactly one line per message: - [TS] product= mac= fw= +NATS Registration Listener (with optional NetBox hostname lookup for fox100) +---------------------------------------------------------------------------- -- Listener-only (does NOT publish/reply). -- Writes are serialized to avoid interleaved output. -- Strict dedupe by exact payload bytes within a short TTL (env DEDUP_TTL, default 2.0s). -- Falls back gracefully if fields/structure are missing. +Adds: For product=fox100, lookup device hostname in NetBox by MAC and append + 'host=' to the one-line output if found. + +Config (env or CLI): + NB_URL (e.g., http://netbox.gt-tiso.ikeja.co.za) + NB_TOKEN (API token) + NB_TIMEOUT (seconds, default 3.0) + +Behavior: + - Non-fox100: unchanged. + - fox100: best-effort NetBox lookup inline (sequential), stderr on edge cases. Install deps: - pip install --upgrade nats-py + pip install --upgrade nats-py aiohttp """ import argparse @@ -27,7 +32,10 @@ from datetime import datetime, timezone from typing import Optional, Dict, Any import nats +import aiohttp +from aiohttp import ClientTimeout +# ----------------- existing helpers (unchanged) ----------------- def parse_args(): p = argparse.ArgumentParser(description="Listen to a NATS subject and print one line per device.") @@ -59,24 +67,25 @@ def parse_args(): # Misc p.add_argument("--include-subject", action="store_true", help="Append subject=... to each output line.") + + # --- NetBox cfg --- + p.add_argument("--nb-url", default=os.environ.get("NB_URL"), help="NetBox base URL (env NB_URL).") + p.add_argument("--nb-token", default=os.environ.get("NB_TOKEN"), help="NetBox API token (env NB_TOKEN).") + p.add_argument("--nb-timeout", type=float, default=float(os.environ.get("NB_TIMEOUT", "3.0")), help="NetBox HTTP timeout seconds (default 3.0).") return p.parse_args() def make_ssl_context(args) -> Optional[ssl.SSLContext]: if not any([args.tls_ca, args.tls_cert, args.tls_key]) and not any(url.startswith("tls://") for url in args.servers): return None - ctx = ssl.create_default_context(purpose=ssl.Purpose.SERVER_AUTH) if args.tls_ca: ctx.load_verify_locations(args.tls_ca) - if args.tls_cert and args.tls_key: ctx.load_cert_chain(certfile=args.tls_cert, keyfile=args.tls_key) - if args.insecure: ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE - return ctx @@ -85,7 +94,6 @@ def ts() -> str: def pick_first_interface(eths: Dict[str, Any]) -> Optional[Dict[str, Any]]: - # Prefer eth0; otherwise first key deterministically (sorted) if "eth0" in eths and isinstance(eths["eth0"], dict): return eths["eth0"] for name in sorted(eths.keys()): @@ -95,30 +103,17 @@ def pick_first_interface(eths: Dict[str, Any]) -> Optional[Dict[str, Any]]: def extract_fields(obj: Dict[str, Any]): - """ - Expected structure (from examples): - { "data": { - "productName": "...", - "ethernetInterfaces": { "eth0": { "macAddress": "..." } }, - "firmwareVersion": { "active": "..." } - }, - "metadata": { "type": "REGISTRATION_REQUEST" } - } - """ root = obj if isinstance(root.get("data"), dict): d = root["data"] else: - # Some producers might not wrap with "data" d = root - product = d.get("productName") or "-" fw_active = "-" try: fw_active = d.get("firmwareVersion", {}).get("active") or "-" except Exception: pass - mac = "-" try: eths = d.get("ethernetInterfaces", {}) @@ -128,26 +123,108 @@ def extract_fields(obj: Dict[str, Any]): mac = chosen["macAddress"] except Exception: pass - return product, mac, fw_active +# ----------------- new: NetBox helpers ----------------- + +def normalize_mac(mac: str) -> Optional[str]: + """ + Return lowercase colon-separated MAC or None if invalid-ish. + Accepts colon, dash, or bare hex and tries to normalize. + """ + if not mac or not isinstance(mac, str): + return None + s = mac.strip().lower() + # replace dashes with colons + s = s.replace('-', ':') + hex_only = ''.join(ch for ch in s if ch in '0123456789abcdef') + if len(hex_only) == 12: + # reconstruct as colon-separated + s = ':'.join(hex_only[i:i+2] for i in range(0, 12, 2)) + return s + # already colonized? + parts = s.split(':') + if len(parts) == 6 and all(len(p) == 2 and all(c in '0123456789abcdef' for c in p) for p in parts): + return s + return None + + +async def nb_get_hostname_by_mac(session: aiohttp.ClientSession, base_url: str, token: str, mac: str, timeout_s: float, log_status): + """ + Best-effort: + 1) GET /api/dcim/mac-addresses/?mac_address= + 2) If one result and assigned to interface -> GET /api/dcim/interfaces/{id}/ -> device.name + Returns (hostname or None). + Emits concise diagnostics to stderr via log_status on edge cases. + """ + mac_norm = normalize_mac(mac) + if not mac_norm: + await log_status(f"[{ts()}] nb: invalid mac format mac={mac!r}") + return None + + # 1) find MAC record(s) + url_mac = f"{base_url.rstrip('/')}/api/dcim/mac-addresses/" + try: + async with session.get(url_mac, params={"mac_address": mac_norm, "limit": 2}, headers={"Authorization": f"Token {token}"}, timeout=timeout_s) as r: + if r.status != 200: + await log_status(f"[{ts()}] nb: mac query http={r.status} mac={mac_norm}") + return None + data = await r.json() + except Exception as e: + await log_status(f"[{ts()}] nb: mac query error mac={mac_norm} err={e!r}") + return None + + results = (data or {}).get("results") or [] + if not results: + await log_status(f"[{ts()}] nb: mac not found mac={mac_norm}") + return None + if len(results) > 1: + await log_status(f"[{ts()}] nb: multiple mac records mac={mac_norm}") + + rec = results[0] + aot = (rec.get("assigned_object_type") or "").strip() + aoid = rec.get("assigned_object_id") + if not aot or aoid is None: + await log_status(f"[{ts()}] nb: mac unassigned mac={mac_norm}") + return None + if aot != "dcim.interface": + await log_status(f"[{ts()}] nb: mac assigned to {aot} mac={mac_norm}") + return None + + # 2) fetch interface -> device + url_iface = f"{base_url.rstrip('/')}/api/dcim/interfaces/{aoid}/" + try: + async with session.get(url_iface, headers={"Authorization": f"Token {token}"}, timeout=timeout_s) as r2: + if r2.status != 200: + await log_status(f"[{ts()}] nb: iface fetch http={r2.status} iface_id={aoid}") + return None + iface = await r2.json() + except Exception as e: + await log_status(f"[{ts()}] nb: iface fetch error iface_id={aoid} err={e!r}") + return None + + dev = iface.get("device") or {} + host = dev.get("name") or dev.get("display") + if not host: + await log_status(f"[{ts()}] nb: iface has no device iface_id={aoid}") + return None + return host + +# ----------------- main ----------------- async def main(): args = parse_args() ssl_ctx = make_ssl_context(args) - # Serialize stdout writes to avoid interleaving print_lock = asyncio.Lock() async def log_status(s: str): async with print_lock: print(s, file=sys.stderr, flush=True) - # --- Strict read-only dedupe config (env) --- - # Drop exact duplicate payloads seen within this TTL window. + # dedupe DEDUPE_TTL = float(os.environ.get("DEDUP_TTL", "2.0")) - recent_payloads: Dict[bytes, float] = {} # digest -> expires_at (monotonic seconds) - + recent_payloads: Dict[bytes, float] = {} reconnect_time_wait = 2 max_reconnect_attempts = -1 @@ -180,29 +257,33 @@ async def main(): tls=ssl_ctx, ) + # --- aiohttp session for NetBox --- + nb_enabled = bool(args.nb_url and args.nb_token) + if nb_enabled: + timeout = ClientTimeout(total=None) # per-call timeout set explicitly + http = aiohttp.TCPConnector(limit=32) + session = aiohttp.ClientSession(timeout=timeout, connector=http) + else: + session = None + async def message_handler(msg: nats.aio.msg.Msg): now_s = ts() payload = msg.data - # --- STRICT DEDUPE by exact payload bytes (read-only) --- - # Use a stable, compact digest to key the recent map. + # dedupe by payload digest = hashlib.blake2b(payload, digest_size=16).digest() nowm = monotonic() exp = recent_payloads.get(digest) if exp and exp > nowm: - return # drop exact duplicate seen very recently + return recent_payloads[digest] = nowm + DEDUPE_TTL - - # Optional light cleanup to keep the dict bounded if len(recent_payloads) > 4096: - # remove expired entries cutoff = nowm - recent_payloads_keys = list(recent_payloads.keys()) - for k in recent_payloads_keys: + for k in list(recent_payloads.keys()): if recent_payloads.get(k, 0) <= cutoff: recent_payloads.pop(k, None) - # Parse and print one line + # parse payload product = mac = fw = "-" try: text = payload.decode("utf-8", errors="replace") @@ -211,25 +292,42 @@ async def main(): except Exception: pass - line = f"[{now_s}] product={product} mac={mac} fw={fw}" + # optional NetBox lookup for fox100 + host_suffix = "" + if nb_enabled and product == "fox100": + try: + host = await nb_get_hostname_by_mac( + session=session, + base_url=args.nb_url, + token=args.nb_token, + mac=mac, + timeout_s=args.nb_timeout, + log_status=log_status, + ) + if host: + host_suffix = f" host={host}" + except Exception as e: + await log_status(f"[{ts()}] nb: unexpected error mac={mac!r} err={e!r}") + + # print one line + line = f"[{now_s}] product={product} mac={mac} fw={fw}{host_suffix}" if args.include_subject: line += f" subject={msg.subject}" - async with print_lock: sys.stdout.write(line + "\n") sys.stdout.flush() + # subscribe if args.queue: await nc.subscribe(args.subject, queue=args.queue, cb=message_handler) else: await nc.subscribe(args.subject, cb=message_handler) await log_status( - f"[{ts()}] Listening on subject '{args.subject}' (queue={getattr(args, 'queue', None) or '-'}) via {args.servers} " - f"| mode=one-line dedupe=payload ttl={DEDUPE_TTL}s tx=disabled" + f"[{ts()}] Listening on '{args.subject}' (queue={getattr(args, 'queue', None) or '-'}) via {args.servers} | mode=one-line dedupe=payload ttl={DEDUPE_TTL}s tx=disabled nb={'on' if nb_enabled else 'off'}" ) - # Graceful shutdown + # graceful shutdown stop_event = asyncio.Event() def handle_signal(*_): @@ -248,7 +346,8 @@ async def main(): await nc.drain() finally: await nc.close() - + if session: + await session.close() if __name__ == "__main__": try: