From acdebf568ee575d011dafdadd1b10b37e4ab82dd Mon Sep 17 00:00:00 2001 From: pavel Date: Thu, 30 Oct 2025 07:02:31 +0200 Subject: [PATCH] feat: startign with netbox 07:02 --- files/nats_registration_listener.py | 225 ++++++++++++++++------------ 1 file changed, 133 insertions(+), 92 deletions(-) diff --git a/files/nats_registration_listener.py b/files/nats_registration_listener.py index 064c77a..a145245 100644 --- a/files/nats_registration_listener.py +++ b/files/nats_registration_listener.py @@ -1,22 +1,32 @@ #!/usr/bin/env python3 """ -NATS Registration Listener (with optional NetBox hostname lookup for fox100) ----------------------------------------------------------------------------- - -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) +NATS Registration Listener (with NetBox hostname lookup for fox100, no extra deps) +---------------------------------------------------------------------------------- Behavior: - - Non-fox100: unchanged. - - fox100: best-effort NetBox lookup inline (sequential), stderr on edge cases. + - Prints exactly one line per message: + [TS] product= mac= fw= [host= for fox100] + - Listener-only (no publish/reply). + - Payload-byte strict dedupe within a TTL window. + - Graceful shutdown + reconnect logging. + - For product=fox100: best-effort NetBox lookup by MAC -> interface -> device.name. + (Inline, sequential "as-is" call; no concurrency / worker pool.) -Install deps: - pip install --upgrade nats-py aiohttp +NetBox (hardcoded as requested): + - Edit NB_TOKEN below. + - Uses stdlib urllib (no 'requests' / 'aiohttp'). + +CLI (same spirit as before; your entrypoint passes these): + --servers nats://host:4222 [one or more] + --subject v1.registration + --queue (optional) + --name registration-listener + --tls-ca /path/to/ca.pem + --tls-cert/--tls-key/--insecure + --include-subject + +Install deps in image: + pip install --no-cache-dir --upgrade nats-py """ import argparse @@ -30,27 +40,38 @@ import hashlib from time import monotonic from datetime import datetime, timezone from typing import Optional, Dict, Any +from urllib.parse import urlencode +from urllib.request import Request, urlopen +from urllib.error import URLError, HTTPError import nats -import aiohttp -from aiohttp import ClientTimeout -# ----------------- existing helpers (unchanged) ----------------- +# ========================= +# NetBox hardcoded config +# ========================= +NB_URL = "http://netbox.gt-tiso.ikeja.co.za" # Base URL +NB_TOKEN = "CHANGE_ME_NETBOX_TOKEN" # <-- paste the real token here +NB_TIMEOUT = 3.0 # seconds per HTTP GET + + +# ========================= +# Arg parsing (mirrors original) +# ========================= def parse_args(): p = argparse.ArgumentParser(description="Listen to a NATS subject and print one line per device.") p.add_argument( "--servers", nargs="+", help="One or more NATS server URLs (e.g., nats://127.0.0.1:4222).", - default=[os.environ.get("NATS_URL", "nats://127.0.0.1:4222")], + default=["nats://127.0.0.1:4222"], ) p.add_argument( "--subject", help="NATS subject to subscribe to.", - default=os.environ.get("NATS_SUBJECT", "registration"), + default="registration", ) - p.add_argument("--queue", help="Optional queue group name.", default=os.environ.get("NATS_QUEUE")) + p.add_argument("--queue", help="Optional queue group name.", default=None) p.add_argument("--name", help="Client connection name.", default="registration-listener") # Auth @@ -67,29 +88,32 @@ 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() +# ========================= +# Helpers +# ========================= 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 def ts() -> str: + # local time with offset like your previous output return datetime.now(timezone.utc).astimezone().strftime("%Y-%m-%d %H:%M:%S%z") @@ -103,17 +127,29 @@ 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: 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", {}) @@ -123,10 +159,13 @@ def extract_fields(obj: Dict[str, Any]): mac = chosen["macAddress"] except Exception: pass + return product, mac, fw_active -# ----------------- new: NetBox helpers ----------------- +# ========================= +# NetBox lookup (stdlib urllib) +# ========================= def normalize_mac(mac: str) -> Optional[str]: """ Return lowercase colon-separated MAC or None if invalid-ish. @@ -134,84 +173,98 @@ def normalize_mac(mac: str) -> Optional[str]: """ 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') + s = mac.strip().lower().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 ":".join(hex_only[i:i+2] for i in range(0, 12, 2)) + 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): +def http_get_json(url: str, params: Optional[Dict[str, str]] = None, headers: Optional[Dict[str, str]] = None, timeout: float = NB_TIMEOUT): + if params: + url = f"{url}?{urlencode(params)}" + req = Request(url, headers=headers or {}, method="GET") + try: + with urlopen(req, timeout=timeout) as resp: + if resp.status != 200: + return None, resp.status + data = resp.read() + try: + return json.loads(data.decode("utf-8", errors="replace")), 200 + except Exception: + return None, 200 + except HTTPError as e: + return None, getattr(e, "code", 599) + except URLError: + return None, 598 + except Exception: + return None, 597 + + +def nb_get_hostname_by_mac(mac: str, log_status) -> Optional[str]: """ - Best-effort: + Steps: 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). + 2) If assigned to dcim.interface -> GET /api/dcim/interfaces/{id}/ + 3) Return device.name 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}") + # invalid or missing mac; keep stdout clean return None + base = NB_URL.rstrip("/") + h = { + "Accept": "application/json", + "Content-Type": "application/json", + "Authorization": f"Token {NB_TOKEN}", + } + # 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}") + data, code = http_get_json(f"{base}/api/dcim/mac-addresses/", params={"mac_address": mac_norm, "limit": "2"}, headers=h) + if code != 200 or not data: + # network/HTTP issue or parse fail + asyncio.create_task(log_status(f"[{ts()}] nb: mac query http={code} mac={mac_norm}")) return None results = (data or {}).get("results") or [] if not results: - await log_status(f"[{ts()}] nb: mac not found mac={mac_norm}") + asyncio.create_task(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}") + asyncio.create_task(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}") + asyncio.create_task(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}") + asyncio.create_task(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}") + iface, code2 = http_get_json(f"{base}/api/dcim/interfaces/{aoid}/", headers=h) + if code2 != 200 or not iface: + asyncio.create_task(log_status(f"[{ts()}] nb: iface fetch http={code2} iface_id={aoid}")) 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}") + asyncio.create_task(log_status(f"[{ts()}] nb: iface has no device iface_id={aoid}")) return None return host -# ----------------- main ----------------- +# ========================= +# Main +# ========================= async def main(): args = parse_args() ssl_ctx = make_ssl_context(args) @@ -222,9 +275,10 @@ async def main(): async with print_lock: print(s, file=sys.stderr, flush=True) - # dedupe - DEDUPE_TTL = float(os.environ.get("DEDUP_TTL", "2.0")) + # --- Strict read-only dedupe (payload bytes) --- + DEDUPE_TTL = float(os.environ.get("DEDUP_TTL", "2.0")) # you can tune via env if you want; otherwise 2.0s recent_payloads: Dict[bytes, float] = {} + reconnect_time_wait = 2 max_reconnect_attempts = -1 @@ -257,20 +311,11 @@ 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 - # dedupe by payload + # --- dedupe by exact payload bytes --- digest = hashlib.blake2b(payload, digest_size=16).digest() nowm = monotonic() exp = recent_payloads.get(digest) @@ -283,7 +328,7 @@ async def main(): if recent_payloads.get(k, 0) <= cutoff: recent_payloads.pop(k, None) - # parse payload + # parse product = mac = fw = "-" try: text = payload.decode("utf-8", errors="replace") @@ -292,24 +337,20 @@ async def main(): except Exception: pass - # optional NetBox lookup for fox100 + # optional NetBox lookup for fox100 (inline, sequential) host_suffix = "" - if nb_enabled and product == "fox100": + if 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, - ) + host = nb_get_hostname_by_mac(mac=mac, log_status=log_status) if host: host_suffix = f" host={host}" + else: + # minimal diag already sent to stderr by helper + pass except Exception as e: await log_status(f"[{ts()}] nb: unexpected error mac={mac!r} err={e!r}") - # print one line + # print one compact line line = f"[{now_s}] product={product} mac={mac} fw={fw}{host_suffix}" if args.include_subject: line += f" subject={msg.subject}" @@ -324,7 +365,8 @@ async def main(): await nc.subscribe(args.subject, cb=message_handler) await log_status( - 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'}" + 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 nb=on" ) # graceful shutdown @@ -346,8 +388,7 @@ async def main(): await nc.drain() finally: await nc.close() - if session: - await session.close() + if __name__ == "__main__": try: