#!/usr/bin/env python3 """NetOps Widgets - a local widget dashboard for the jobs that eat an afternoon. python3 netops-widgets.py opens a small web UI on 127.0.0.1 and binds nothing else. Every call goes from YOUR machine to YOUR controllers with YOUR credentials. ciscotools.dev never touches a device, never sees a credential, and never proxies a connection: it serves this file, its docs, and a license check that carries your API key and this app's version and nothing else. Lite (free, no key needed) * Catalyst Center device picker (Intent API v1, token auth, read only). * Port status for a selected device. * Find a MAC address across the fabric. * SecureCRT session export from the Catalyst Center inventory (zip of .ini files plus sessions.csv). Pro (unlocked with your ciscotools.dev API key) * Cisco ISE endpoint push for MAB: endpoint groups pulled from the API, MAC normalize and dedupe, create-or-update per endpoint, and the bulk submit path once the list reaches 20. Every MAC is looked up first and reported by identity group NAME, not by ISE's internal group UUID, and a push that would move a MAC between groups names both of them. * Windows DHCP reservations over SSH plus PowerShell, with every argument validated against an allowlist before it is composed into a command. * An audit log line for every real write. Safety rails, always on: * DRY RUN IS ON BY DEFAULT. A dry run composes the request and shows it to you; it sends nothing. * A real write requires an explicit confirm flag in the same call. * Every real write appends one JSON line to the audit log, after it succeeds, never before. * Passwords are never logged, never written to the audit log, and never placed on a command line. Requirements: Python 3.9 or newer, plus the system `ssh` client for the DHCP widget. No pip, no venv, no third-party packages. Tested on 3.9 and 3.13. Copyright (c) 2026 Brody Networks. Provided as-is, no warranty. You are responsible for what you run against your own network. """ from __future__ import annotations APP_NAME = "netops-widgets" APP_TITLE = "NetOps Widgets" APP_VERSION = "1.0.1" import argparse import base64 import csv import datetime import getpass import io import json import os import re import secrets import shutil import ssl import subprocess import sys import threading import time import urllib.error import urllib.parse import urllib.request import webbrowser import zipfile from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer # --- BEGIN ED25519 SHARED CORE (verbatim from backend/services/ed25519_pure.py; sync with backend/scripts/sync_ed25519_core.py) --- import hashlib _P = 2 ** 255 - 19 _L = 2 ** 252 + 27742317777372353535851937790883648493 def _sha512(data): return hashlib.sha512(data).digest() def _inv(x): return pow(x, _P - 2, _P) _D = -121665 * _inv(121666) % _P _SQRT_M1 = pow(2, (_P - 1) // 4, _P) def _recover_x(y, sign_bit): if y >= _P: return None x2 = (y * y - 1) * _inv(_D * y * y + 1) % _P if x2 == 0: return None if sign_bit else 0 x = pow(x2, (_P + 3) // 8, _P) if (x * x - x2) % _P != 0: x = x * _SQRT_M1 % _P if (x * x - x2) % _P != 0: return None if (x & 1) != sign_bit: x = _P - x return x _G_Y = 4 * _inv(5) % _P _G_X = _recover_x(_G_Y, 0) # Extended homogeneous coordinates: (X, Y, Z, T) with x = X/Z, y = Y/Z, xy = T/Z. _G = (_G_X, _G_Y, 1, _G_X * _G_Y % _P) _IDENTITY = (0, 1, 1, 0) def _point_add(p, q): a = (p[1] - p[0]) * (q[1] - q[0]) % _P b = (p[1] + p[0]) * (q[1] + q[0]) % _P c = 2 * p[3] * q[3] * _D % _P d = 2 * p[2] * q[2] % _P e, f, g, h = b - a, d - c, d + c, b + a return (e * f % _P, g * h % _P, f * g % _P, e * h % _P) def _point_mul(scalar, point): out = _IDENTITY while scalar > 0: if scalar & 1: out = _point_add(out, point) point = _point_add(point, point) scalar >>= 1 return out def _point_equal(p, q): if (p[0] * q[2] - q[0] * p[2]) % _P != 0: return False return (p[1] * q[2] - q[1] * p[2]) % _P == 0 def _point_compress(p): z_inv = _inv(p[2]) x = p[0] * z_inv % _P y = p[1] * z_inv % _P return (y | ((x & 1) << 255)).to_bytes(32, "little") def _point_decompress(raw): if len(raw) != 32: return None y = int.from_bytes(raw, "little") sign_bit = y >> 255 y &= (1 << 255) - 1 x = _recover_x(y, sign_bit) if x is None: return None return (x, y, 1, x * y % _P) def _secret_expand(seed): if len(seed) != 32: raise ValueError("Ed25519 seed must be exactly 32 bytes") digest = _sha512(seed) scalar = int.from_bytes(digest[:32], "little") scalar &= (1 << 254) - 8 scalar |= 1 << 254 return scalar, digest[32:] def ed25519_public_key(seed): """32-byte seed -> 32-byte public key.""" scalar, _ = _secret_expand(seed) return _point_compress(_point_mul(scalar, _G)) def ed25519_sign(seed, message): """32-byte seed + message bytes -> 64-byte signature.""" scalar, prefix = _secret_expand(seed) pub = _point_compress(_point_mul(scalar, _G)) r = int.from_bytes(_sha512(prefix + message), "little") % _L big_r = _point_compress(_point_mul(r, _G)) k = int.from_bytes(_sha512(big_r + pub + message), "little") % _L s = (r + k * scalar) % _L return big_r + s.to_bytes(32, "little") def ed25519_verify(public_key, message, signature): """True only for a well-formed signature that checks out. Never raises.""" try: if len(public_key) != 32 or len(signature) != 64: return False point_a = _point_decompress(public_key) if point_a is None: return False r_raw = signature[:32] point_r = _point_decompress(r_raw) if point_r is None: return False s = int.from_bytes(signature[32:], "little") if s >= _L: return False k = int.from_bytes(_sha512(r_raw + public_key + message), "little") % _L return _point_equal(_point_mul(s, _G), _point_add(point_r, _point_mul(k, point_a))) except (ValueError, TypeError): return False # --- END ED25519 SHARED CORE --- # --------------------------------------------------------------------------- # License verification # --------------------------------------------------------------------------- # The production signing key for ciscotools.dev license grants. Replace the # empty string with the hex public key printed by # `python -m backend.services.license_grants --new-key` at release time; the # app refuses every grant it cannot trace to a trusted key. PROD_PUBLIC_KEY_HEX = "7d498657a7dc1206d1942ea537251836151d1ddf005c2684b0ff72d31492820f" # !!! DEVELOPMENT KEY - NOT A SECRET !!! # The matching seed is published in the server source so the suite and a # local run work with no setup. A grant signed with it is REFUSED unless the # operator sets ALLOW_DEV_LICENSE_KEY=true in config.env. DEV_PUBLIC_KEY_HEX = "ff57575dc7af8bfc4d0837cc1ce2017b686a88145dc5579a958e3462fe9a908e" # Environment-only escape hatch for developing against a dev license server. # Deliberately NOT a config.env field: see LicenseClient.dev_override_key. DEV_OVERRIDE_ENV = "NETOPS_DEV_LICENSE_KEY" UNPAID_BANNER = "DEV KEY, NOT A PAID LICENSE." GRACE_DAYS = 30 CLOCK_SKEW_SECONDS = 24 * 3600 LICENSE_PATH = "/api/v1/license" def _utcnow(): return datetime.datetime.now(datetime.timezone.utc) def _iso(moment=None): moment = moment or _utcnow() return moment.replace(microsecond=0).isoformat().replace("+00:00", "Z") def parse_iso8601(text): """Parse the server's timestamps. Returns None for anything unexpected. datetime.fromisoformat on 3.9 does not accept a trailing Z, and the app has to run on stock macOS python3, so the Z is normalised by hand. """ if not isinstance(text, str) or not text: return None cleaned = text.strip() if cleaned.endswith("Z"): cleaned = cleaned[:-1] + "+00:00" try: moment = datetime.datetime.fromisoformat(cleaned) except ValueError: return None if moment.tzinfo is None: moment = moment.replace(tzinfo=datetime.timezone.utc) return moment.astimezone(datetime.timezone.utc) def _b64url_decode(text): pad = "=" * (-len(text) % 4) return base64.urlsafe_b64decode(text + pad) def verify_grant_token(token, trusted_keys): """(grant dict, key_hex) for a token signed by one of `trusted_keys`. Returns (None, reason) for anything else. Never raises: a truncated token, a tampered payload, and a signature from an untrusted key all end the same way, with the Pro cards locked. """ if not isinstance(token, str) or "." not in token: return None, "malformed grant" payload_b64, _, signature_b64 = token.partition(".") if not payload_b64 or not signature_b64: return None, "malformed grant" try: payload = _b64url_decode(payload_b64) signature = _b64url_decode(signature_b64) except Exception: return None, "malformed grant" for key_hex in trusted_keys: if not key_hex: continue try: key = bytes.fromhex(key_hex) except ValueError: continue if ed25519_verify(key, payload, signature): try: grant = json.loads(payload.decode("utf-8")) except Exception: return None, "malformed grant" if not isinstance(grant, dict): return None, "malformed grant" return grant, key_hex return None, "signature does not match a trusted license key" class LicenseState(object): def __init__(self, pro=False, reason="No license key configured.", grant=None, needs_refresh=True, key_hex=""): self.pro = pro self.reason = reason self.grant = grant or {} self.needs_refresh = needs_refresh self.key_hex = key_hex def as_dict(self): return { "pro": self.pro, "reason": self.reason, "account": self.grant.get("account", ""), "plan": self.grant.get("plan", ""), "issued": self.grant.get("issued", ""), "expires": self.grant.get("expires", ""), "key_id": self.grant.get("key_id", ""), "needs_refresh": self.needs_refresh, "dev_key": bool(self.key_hex) and self.key_hex == DEV_PUBLIC_KEY_HEX, # True whenever Pro is unlocked by anything other than the # production key. The UI shouts about it; see UNPAID_BANNER. "unpaid": bool(self.pro) and not ( PROD_PUBLIC_KEY_HEX and self.key_hex == PROD_PUBLIC_KEY_HEX), } def evaluate_grant(grant, key_hex, now=None): """Decide whether a verified grant still unlocks Pro, right now. Three separate clocks have to line up and any one of them can lock the app: the grant's own expiry, the 30-day offline grace from `issued`, and a sanity check that the machine's clock is not set before the grant was issued (which is how you would try to live in the grace window forever). """ now = now or _utcnow() issued = parse_iso8601(grant.get("issued")) expires = parse_iso8601(grant.get("expires")) if issued is None or expires is None: return False, "grant is missing a usable issued/expires timestamp", True if grant.get("plan") != "pro": return False, "this account is not on the Pro plan", True if now < issued - datetime.timedelta(seconds=CLOCK_SKEW_SECONDS): return False, ("this machine's clock is set before the license was " "issued; fix the clock and refresh"), True grace_end = issued + datetime.timedelta(days=GRACE_DAYS) if now >= grace_end: return False, "offline grace period has run out; refresh the license", True if now >= expires: return False, "license grant has expired; refresh the license", True remaining = min(expires, grace_end) - now days = max(0, int(remaining.total_seconds() // 86400)) return True, "Pro unlocked (%d day%s left before the next check)" % ( days, "" if days == 1 else "s"), False class LicenseClient(object): """Fetches, caches, and verifies the offline grant. `opener` is injected so the test suite can exercise every path (valid, invalid key, lapsed subscription, revoked key, malformed body, server unreachable) without a socket ever being opened. """ def __init__(self, config, opener=None, now=None): self.config = config self.cache_path = config.path("LICENSE_CACHE") self.api_key = config.get("CISCOTOOLS_API_KEY", "") self.base_url = config.get("CISCOTOOLS_BASE_URL", "https://ciscotools.dev").rstrip("/") self.allow_dev = config.bool("ALLOW_DEV_LICENSE_KEY", False) self._opener = opener or self._http_get self._now = now or _utcnow def dev_override_key(self): """A build-your-own trust anchor, for developing against a dev server. Environment ONLY, never config.env: a setting an engineer can fill in from a documented example file is not a developer tool, it is a paywall with a form field. It is also refused outright once a production key is compiled in, so a shipped build cannot be talked into trusting a self-minted grant without editing the source (which anyone who owns the file can do anyway, and which leaves a trace). """ raw = (os.environ.get(DEV_OVERRIDE_ENV) or "").strip() if not raw or PROD_PUBLIC_KEY_HEX: return "" if not re.match(r"^[0-9a-fA-F]{64}$", raw): return "" return raw.lower() def trusted_keys(self): """Keys whose signature unlocks Pro, in order of authority.""" keys = [] if PROD_PUBLIC_KEY_HEX: keys.append(PROD_PUBLIC_KEY_HEX) if self.allow_dev: keys.append(DEV_PUBLIC_KEY_HEX) override = self.dev_override_key() if override: keys.append(override) return keys def is_paid_key(self, key_hex): return bool(PROD_PUBLIC_KEY_HEX) and key_hex == PROD_PUBLIC_KEY_HEX def read_cache(self): try: with open(self.cache_path, "r", encoding="utf-8") as handle: blob = json.load(handle) except (OSError, ValueError): return "" token = blob.get("grant_token") if isinstance(blob, dict) else None return token if isinstance(token, str) else "" def write_cache(self, token, extra=None): blob = {"grant_token": token, "cached_at": _iso(self._now()), "app": APP_NAME, "app_version": APP_VERSION} if isinstance(extra, dict): for name in ("key_id", "grace_days"): if name in extra: blob[name] = extra[name] directory = os.path.dirname(os.path.abspath(self.cache_path)) if directory and not os.path.isdir(directory): os.makedirs(directory, exist_ok=True) temp = self.cache_path + ".tmp" with open(temp, "w", encoding="utf-8") as handle: json.dump(blob, handle, indent=2) os.replace(temp, self.cache_path) try: os.chmod(self.cache_path, 0o600) except OSError: pass def state(self): token = self.read_cache() if not token: if not self.api_key: return LicenseState( False, "No API key configured. Put your ciscotools.dev key in " "config.env as CISCOTOOLS_API_KEY to unlock Pro.", ) return LicenseState(False, "No cached license yet. Refresh to fetch one.") grant, key_hex = verify_grant_token(token, self.trusted_keys()) if grant is None: return LicenseState(False, key_hex) if grant.get("app") not in (APP_NAME, "any"): return LicenseState(False, "this grant was issued for another app", grant, True, key_hex) ok, reason, needs_refresh = evaluate_grant(grant, key_hex, self._now()) if ok and not self.is_paid_key(key_hex): reason = "%s %s" % (UNPAID_BANNER, reason) return LicenseState(ok, reason, grant, needs_refresh, key_hex) def _http_get(self, url, headers, timeout): # pragma: no cover - network request = urllib.request.Request(url, headers=headers, method="GET") context = ssl.create_default_context() with urllib.request.urlopen(request, timeout=timeout, context=context) as response: return response.status, response.read().decode("utf-8", "replace") def refresh(self): """Ask the server for a fresh grant. Returns (ok, message).""" if not self.api_key: return False, ("No API key configured. Add CISCOTOOLS_API_KEY to " "config.env (find it on your account page).") query = urllib.parse.urlencode({"app": APP_NAME, "version": APP_VERSION}) url = "%s%s?%s" % (self.base_url, LICENSE_PATH, query) headers = { "X-API-Key": self.api_key, "Accept": "application/json", "User-Agent": "%s/%s" % (APP_NAME, APP_VERSION), } try: status, body = self._opener(url, headers, 20) except urllib.error.HTTPError as error: # pragma: no cover - network status = error.code try: body = error.read().decode("utf-8", "replace") except Exception: body = "" except Exception as error: # pragma: no cover - network return False, "Could not reach %s (%s)." % (self.base_url, error) if status == 401: return False, ("The server rejected this API key. Issue a new one " "on your account page.") if status == 402: return False, ("This account is not on the Pro plan right now. " "Lite keeps working.") if status != 200: return False, "License server returned HTTP %s." % status try: payload = json.loads(body) except ValueError: return False, "License server returned a response this app could not read." token = payload.get("grant_token") if isinstance(payload, dict) else None if not isinstance(token, str) or not token: return False, "License server returned no grant." grant, key_hex = verify_grant_token(token, self.trusted_keys()) if grant is None: hint = "" if key_hex.startswith("signature") and not PROD_PUBLIC_KEY_HEX: hint = (" This build has no production license key compiled in; " "set ALLOW_DEV_LICENSE_KEY=true only against a dev server.") return False, "Refused the grant: %s.%s" % (key_hex, hint) # Scope check BEFORE the cache write. state() has always enforced this; # refresh() used to cache the grant and report "Pro unlocked" while # every Pro card stayed locked, which is a lie the UI then repeats. if grant.get("app") not in (APP_NAME, "any"): return False, ("Refused the grant: it was issued for another app " "(%r), not this one." % str(grant.get("app"))[:40]) ok, reason, _needs = evaluate_grant(grant, key_hex, self._now()) if not ok: return False, "Refused the grant: %s." % reason self.write_cache(token, payload) if not self.is_paid_key(key_hex): reason = "%s %s" % (UNPAID_BANNER, reason) return True, reason # --------------------------------------------------------------------------- # config.env # --------------------------------------------------------------------------- CONFIG_DEFAULTS = { # Catalyst Center (Lite, read only) "CATALYST_URL": "", "CATALYST_USERNAME": "", "CATALYST_PASSWORD": "", "CATALYST_VERIFY_TLS": "true", # Cisco ISE (Pro) "ISE_URL": "", "ISE_USERNAME": "", "ISE_PASSWORD": "", "ISE_VERIFY_TLS": "true", "ISE_BULK_THRESHOLD": "20", # Windows DHCP over SSH + PowerShell (Pro) "DHCP_HOST": "", "DHCP_SSH_USERNAME": "", "DHCP_SSH_PASSWORD": "", "DHCP_SSH_PORT": "22", "DHCP_SERVER": "", # Behaviour "DRY_RUN": "true", "HTTP_TIMEOUT": "30", "AUDIT_LOG": "./netops-audit.log", "EXPORT_DIR": "./exports", # License "CISCOTOOLS_API_KEY": "", "CISCOTOOLS_BASE_URL": "https://ciscotools.dev", "LICENSE_CACHE": "./license.json", "ALLOW_DEV_LICENSE_KEY": "false", # UI "BIND_HOST": "127.0.0.1", "BIND_PORT": "8782", "OPEN_BROWSER": "true", } SECRET_KEYS = ("CATALYST_PASSWORD", "ISE_PASSWORD", "DHCP_SSH_PASSWORD", "CISCOTOOLS_API_KEY") TRUE_WORDS = ("1", "true", "yes", "on") FALSE_WORDS = ("0", "false", "no", "off", "") def parse_config_env(text): """Parse a config.env body into a dict. Tolerant, never raises. Accepts KEY=value, ignores blank lines and # comments, strips one layer of matching quotes, and keeps a trailing inline comment only when the value is unquoted (so a password containing a # is not truncated). """ values = {} for raw_line in text.splitlines(): line = raw_line.strip() if not line or line.startswith("#"): continue if line.startswith("export "): line = line[len("export "):].strip() if "=" not in line: continue name, _, value = line.partition("=") name = name.strip() if not name or not re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", name): continue value = value.strip() if len(value) >= 2 and value[0] == value[-1] and value[0] in ("'", '"'): value = value[1:-1] values[name] = value return values class Config(object): def __init__(self, values=None, base_dir=None): self.base_dir = os.path.abspath(base_dir or os.getcwd()) self.values = dict(CONFIG_DEFAULTS) if values: self.values.update(values) self.unknown = sorted(set(values or {}) - set(CONFIG_DEFAULTS)) @classmethod def load(cls, path): base_dir = os.path.dirname(os.path.abspath(path)) or os.getcwd() try: with open(path, "r", encoding="utf-8") as handle: values = parse_config_env(handle.read()) except OSError: values = {} for name in CONFIG_DEFAULTS: if os.environ.get(name): values[name] = os.environ[name] return cls(values, base_dir) def get(self, name, default=""): value = self.values.get(name, default) return value if value is not None else default def bool(self, name, default=False): raw = str(self.get(name, "")).strip().lower() if raw in TRUE_WORDS: return True if raw in FALSE_WORDS: return False if raw != "" else default return default def int(self, name, default): try: return int(str(self.get(name, "")).strip()) except (TypeError, ValueError): return default def path(self, name): raw = self.get(name, "") if not raw: raw = CONFIG_DEFAULTS.get(name, "") if os.path.isabs(raw): return raw return os.path.normpath(os.path.join(self.base_dir, raw)) def redacted(self): out = {} for name, value in sorted(self.values.items()): if name in SECRET_KEYS: out[name] = "(set)" if value else "(not set)" else: out[name] = value return out def prompt_for_missing_secrets(self, interactive=True, prompter=None): """Blank password fields are asked for at startup, never stored.""" prompter = prompter or getpass.getpass if not interactive: return pairs = (("CATALYST_URL", "CATALYST_USERNAME", "CATALYST_PASSWORD", "Catalyst Center"), ("ISE_URL", "ISE_USERNAME", "ISE_PASSWORD", "Cisco ISE"), ("DHCP_HOST", "DHCP_SSH_USERNAME", "DHCP_SSH_PASSWORD", "the Windows DHCP host")) for host_key, user_key, secret_key, label in pairs: if not self.get(host_key) or not self.get(user_key): continue if self.get(secret_key): continue try: self.values[secret_key] = prompter( "Password for %s on %s: " % (self.get(user_key), label)) except (EOFError, KeyboardInterrupt): self.values[secret_key] = "" # --------------------------------------------------------------------------- # Audit log # --------------------------------------------------------------------------- _AUDIT_LOCK = threading.Lock() # A device command carries its secret as a positional argument, not under a # field name, so redact() cannot see it by key. The list is deliberately a # little over-eager: losing the argument of "crypto key generate" from an # audit line costs nothing, and keeping a TACACS key in a file on someone's # laptop costs a lot. # # The trailing lookahead accepts `:` `=` and `,` as well as whitespace, so # "Secret: Tr0ub4dor" in banner prose is caught the same as "secret X". _SECRET_ARG_RE = re.compile( r"(?i)(?:^|[\s;])(password|passwd|passphrase|secret|community|" r"key-string|key|pre-shared-key|shared-secret|psk|wpa-psk|md5|" r"authentication-key|auth-key|encryption|credential)(?=[\s:=,]|$)") _REDACTED = "***" # Fields whose value is a device command (or a list of them). _COMMAND_FIELDS = ("commands", "command", "config_lines", "lines") # Multi-line blocks whose BODY is free prose. A banner can carry a secret with # no keyword anywhere near it ("the root pw is Tr0ub4dor"), so once one opens # every line up to its delimiter is dropped wholesale rather than scanned. # Group 1 is the header through the delimiter, 2 the delimiter, 3 the rest. _BANNER_OPEN_RE = re.compile(r"(?i)^(\s*banner\s+\S+\s+(\S+))(.*)$") # `certificate self-signed 01` ... `quit` is the other block shape that # carries opaque material line by line. _QUIT_BLOCK_OPEN_RE = re.compile(r"(?i)^\s*certificate(\s|$)") # Punctuation that can sit between a keyword and its value, or trail a bare # prompt. Stripping it decides whether anything actually FOLLOWS the keyword. _SECRET_PUNCTUATION = " \t:=," def _segment_shape(segment): """(match, has_value, has_prefix) for one command segment. Three shapes, and the difference decides how much is kept: "snmp-server community S3cret RO" the keyword names a parameter and the VALUE follows -> drop the tail. "Hunter2 is the password" the keyword is last, so whatever it refers to came BEFORE -> drop it all. "Password: " the keyword is the whole segment: a device's own prompt, with no secret on it at all -> leave it alone. """ if not isinstance(segment, str): return None, False, False match = _SECRET_ARG_RE.search(segment) if match is None: return None, False, False has_value = bool(segment[match.end(1):].strip(_SECRET_PUNCTUATION).strip()) has_prefix = bool(segment[:match.start(1)].strip(_SECRET_PUNCTUATION).strip()) return match, has_value, has_prefix def _segment_secret(segment): """The part of one command segment that must not be kept. "" if none.""" match, has_value, has_prefix = _segment_shape(segment) if match is None: return "" if has_value: return segment[match.end(1):].strip() if has_prefix: return segment.strip() return "" def _redact_segment(segment): match, has_value, has_prefix = _segment_shape(segment) if match is None: return segment if has_value: return segment[:match.end(1)] + " " + _REDACTED if has_prefix: return _REDACTED # Nothing before it and nothing after it: a bare "Password:" prompt echoed # back by the device. Rewriting that to "Password ***" only made a # transcript read oddly; there is no secret on the line to remove. return segment def redact_command(text): """Scrub one device command. `;` is a command separator on NX-OS, so each segment is judged on its own: a secret chained behind a harmless first segment cannot hide behind it. """ if not isinstance(text, str): return text if ";" not in text: return _redact_segment(text) return "".join(part if part == ";" else _redact_segment(part) for part in re.split(r"(;)", text)) def _banner_block(line): """(header, delimiter, remainder) when `line` opens a banner block.""" match = _BANNER_OPEN_RE.match(line) if match is None: return None header, delimiter, remainder = match.group(1), match.group(2), match.group(3) if delimiter in remainder: return None # single-line banner: the body is right here, not a block return header, delimiter, remainder def redact_command_list(commands): """Redact a list of commands, treating multi-line blocks as blocks.""" out = [] block = None for item in commands: if not isinstance(item, str): out.append(item) continue if block is not None: kind, delimiter = block if kind == "quit" and item.strip().lower() == "quit": out.append(item) block = None elif kind == "banner" and delimiter in item: out.append(_REDACTED) block = None else: out.append(_REDACTED) continue opened = _banner_block(item) if opened is not None: header, delimiter, remainder = opened out.append(header + (" " + _REDACTED if remainder.strip() else "")) block = ("banner", delimiter) continue if _QUIT_BLOCK_OPEN_RE.match(item): out.append(item) block = ("quit", "") continue out.append(redact_command(item)) return out def secret_fragments(commands): """Every substring the command redaction dropped, for scrubbing elsewhere. The per-device transcript is the device echoing back what we sent, so it holds the same secrets. Rather than guess at the structure of a terminal stream, the exact values we know we sent are removed from it by name. """ fragments = set() block = None for item in commands: if not isinstance(item, str): continue if block is not None: kind, delimiter = block if kind == "quit" and item.strip().lower() == "quit": block = None else: body = item.strip() # A bare delimiter or a two-character marker is not a secret, # and replacing it across a transcript is pure noise. if len(body) >= 3 and body != delimiter: fragments.add(body) if kind == "banner" and delimiter in item: block = None continue opened = _banner_block(item) if opened is not None: _header, delimiter, remainder = opened if len(remainder.strip()) >= 3: fragments.add(remainder.strip()) block = ("banner", delimiter) continue if _QUIT_BLOCK_OPEN_RE.match(item): block = ("quit", "") continue for part in re.split(r"(;)", item): if part == ";": continue secret = _segment_secret(part) if len(secret) >= 3: fragments.add(secret) return fragments def redact_transcript(text, commands): """Scrub a CONFIG-PUSH transcript of the secrets we sent. Applied only to config-mode runs. A show run's transcript is the device's own output and is the product -- scrubbing a `show running-config` dump would break the feature people download this for -- so it is written as the device printed it. The README says exactly that. Two passes: the exact values we know we sent are removed wherever they appear (including the device's echo), then every line is run through the command scrubber to catch anything the device echoed differently. """ if not isinstance(text, str) or not text: return text out = text for fragment in sorted(secret_fragments(commands), key=len, reverse=True): out = out.replace(fragment, _REDACTED) lines = [] for line in out.split("\n"): suffix = "\r" if line.endswith("\r") else "" body = line[:-1] if suffix else line lines.append(redact_command(body) + suffix) return "\n".join(lines) def redact(record): """Strip anything that smells like a credential, at any depth.""" if isinstance(record, dict): out = {} for name, value in record.items(): lowered = str(name).lower() if any(word in lowered for word in ("password", "passwd", "secret", "api_key", "apikey", "token", "credential")): out[name] = "***" elif lowered in _COMMAND_FIELDS: if isinstance(value, (list, tuple)): out[name] = redact_command_list(list(value)) else: out[name] = redact_command(value) else: out[name] = redact(value) return out if isinstance(record, list): return [redact(item) for item in record] return record def append_audit(path, record): """Append ONE json line. Called only after a real write succeeded.""" entry = redact(dict(record)) entry.setdefault("at", _iso()) entry.setdefault("app", APP_NAME) entry.setdefault("app_version", APP_VERSION) line = json.dumps(entry, sort_keys=True) + "\n" directory = os.path.dirname(os.path.abspath(path)) if directory and not os.path.isdir(directory): os.makedirs(directory, exist_ok=True) with _AUDIT_LOCK: # 0600 at creation: this file records what was pushed to devices, and # even after redaction that is nobody else's business on a shared # machine. An existing file keeps whatever mode its owner chose. handle = _open_private(path, "a") try: handle.write(line) finally: handle.close() return entry def _open_private(path, mode="w"): """open() that creates the file 0600 instead of 0644.""" flags = os.O_WRONLY | os.O_CREAT flags |= os.O_APPEND if "a" in mode else os.O_TRUNC fd = os.open(path, flags, 0o600) return os.fdopen(fd, mode, encoding="utf-8") # --------------------------------------------------------------------------- # MAC addresses # --------------------------------------------------------------------------- # A MAC may be written with colons, hyphens, Cisco dots, or nothing at all. # Anything outside that alphabet is a typo, not a format worth guessing at. _MAC_ALPHABET_RE = re.compile(r"^[0-9a-fA-F.:\- ]+$") _MAC_SEPARATORS_RE = re.compile(r"[.:\- ]") def normalize_mac(value): """Any common MAC spelling -> AA:BB:CC:DD:EE:FF. None if it is not one. Accepts colon, hyphen, dot (Cisco 0000.1111.2222) and bare-hex forms, plus stray whitespace. Refuses the all-zero address and any multicast address: neither is a thing you put in a MAB endpoint group, and letting one through would create junk in ISE that somebody has to clean up. """ if not isinstance(value, str): return None stripped = value.strip() if not stripped or not _MAC_ALPHABET_RE.match(stripped): return None digits = _MAC_SEPARATORS_RE.sub("", stripped) if len(digits) != 12: return None octets = [digits[index:index + 2].upper() for index in range(0, 12, 2)] if all(octet == "00" for octet in octets): return None if int(octets[0], 16) & 0x01: return None # multicast / broadcast return ":".join(octets) def normalize_mac_list(text): """Split, normalise, dedupe (order preserved). Returns (macs, rejected). Splitting is deliberately two-stage. "AA BB CC DD EE FF" is one MAC written with spaces; "aabb.ccdd.eeff 0011.2233.4455" is two MACs written with a space between them. Splitting on whitespace first turned the former into six bogus rejects, so each comma/semicolon/newline-delimited chunk is tried WHOLE first, and only split on spaces if that fails. """ if isinstance(text, (list, tuple)): chunks = [str(item) for item in text] else: chunks = re.split(r"[\n\r,;]+", str(text or "")) macs = [] seen = set() rejected = [] def take(candidate): candidate = str(candidate).strip() if not candidate: return mac = normalize_mac(candidate) if mac is None: rejected.append(candidate[:64]) return if mac in seen: return seen.add(mac) macs.append(mac) for chunk in chunks: chunk = chunk.strip() if not chunk: continue if normalize_mac(chunk) is not None: take(chunk) continue pieces = chunk.split() if len(pieces) > 1: for piece in pieces: take(piece) else: take(chunk) return macs, rejected def mac_for_dhcp_clientid(mac): """Windows DHCP wants the client id as hyphen separated hex.""" return mac.replace(":", "-").lower() # --------------------------------------------------------------------------- # HTTP # --------------------------------------------------------------------------- class HttpResponse(object): def __init__(self, status, body="", headers=None): self.status = status self.body = body self.headers = headers or {} def json(self): try: return json.loads(self.body) except (ValueError, TypeError): return None class HttpClient(object): """Thin urllib wrapper. Every controller client takes one of these, so the test suite substitutes a fake and nothing opens a socket.""" def __init__(self, timeout=30, verify_tls=True): self.timeout = timeout self.verify_tls = verify_tls self.calls = [] def _context(self): # pragma: no cover - network context = ssl.create_default_context() if not self.verify_tls: context.check_hostname = False context.verify_mode = ssl.CERT_NONE return context def request(self, method, url, headers=None, body=None): # pragma: no cover data = None if body is not None: data = body if isinstance(body, bytes) else json.dumps(body).encode("utf-8") request = urllib.request.Request(url, data=data, headers=headers or {}, method=method) try: with urllib.request.urlopen(request, timeout=self.timeout, context=self._context()) as response: return HttpResponse(response.status, response.read().decode("utf-8", "replace"), dict(response.headers)) except urllib.error.HTTPError as error: try: text = error.read().decode("utf-8", "replace") except Exception: text = "" return HttpResponse(error.code, text, dict(error.headers or {})) except Exception as error: return HttpResponse(0, "%s" % error) class FakeHttp(HttpClient): """Routing table keyed by (METHOD, path). Used by the tests only.""" def __init__(self, routes=None): HttpClient.__init__(self) self.routes = dict(routes or {}) self.calls = [] def request(self, method, url, headers=None, body=None): path = urllib.parse.urlsplit(url).path query = urllib.parse.urlsplit(url).query self.calls.append({"method": method, "url": url, "path": path, "query": query, "headers": dict(headers or {}), "body": body}) handler = self.routes.get((method, path)) if handler is None: return HttpResponse(404, json.dumps({"message": "no route %s %s" % (method, path)})) if callable(handler): return handler(self.calls[-1]) return handler def _basic_auth(username, password): raw = ("%s:%s" % (username, password)).encode("utf-8") return "Basic " + base64.b64encode(raw).decode("ascii") # --------------------------------------------------------------------------- # Catalyst Center (read only, Lite) # --------------------------------------------------------------------------- class CatalystError(Exception): pass class CatalystClient(object): """Cisco Catalyst Center Intent API v1, read paths only. There is deliberately no write method on this class. The Lite widgets read inventory, interface state, and client location; nothing in this app can change a Catalyst Center object, so a mis-click cannot. """ AUTH_PATH = "/dna/system/api/v1/auth/token" DEVICE_PATH = "/dna/intent/api/v1/network-device" INTERFACE_PATH = "/dna/intent/api/v1/interface/network-device/%s" CLIENT_PATH = "/dna/intent/api/v1/client-detail" def __init__(self, base_url, username, password, http=None, timeout=30, verify_tls=True): self.base_url = (base_url or "").rstrip("/") self.username = username self._password = password self.http = http or HttpClient(timeout, verify_tls) self._token = "" self._token_at = 0.0 def configured(self): return bool(self.base_url and self.username) def token(self, force=False): if self._token and not force and (time.time() - self._token_at) < 3000: return self._token response = self.http.request( "POST", self.base_url + self.AUTH_PATH, headers={"Authorization": _basic_auth(self.username, self._password), "Content-Type": "application/json"}) if response.status in (401, 403): raise CatalystError("Catalyst Center rejected the credentials.") if response.status not in (200, 201): raise CatalystError("Catalyst Center auth returned HTTP %s." % response.status) payload = response.json() or {} token = payload.get("Token") or payload.get("token") if not token: raise CatalystError("Catalyst Center auth returned no token.") self._token = token self._token_at = time.time() return token def _get(self, path, params=None): url = self.base_url + path if params: url += "?" + urllib.parse.urlencode(params) response = self.http.request("GET", url, headers={ "X-Auth-Token": self.token(), "Accept": "application/json"}) if response.status == 401: # One retry with a fresh token: the controller expires tokens # aggressively and a stale one is not an error worth showing. response = self.http.request("GET", url, headers={ "X-Auth-Token": self.token(force=True), "Accept": "application/json"}) if response.status != 200: raise CatalystError("Catalyst Center returned HTTP %s for %s" % (response.status, path)) payload = response.json() if payload is None: raise CatalystError("Catalyst Center returned a body this app could not read.") return payload def devices(self, limit=500): payload = self._get(self.DEVICE_PATH, {"limit": max(1, min(limit, 500))}) out = [] for item in (payload.get("response") or []): if not isinstance(item, dict): continue out.append({ "id": str(item.get("id", ""))[:128], "hostname": str(item.get("hostname", ""))[:128], "ip": str(item.get("managementIpAddress", ""))[:64], "platform": str(item.get("platformId", ""))[:64], "family": str(item.get("family", ""))[:64], "software": str(item.get("softwareVersion", ""))[:64], "reachability": str(item.get("reachabilityStatus", ""))[:32], "site": str(item.get("locationName") or "")[:128], }) return out def interfaces(self, device_id): if not re.match(r"^[A-Za-z0-9._:-]{1,128}$", str(device_id or "")): raise CatalystError("That does not look like a device id.") payload = self._get(self.INTERFACE_PATH % urllib.parse.quote(device_id)) out = [] for item in (payload.get("response") or []): if not isinstance(item, dict): continue out.append({ "name": str(item.get("portName", ""))[:64], "status": str(item.get("status", ""))[:32], "admin": str(item.get("adminStatus", ""))[:32], "vlan": str(item.get("vlanId") or "")[:16], "speed": str(item.get("speed") or "")[:24], "duplex": str(item.get("duplex") or "")[:16], "description": str(item.get("description") or "")[:128], "mac": str(item.get("macAddress") or "")[:32], }) return out def find_mac(self, mac): normalised = normalize_mac(mac) if normalised is None: raise CatalystError("That is not a MAC address this app recognises.") payload = self._get(self.CLIENT_PATH, { "macAddress": normalised.lower(), "timestamp": int(time.time() * 1000)}) detail = payload.get("detail") or {} connection = payload.get("connectionInfo") or {} return { "mac": normalised, "found": bool(detail), "hostname": str(detail.get("hostName") or "")[:128], "ip": str(detail.get("hostIpV4") or "")[:64], "type": str(detail.get("hostType") or "")[:32], "status": str(detail.get("connectionStatus") or "")[:32], "switch": str(connection.get("nwDeviceName") or "")[:128], "port": str(connection.get("interfaceName") or detail.get("port") or "")[:64], "vlan": str(detail.get("vlanId") or "")[:16], "ssid": str(detail.get("ssid") or "")[:64], } # --------------------------------------------------------------------------- # SecureCRT session export # --------------------------------------------------------------------------- _UNSAFE_NAME_RE = re.compile(r"[^A-Za-z0-9._-]") _INI_STRIP_RE = re.compile(r"[\x00-\x1f\x7f]") def safe_filename(name, fallback="device"): """One path COMPONENT, with no way out of the directory it lands in.""" cleaned = _UNSAFE_NAME_RE.sub("_", str(name or "").strip()) cleaned = cleaned.lstrip(".") cleaned = cleaned[:64] return cleaned or fallback def safe_archive_name(*parts): """Build a zip entry path that cannot escape the extraction directory. This is the Zip Slip guard: every component is reduced to a safe filename, so "..", "/etc/passwd", "C:\\x", and a name carrying a backslash all collapse to something harmless before they are written. """ cleaned = [safe_filename(part) for part in parts if str(part or "").strip()] cleaned = [part for part in cleaned if part not in ("", ".", "..")] if not cleaned: cleaned = ["item"] return "/".join(cleaned) def ini_escape(value): """SecureCRT .ini values are one line. Control characters are removed. A controller is an untrusted source as far as this app is concerned: a hostname carrying a newline would otherwise append a second directive to the session file. """ return _INI_STRIP_RE.sub("", str(value or "")) def ini_dword(number): return "%08x" % (int(number) & 0xFFFFFFFF) def build_session_ini(device, username="", port=22, protocol="SSH2", emulation="Xterm"): lines = [ "S:\"Hostname\"=%s" % ini_escape(device.get("ip", "")), "S:\"Protocol Name\"=%s" % ini_escape(protocol), "D:\"[SSH2] Port\"=%s" % ini_dword(port), "D:\"Port\"=%s" % ini_dword(port), "S:\"Emulation\"=%s" % ini_escape(emulation), "S:\"Description\"=%s" % ini_escape( "%s %s exported by %s %s" % (device.get("hostname", ""), device.get("platform", ""), APP_TITLE, APP_VERSION)), "D:\"Session Password Saved\"=%s" % ini_dword(0), "D:\"Auth Prompts in Window\"=%s" % ini_dword(1), ] if username: lines.append("S:\"Username\"=%s" % ini_escape(username)) return "\r\n".join(lines) + "\r\n" def build_sessions_csv(devices): buffer = io.StringIO() writer = csv.writer(buffer, lineterminator="\n") writer.writerow(["hostname", "ip", "platform", "site", "session_path"]) for device in devices: writer.writerow([ device.get("hostname", ""), device.get("ip", ""), device.get("platform", ""), device.get("site", ""), safe_archive_name("Sessions", device.get("site") or "CatalystCenter", (device.get("hostname") or "device") + ".ini"), ]) return buffer.getvalue() SESSION_README = ( "SecureCRT sessions exported by %s %s\r\n" "\r\n" "Import on macOS:\r\n" " 1. Quit SecureCRT.\r\n" " 2. In Finder press Shift-Cmd-G and go to\r\n" " ~/Library/Application Support/VanDyke/SecureCRT/Config/Sessions/\r\n" " 3. Copy the Sessions folder from this zip into that folder.\r\n" " 4. Start SecureCRT. The sessions appear in the Session Manager.\r\n" "\r\n" "Import on Windows:\r\n" " Copy the Sessions folder into\r\n" " %%APPDATA%%\\VanDyke\\Config\\Sessions\\\r\n" "\r\n" "No password is stored in these files. SecureCRT will prompt you.\r\n" ) % (APP_TITLE, APP_VERSION) def build_session_zip(devices, username="", port=22): """Zip of one .ini per device plus a sessions.csv index. Bytes, in memory.""" buffer = io.BytesIO() with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as archive: seen = set() for device in devices: name = safe_archive_name("Sessions", device.get("site") or "CatalystCenter", (device.get("hostname") or "device") + ".ini") candidate = name counter = 2 while candidate in seen: candidate = name[:-4] + ("_%d.ini" % counter) counter += 1 seen.add(candidate) archive.writestr(candidate, build_session_ini(device, username, port)) archive.writestr("sessions.csv", build_sessions_csv(devices)) archive.writestr("README.txt", SESSION_README) return buffer.getvalue() # --------------------------------------------------------------------------- # Cisco ISE (Pro) # --------------------------------------------------------------------------- class IseError(Exception): pass # ISE ERS answers 2xx with an error envelope often enough that a status code # alone over-reports success -- and an audit line claiming a write the # controller refused is worse than no line at all. # # STRUCTURED SIGNALS ONLY. An earlier version also scanned the message text # for words like "fail", "invalid" and "error", which meant a SUCCESSFUL write # into an endpoint group called `failover-lab`, `invalid-mac-quarantine` or # `error-budget` was reported as a failure and got no audit line -- the exact # property the audit log exists to provide, lost to a substring match on a # customer's naming convention. Group names and descriptions are user data and # are never evidence of anything. A genuine ERS refusal always carries one of # the markers below (or a non-2xx status, which the caller checks first), so # nothing real is lost by refusing to guess. def _is_ers_error_type(value): """True for an ERS message `type` that means refusal. Covers "ERROR", "error", "ERROR_VALIDATION", "ERROR-CRUD" and "FATAL", but not a word that merely starts with those letters. """ token = str(value or "").strip().upper() for marker in ("ERROR", "FATAL"): if token == marker: return True if token.startswith(marker) and not token[len(marker):len(marker) + 1].isalnum(): return True return False def ers_error(response): """Return a reason string when a 2xx from ISE is really a refusal. Three structured signals, cheapest first: an explicit `success: false`, an ERSResponse envelope carrying a message whose `type` is an error, and a top-level `error`/`errors` key. An empty or non-JSON body on a 2xx is the ordinary success shape for a create, so that is not an error, and neither is prose that happens to contain an alarming word. """ payload = response.json() if not isinstance(payload, dict): return "" if payload.get("success") is False: return (_ers_first_message(payload) or "ISE reported success: false")[:200] envelope = payload.get("ERSResponse") or payload.get("ersResponse") if isinstance(envelope, dict): for message in envelope.get("messages") or []: if not isinstance(message, dict): continue if _is_ers_error_type(message.get("type")): return (str(message.get("title") or message.get("description") or "ISE returned an ERROR message")[:200]) for name in ("error", "errors"): value = payload.get(name) if value: return (_ers_text(value) or "ISE returned an error")[:200] return "" def _ers_text(value): """Flatten whatever an `error`/`errors` key held into one line.""" if isinstance(value, str): return value if isinstance(value, dict): for name in ("title", "message", "description", "detail"): if value.get(name): return str(value[name]) return "" if isinstance(value, (list, tuple)): for item in value: text = _ers_text(item) if text: return text return "" def _ers_first_message(payload): """Best-effort human text out of an ERS body, whatever shape it took.""" envelope = payload.get("ERSResponse") or payload.get("ersResponse") if isinstance(envelope, dict): for message in envelope.get("messages") or []: if isinstance(message, dict): text = message.get("title") or message.get("description") if text: return str(text) if envelope.get("detail"): return str(envelope["detail"]) for name in ("message", "title", "detail", "error"): if payload.get(name): return str(payload[name]) return "" # An ISE object id that is safe to interpolate into a request path. ISE hands # out UUIDs, but the id inside an endpoint record is data from the controller, # so it is checked BEFORE it is used to build a URL, not after. ISE_ID_RE = re.compile(r"^[A-Za-z0-9._:-]{1,64}$") def _ers_bool(value): """Tri-state: True, False, or None when ISE did not say. ERS spells booleans both ways depending on the node and the media type: JSON `true` and the string `"true"` both turn up. `bool("false")` is True, so a plain cast would report a dynamically profiled endpoint as a static assignment, which is a confident and wrong claim about somebody's ISE. Absent stays None so the line can leave it out instead of guessing. """ if isinstance(value, bool): return value if isinstance(value, str): token = value.strip().lower() if token in ("true", "yes", "1"): return True if token in ("false", "no", "0"): return False return None if isinstance(value, int): return bool(value) return None def _one_line(value, limit=128): """Controller text, flattened to one line. Group names and descriptions end up in a CLI line, a browser message and an audit record. A description carrying a newline would otherwise split the line in two and read as a second, separate fact. """ return _INI_STRIP_RE.sub(" ", str(value or "")).strip()[:limit] def describe_ise_endpoint(mac, record, group_name=""): """One unambiguous line saying where a MAC sits in ISE right now. The identity group NAME is the point. The endpoint record carries `groupId`, an ISE UUID, which told an engineer reading the result nothing at all; this says the name out loud. Anything the record did not carry is left out rather than guessed at: "static assignment" is only printed when ISE actually said so. """ if not record: return "%s is not in ISE." % mac if group_name: core = "identity group '%s'" % group_name elif record.get("group_id"): core = ("identity group id %s (ISE gave no name for it)" % record["group_id"]) else: core = "no identity group" static = record.get("static_group_assignment") if static is True: core += " (static assignment)" elif static is False: core += " (dynamic assignment)" parts = [core] if record.get("description"): parts.append("description '%s'" % record["description"]) # A full stop, always: this sentence gets a second one glued onto it by # describe_ise_action, and "...'known endpoint' Push will MOVE it" reads # as one run-on claim. return "%s is in ISE: %s." % (mac, ", ".join(parts)) def describe_ise_action(action, target_group_name=""): """The plan line for one MAC: where it is now, and what a push would do. A MOVE names BOTH groups. Moving a MAC out of the group its authorisation policy matches is the change that takes something off the network, and the word "update" on its own hides it completely. """ target = target_group_name or action.get("group_id") or "the target group" state = describe_ise_endpoint(action["mac"], action.get("current"), action.get("current_group_name", "")) if action.get("action") == "create": return "%s Push will CREATE it in identity group '%s'." % (state, target) if action.get("moves"): current = (action.get("current_group_name") or action.get("current_group_id")) if not current: return "%s Push will SET its identity group to '%s'." % (state, target) return "%s Push will MOVE it: '%s' -> '%s'." % (state, current, target) return "%s Push will UPDATE it in place (already in '%s')." % (state, target) def _plan_summary(plan): """The one-line headline for a whole push: where to, and what changes. The move clause is never folded into "updates", because a MAC being updated in the group it is already in and a MAC being dragged out of another group are not the same event, and only one of them can drop a device off the network. """ target = plan.get("target_group_name") or plan.get("group_name") \ or plan.get("group_id") or "the target group" same = plan["updates"] - plan["moves"] bits = [] if plan["creates"]: bits.append("%d new" % plan["creates"]) if same: bits.append("%d already in this group" % same) if plan["moves"]: bits.append("%d MOVING from another group (%s)" % (plan["moves"], "; ".join(plan["move_pairs"]))) return ("%d endpoint(s) into identity group '%s' over the %s path: %s." % (plan["count"], target, plan["transport"], ", ".join(bits) or "nothing to change")) class IseClient(object): """Cisco ISE ERS: endpoint groups, and create-or-update MAB endpoints. Every write path takes `dry_run` and returns the request it WOULD have sent when it is true. Nothing in this class sends a write unless the caller explicitly passed dry_run=False, which the UI only does after a confirm. """ GROUPS_PATH = "/ers/config/endpointgroup" ENDPOINT_PATH = "/ers/config/endpoint" BULK_PATH = "/ers/config/endpoint/bulk/submit" def __init__(self, base_url, username, password, http=None, timeout=30, verify_tls=True, bulk_threshold=20): self.base_url = (base_url or "").rstrip("/") self.username = username self._password = password self.http = http or HttpClient(timeout, verify_tls) self.bulk_threshold = max(2, int(bulk_threshold or 20)) # groupId -> {id, name, description} or None for a resolution that # failed. Filled by endpoint_groups() (the dropdown already fetches # the list) and topped up by group() for anything that list missed. # Per RUN, not per process: a group renamed in ISE shows its new name # the next time the app starts. self._group_cache = {} def configured(self): return bool(self.base_url and self.username) def _headers(self, write=False): headers = { "Authorization": _basic_auth(self.username, self._password), "Accept": "application/json", } if write: headers["Content-Type"] = "application/json" return headers def _call(self, method, path, body=None, params=None): url = self.base_url + path if params: url += "?" + urllib.parse.urlencode(params) response = self.http.request(method, url, headers=self._headers(body is not None), body=body) if response.status in (401, 403): raise IseError("ISE rejected the ERS credentials (HTTP %s). The " "account needs the ERS Admin role and ERS must be " "enabled." % response.status) return response def endpoint_groups(self): response = self._call("GET", self.GROUPS_PATH, params={"size": 100}) if response.status != 200: raise IseError("ISE returned HTTP %s listing endpoint groups." % response.status) payload = response.json() or {} resources = ((payload.get("SearchResult") or {}).get("resources") or []) groups = [{"id": str(item.get("id", ""))[:64], "name": _one_line(item.get("name")), "description": _one_line(item.get("description"))} for item in resources if isinstance(item, dict)] for group in groups: if group["id"]: self._group_cache[group["id"]] = group return groups def group(self, group_id): """Resolve an endpoint group id to {id, name, description}, or None. The dropdown's list is fetched once per run and cached here, so the common case costs no extra call at all. Anything that list did not contain -- a group created since, or one past the size=100 page -- is fetched by id. A miss is cached too: fifty MACs sitting in one group this app cannot name must cost ONE request, not fifty. """ key = str(group_id or "") if not key: return None if key in self._group_cache: return self._group_cache[key] record = None if ISE_ID_RE.match(key): try: response = self._call("GET", "%s/%s" % (self.GROUPS_PATH, key)) except IseError: response = None if response is not None and response.status == 200: payload = response.json() or {} item = (payload.get("EndPointGroup") or payload.get("endPointGroup") or payload) if isinstance(item, dict) and (item.get("name") or item.get("id")): record = {"id": str(item.get("id") or key)[:64], "name": _one_line(item.get("name")), "description": _one_line(item.get("description"))} self._group_cache[key] = record return record def group_name(self, group_id, fallback=""): """The group's NAME, or `fallback` when ISE would not give one.""" record = self.group(group_id) if record and record.get("name"): return record["name"] return fallback def _endpoint_record(self, item): """Shape one ERS endpoint resource, whatever depth it arrived at.""" return { "id": str(item.get("id") or "")[:64], "name": _one_line(item.get("name")), "description": _one_line(item.get("description")), "group_id": str(item.get("groupId") or "")[:64], "static_group_assignment": _ers_bool( item.get("staticGroupAssignment")), "profile_id": str(item.get("profileId") or "")[:64], } def _endpoint_detail(self, record): """Fill in what the filtered search left out. Never raises. A detail read that fails leaves the record exactly as it was, and the plan then says the identity group is unknown rather than inventing one. A wrong group name here would be worse than the UUID this replaced. """ if not ISE_ID_RE.match(record.get("id") or ""): return record try: response = self._call("GET", "%s/%s" % (self.ENDPOINT_PATH, record["id"])) except IseError: return record if response.status != 200: return record payload = response.json() or {} item = payload.get("ERSEndPoint") or payload.get("ersEndPoint") if not isinstance(item, dict): return record for key, value in self._endpoint_record(item).items(): if value != "" and value is not None: record[key] = value return record def find_endpoint(self, mac, detail=True): """The endpoint record for one MAC, or None if ISE does not have it. ERS answers a filtered endpoint search with id and name and little else on most versions -- no `groupId` -- so when the search result does not carry the group, the endpoint is read by id to get it. That second read is what makes "identity group 'Printers'" possible at all; without it there is a UUID, or nothing. """ response = self._call("GET", self.ENDPOINT_PATH, params={"filter": "mac.EQ.%s" % mac}) if response.status != 200: raise IseError("ISE returned HTTP %s looking up %s." % (response.status, mac)) payload = response.json() or {} resources = ((payload.get("SearchResult") or {}).get("resources") or []) for item in resources: if isinstance(item, dict) and item.get("id"): record = self._endpoint_record(item) if detail and not record["group_id"]: record = self._endpoint_detail(record) return record return None def lookup(self, mac): """Read-only: where one MAC sits in ISE, named, as a sentence. Nothing here writes, so this is safe to call from any read path. """ record = self.find_endpoint(mac) group_id = (record or {}).get("group_id", "") group = self.group(group_id) if group_id else None name = (group or {}).get("name", "") return { "mac": mac, "found": bool(record), "endpoint": record, "group_id": group_id, "group_name": name, "group_description": (group or {}).get("description", ""), "message": describe_ise_endpoint(mac, record, name), } # -- the write paths ---------------------------------------------------- def require_group_id(self, group_id): """Refuse a missing or malformed target group before any call.""" if not group_id or not ISE_ID_RE.match(str(group_id)): raise IseError("Pick an endpoint group first.") return str(group_id) def plan_push(self, macs, group_id, group_name="", description=""): """What a push WOULD do, per MAC. For each address: whether ISE already has it, the NAME of the identity group it is in today, whether this push is a create, an update in place, or a MOVE between two named groups, and which transport the batch will use. `summary` on each action is the sentence the UI and the CLI both print, so all three surfaces say the same thing. """ target_name = self.group_name(group_id, group_name) actions = [] move_pairs = [] for mac in macs: existing = self.find_endpoint(mac) current_group_id = (existing or {}).get("group_id", "") current_group_name = (self.group_name(current_group_id) if current_group_id else "") action = { "mac": mac, "action": "update" if existing else "create", "endpoint_id": existing["id"] if existing else "", "current": existing, "current_group_id": current_group_id, "current_group_name": current_group_name, "group_id": group_id, "moves": bool(existing) and current_group_id != str(group_id), } action["summary"] = describe_ise_action(action, target_name) actions.append(action) if action["moves"]: pair = ("'%s' -> '%s'" % (current_group_name or current_group_id or "no group", target_name or group_id)) if pair not in move_pairs: move_pairs.append(pair) use_bulk = len(actions) >= self.bulk_threshold creates = sum(1 for a in actions if a["action"] == "create") moves = sum(1 for a in actions if a["moves"]) updates = len(actions) - creates plan = { "group_id": group_id, "group_name": target_name or group_name, "target_group_name": target_name, "description": description, "count": len(actions), "creates": creates, "updates": updates, "moves": moves, "move_pairs": move_pairs, "transport": "bulk" if use_bulk else "per-endpoint", "bulk_threshold": self.bulk_threshold, "endpoint_url": self.base_url + (self.BULK_PATH if use_bulk else self.ENDPOINT_PATH), "actions": actions, } plan["summary"] = _plan_summary(plan) return plan def _endpoint_body(self, mac, group_id, description=""): return {"ERSEndPoint": { "name": mac, "description": description[:128], "mac": mac, "groupId": group_id, "staticGroupAssignment": True, "staticProfileAssignment": False, }} def _bulk_body(self, actions, group_id, description=""): # ERS bulk submit takes an operation plus the resource list. return {"ns3:endpointBulkRequest": { "@operationType": "create", "@resourceMediaType": "vnd.com.cisco.ise.identity.endpoint.1.0+xml", "ns3:resourcesList": { "ns3:resource": [self._endpoint_body(action["mac"], group_id, description)["ERSEndPoint"] for action in actions], }, }} def push(self, macs, group_id, group_name="", description="", dry_run=True, audit=None): """Create-or-update every MAC in `group_id`. Returns a result envelope. When dry_run is true this composes the exact bodies and returns them WITHOUT sending anything: the http client sees only the read calls the plan needed. `audit` is called once per write ISE ACCEPTED, from inside the loop, immediately after that write returns. It is not called at the end and it is not conditional on the whole batch succeeding: a half-applied batch is precisely the case an audit log exists for, and gating on "everything worked" meant a partial batch recorded nothing at all. Definition of "one write": one request. The per-endpoint path makes one request per MAC and so writes one line per MAC. The bulk path submits the whole list as a single request that ISE accepts or refuses as a unit, and reports no per-endpoint outcome, so it writes one line carrying the MACs the request contained. """ self.require_group_id(group_id) plan = self.plan_push(macs, group_id, group_name, description) if dry_run: sample = [self._endpoint_body(a["mac"], group_id, description) for a in plan["actions"][:3]] return {"ok": True, "dry_run": True, "plan": plan, "sample_bodies": sample, "lines": [a["summary"] for a in plan["actions"]], "message": "Dry run: nothing was sent. " + plan["summary"]} results = [] if plan["transport"] == "bulk": response = self._call("PUT", self.BULK_PATH, body=self._bulk_body(plan["actions"], group_id, description)) reason = ("" if response.status not in (200, 201, 202) else ers_error(response)) ok = response.status in (200, 201, 202) and not reason results.append({"transport": "bulk", "status": response.status, "ok": ok, "count": plan["count"], "error": reason}) if ok and audit: audit({"op": "ise_endpoint_bulk_submit", "transport": "bulk", "status": response.status, "mac_count": plan["count"], "macs": [item["mac"] for item in plan["actions"]], "creates": plan["creates"], "updates": plan["updates"], "to_group_name": plan["group_name"], "moves": plan["moves"], "move_pairs": plan["move_pairs"]}) else: for action in plan["actions"]: body = self._endpoint_body(action["mac"], group_id, description) if action["action"] == "update": body["ERSEndPoint"]["id"] = action["endpoint_id"] response = self._call( "PUT", "%s/%s" % (self.ENDPOINT_PATH, action["endpoint_id"]), body=body) else: response = self._call("POST", self.ENDPOINT_PATH, body=body) reason = ("" if response.status not in (200, 201, 202, 204) else ers_error(response)) item_ok = (response.status in (200, 201, 202, 204) and not reason) results.append({"mac": action["mac"], "action": action["action"], "status": response.status, "ok": item_ok, "moved": action["moves"], "from_group_name": action["current_group_name"], "to_group_name": plan["group_name"], "summary": action["summary"], "error": reason}) # In the loop, right after the write landed. Not at the end. # `reason` is why a 200 can still be a failure here: ISE says # no in the body and keeps the status code. if item_ok and audit: audit({"op": "ise_endpoint_" + action["action"], "transport": "per-endpoint", "mac": action["mac"], "status": response.status, "endpoint_id": action["endpoint_id"], "moved": action["moves"], "from_group_id": action["current_group_id"], "from_group_name": action["current_group_name"], "to_group_name": plan["group_name"]}) succeeded = sum(1 for item in results if item["ok"]) # Only count the moves that actually LANDED. The per-endpoint path # knows which ones did; the bulk path is one request ISE takes or # refuses as a unit, so it is all of them or none. if plan["transport"] == "bulk": moved = plan["moves"] if (results and results[0]["ok"]) else 0 else: landed = set(item["mac"] for item in results if item["ok"]) moved = sum(1 for a in plan["actions"] if a["moves"] and a["mac"] in landed) message = ("%d of %d write(s) succeeded into identity group '%s'." % (succeeded, len(results), plan["group_name"] or plan["group_id"])) if moved: message += (" %d of them changed identity group: %s." % (moved, "; ".join(plan["move_pairs"]))) return {"ok": succeeded == len(results), "dry_run": False, "plan": plan, "results": results, "moved": moved, "lines": [a["summary"] for a in plan["actions"]], "message": message} # --------------------------------------------------------------------------- # Windows DHCP reservations over SSH + PowerShell (Pro) # --------------------------------------------------------------------------- class DhcpError(Exception): pass # Every value that reaches the PowerShell command line is matched against one # of these. This is an allowlist, not an escaper: a value that does not match # is refused outright rather than quoted and hoped for. Nothing here permits # a space, a quote, a semicolon, a backtick, a dollar sign, or a newline, so # there is no character left that could start a second PowerShell command. DHCP_ARG_RULES = { "scope_id": re.compile(r"^(?:\d{1,3}\.){3}\d{1,3}$"), "ip_address": re.compile(r"^(?:\d{1,3}\.){3}\d{1,3}$"), "client_id": re.compile(r"^[0-9a-f]{2}(?:-[0-9a-f]{2}){5}$"), "name": re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,62}$"), "description": re.compile(r"^[A-Za-z0-9][A-Za-z0-9 ._-]{0,62}$"), "computer_name": re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,62}$"), } def validate_dhcp_reservation(reservation): """(clean dict, errors). Every field must match its allowlist exactly.""" cleaned = {} errors = [] for field in ("scope_id", "ip_address", "client_id"): value = str(reservation.get(field, "") or "").strip() if field == "client_id": mac = normalize_mac(value) value = mac_for_dhcp_clientid(mac) if mac else value.lower() if not DHCP_ARG_RULES[field].match(value): errors.append("%s: %r is not acceptable" % (field, value[:64])) continue cleaned[field] = value for field in ("name", "description", "computer_name"): value = str(reservation.get(field, "") or "").strip() if not value: continue if not DHCP_ARG_RULES[field].match(value): errors.append("%s: %r is not acceptable" % (field, value[:64])) continue cleaned[field] = value for field in ("scope_id", "ip_address"): if field not in cleaned: continue octets = cleaned[field].split(".") if any(int(octet) > 255 for octet in octets): errors.append("%s: %s is not a valid IPv4 address" % (field, cleaned[field])) return cleaned, errors def build_dhcp_command(reservation, dhcp_server=""): """Compose the PowerShell one-liner. EVERY argument is allowlisted here. `dhcp_server` comes from config.env rather than from HTTP or a CSV, so it is not attacker-reachable, but it was the one value that reached PowerShell without passing the allowlist while the validated `computer_name` from the row was silently thrown away. Both now go through the same rule, and the per-row value wins. """ parts = ["Add-DhcpServerv4Reservation", "-ScopeId", reservation["scope_id"], "-IPAddress", reservation["ip_address"], "-ClientId", reservation["client_id"]] if reservation.get("name"): parts += ["-Name", reservation["name"]] if reservation.get("description"): # The only allowlisted field that may contain a space, so it is the # only one that gets quotes, and it cannot contain a quote. parts += ["-Description", "'%s'" % reservation["description"]] computer = reservation.get("computer_name") or dhcp_server if computer: if not DHCP_ARG_RULES["computer_name"].match(str(computer)): raise DhcpError( "%r is not an acceptable DHCP server name. Letters, digits, " "dot, dash and underscore only." % str(computer)[:64]) parts += ["-ComputerName", computer] parts += ["-ErrorAction", "Stop"] return " ".join(parts) class DhcpClient(object): """Runs the composed PowerShell over the system ssh client. `runner` is injected so the test suite drives every path with a fake host; nothing here shells out during tests. """ def __init__(self, host, username, password="", port=22, dhcp_server="", runner=None, timeout=45): self.host = host self.username = username self._password = password self.port = int(port or 22) self.dhcp_server = dhcp_server self.timeout = timeout self._runner = runner or self._ssh_run def configured(self): return bool(self.host and self.username) def ssh_argv(self, command): return ["ssh", "-p", str(self.port), "-o", "StrictHostKeyChecking=accept-new", "-o", "ConnectTimeout=15", "-o", "NumberOfPasswordPrompts=1", "-l", self.username, self.host, "powershell -NoProfile -NonInteractive -Command %s" % command] def _ssh_run(self, command): # pragma: no cover - needs a real host if shutil.which("ssh") is None: raise DhcpError("No `ssh` client found on PATH.") try: completed = subprocess.run( self.ssh_argv(command), stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=self.timeout, check=False) except (OSError, subprocess.SubprocessError) as error: raise DhcpError("ssh failed: %s" % error) return completed.returncode, completed.stdout.decode("utf-8", "replace") def add_reservations(self, reservations, dry_run=True, audit=None): """Validate everything first, then (unless dry run) apply in order. `audit` is called once per reservation the host actually created, from inside the loop, right after that command returned 0. A row that failed gets no line, and a partially-applied batch still records every row that landed. """ planned = [] problems = [] if self.dhcp_server and not DHCP_ARG_RULES["computer_name"].match( str(self.dhcp_server)): return {"ok": False, "error": "DHCP_SERVER in config.env is not an acceptable " "computer name (letters, digits, dot, dash, " "underscore only). Nothing was sent.", "problems": ["DHCP_SERVER: %r is not acceptable" % str(self.dhcp_server)[:64]], "planned": []} for index, reservation in enumerate(reservations, start=1): cleaned, errors = validate_dhcp_reservation(reservation) if errors: problems.extend(["row %d: %s" % (index, error) for error in errors]) continue try: command = build_dhcp_command(cleaned, self.dhcp_server) except DhcpError as error: problems.append("row %d: %s" % (index, error)) continue planned.append({"reservation": cleaned, "command": command}) if problems: return {"ok": False, "error": "Fix the input first.", "problems": problems, "planned": planned} if not planned: return {"ok": False, "error": "Nothing to do.", "problems": [], "planned": []} if dry_run: return {"ok": True, "dry_run": True, "planned": planned, "message": "Dry run: nothing was sent. %d reservation(s) " "would be created on %s." % (len(planned), self.host)} results = [] for item in planned: try: code, output = self._runner(item["command"]) except DhcpError as error: results.append({"ip": item["reservation"]["ip_address"], "ok": False, "output": str(error)}) continue item_ok = code == 0 results.append({"ip": item["reservation"]["ip_address"], "ok": item_ok, "output": output[-2000:]}) if item_ok and audit: audit({"op": "dhcp_reservation_add", "reservation": item["reservation"]}) succeeded = sum(1 for item in results if item["ok"]) return {"ok": succeeded == len(results), "dry_run": False, "planned": planned, "results": results, "message": "%d of %d reservation(s) created." % (succeeded, len(results))} # --------------------------------------------------------------------------- # The engine the UI and the CLI both call # --------------------------------------------------------------------------- class Engine(object): def __init__(self, config, license_client=None, catalyst=None, ise=None, dhcp=None): self.config = config self.license = license_client or LicenseClient(config) timeout = config.int("HTTP_TIMEOUT", 30) self.catalyst = catalyst if catalyst is not None else CatalystClient( config.get("CATALYST_URL"), config.get("CATALYST_USERNAME"), config.get("CATALYST_PASSWORD"), timeout=timeout, verify_tls=config.bool("CATALYST_VERIFY_TLS", True)) self.ise = ise if ise is not None else IseClient( config.get("ISE_URL"), config.get("ISE_USERNAME"), config.get("ISE_PASSWORD"), timeout=timeout, verify_tls=config.bool("ISE_VERIFY_TLS", True), bulk_threshold=config.int("ISE_BULK_THRESHOLD", 20)) self.dhcp = dhcp if dhcp is not None else DhcpClient( config.get("DHCP_HOST"), config.get("DHCP_SSH_USERNAME"), config.get("DHCP_SSH_PASSWORD"), config.int("DHCP_SSH_PORT", 22), config.get("DHCP_SERVER")) self._devices = [] # -- gating ------------------------------------------------------------- def pro(self): return self.license.state().pro def _require_pro(self): state = self.license.state() if state.pro: return None return {"ok": False, "pro_required": True, "error": "This is a Pro feature. %s" % state.reason} def _dry_run(self, requested): """Dry run is ON unless this call explicitly asked for a real write. `DRY_RUN` in config.env sets the DEFAULT for a call that does not say (the UI's checkbox always says). It is not a veto: the Network Configurator behaves the same way, and one of the two apps quietly refusing a real write would be worse than neither doing it. """ if requested is None: return self.config.bool("DRY_RUN", True) return bool(requested) def _account(self): return self.license.state().grant.get("account", "") def _audit_sink(self, action, context): """Build the per-write audit callback handed to a client. The client calls this once per write the controller accepted, from inside its own loop. Writing the line there rather than after the batch is the whole point of finding 2: a batch where 2 of 4 endpoints landed used to record nothing, because the envelope's `ok` means "every item succeeded". The account is resolved ONCE here rather than per line: state() verifies an Ed25519 signature and reads a file, which is not something to do inside a write loop. """ account = self._account() path = self.config.path("AUDIT_LOG") def sink(item): record = {"action": action, "account": account} record.update(context) record.update(item or {}) return append_audit(path, record) return sink # -- Lite widgets ------------------------------------------------------- def catalyst_devices(self, refresh=True): if not self.catalyst.configured(): return {"ok": False, "error": "Set CATALYST_URL and CATALYST_USERNAME " "in config.env."} if refresh or not self._devices: try: self._devices = self.catalyst.devices() except CatalystError as error: return {"ok": False, "error": str(error)} return {"ok": True, "devices": self._devices, "message": "%d device(s)." % len(self._devices)} def catalyst_ports(self, device_id): if not self.catalyst.configured(): return {"ok": False, "error": "Catalyst Center is not configured."} try: ports = self.catalyst.interfaces(device_id) except CatalystError as error: return {"ok": False, "error": str(error)} up = sum(1 for port in ports if port["status"].lower() == "up") return {"ok": True, "ports": ports, "message": "%d port(s), %d up." % (len(ports), up)} def catalyst_find_mac(self, mac): if not self.catalyst.configured(): return {"ok": False, "error": "Catalyst Center is not configured."} try: found = self.catalyst.find_mac(mac) except CatalystError as error: return {"ok": False, "error": str(error)} return {"ok": True, "result": found, "message": ("Found on %s %s." % (found["switch"], found["port"])) if found["found"] else "Not found."} def session_zip(self, site=""): devices = self._devices if not devices: outcome = self.catalyst_devices() if not outcome.get("ok"): return None, outcome devices = outcome["devices"] if site: devices = [device for device in devices if device.get("site") == site] return build_session_zip(devices, self.config.get("CATALYST_USERNAME", "")), \ {"ok": True, "count": len(devices)} # -- Pro widgets -------------------------------------------------------- def ise_groups(self): gate = self._require_pro() if gate: return gate if not self.ise.configured(): return {"ok": False, "error": "Set ISE_URL and ISE_USERNAME in config.env."} try: groups = self.ise.endpoint_groups() except IseError as error: return {"ok": False, "error": str(error)} return {"ok": True, "groups": groups, "message": "%d endpoint group(s)." % len(groups)} def ise_push(self, macs_text, group_id, group_name="", description="", dry_run=None, confirm=False): gate = self._require_pro() if gate: return gate if not self.ise.configured(): return {"ok": False, "error": "Cisco ISE is not configured."} macs, rejected = normalize_mac_list(macs_text) if not macs: return {"ok": False, "error": "No usable MAC addresses.", "rejected": rejected} dry = self._dry_run(dry_run) if not dry and not confirm: # The confirmation is the last thing between an engineer and a # change to live authorisation, so it is built from a real # read-only plan: it names the target group and, for anything # already in ISE, names the group the MAC is leaving. A confirm # box that said only "write 4 endpoints" hid every move. try: self.ise.require_group_id(group_id) plan = self.ise.plan_push(macs, group_id, group_name, description) except IseError as error: return {"ok": False, "error": str(error), "rejected": rejected} question = ("This would write %d endpoint(s) to ISE. %s Confirm to " "proceed." % (len(macs), plan["summary"])) return {"ok": False, "confirm_required": True, "rejected": rejected, "plan": plan, "lines": [a["summary"] for a in plan["actions"]], "error": question} try: resolved_name = self.ise.group_name(group_id, group_name) except IseError: resolved_name = group_name audit = None if dry else self._audit_sink("ise_mab_push", { "target": self.ise.base_url, "group_id": group_id, "group_name": resolved_name or group_name, }) try: outcome = self.ise.push(macs, group_id, resolved_name or group_name, description, dry_run=dry, audit=audit) except IseError as error: return {"ok": False, "error": str(error), "rejected": rejected} outcome["rejected"] = rejected return outcome def ise_lookup(self, mac): """Read-only: which ISE identity group one MAC is configured for. Pro, because it is the ISE widget's credential and the ISE widget's endpoint. Nothing here writes. """ gate = self._require_pro() if gate: return gate if not self.ise.configured(): return {"ok": False, "error": "Cisco ISE is not configured."} normalised = normalize_mac(mac) if normalised is None: return {"ok": False, "error": "That is not a MAC address this app recognises."} try: outcome = self.ise.lookup(normalised) except IseError as error: return {"ok": False, "error": str(error)} outcome["ok"] = True return outcome def find_mac_report(self, mac): """The Catalyst Center find-a-MAC result, plus where ISE has it. Additive: the Catalyst shape is untouched and an `ise` key is added when an ISE node is configured, so an existing `find-mac | jq` keeps working. Two controllers answer about the same address in one place, which is what anyone tracing a MAB failure actually wants. """ report = self.catalyst_find_mac(mac) if self.ise.configured(): report["ise"] = self.ise_lookup(mac) return report def dhcp_reservations(self, rows, dry_run=None, confirm=False): gate = self._require_pro() if gate: return gate if not self.dhcp.configured(): return {"ok": False, "error": "Set DHCP_HOST and DHCP_SSH_USERNAME " "in config.env."} dry = self._dry_run(dry_run) if not dry and not confirm: return {"ok": False, "confirm_required": True, "error": "This would create %d reservation(s) on %s. " "Confirm to proceed." % (len(rows), self.dhcp.host)} audit = None if dry else self._audit_sink("dhcp_reservation_add", { "host": self.dhcp.host, "dhcp_server": self.dhcp.dhcp_server, }) try: outcome = self.dhcp.add_reservations(rows, dry_run=dry, audit=audit) except DhcpError as error: return {"ok": False, "error": str(error)} return outcome def parse_reservation_csv(self, text): """scope_id,ip_address,client_id,name,description - headers optional.""" rows = [] problems = [] if "\x00" in (text or ""): return [], ["That paste contains a NUL byte, so it is not plain " "CSV text."] try: parsed = list(enumerate(csv.reader(io.StringIO(text or "")), start=1)) except csv.Error as error: return [], ["Could not read that as CSV: %s" % error] for index, raw in parsed: cells = [cell.strip() for cell in raw if cell is not None] if not cells or not any(cells): continue if index == 1 and cells[0].lower().replace(" ", "_") in ( "scope_id", "scopeid", "scope"): continue if len(cells) < 3: problems.append("row %d: need at least scope, ip, mac" % index) continue rows.append({ "scope_id": cells[0], "ip_address": cells[1], "client_id": cells[2], "name": cells[3] if len(cells) > 3 else "", "description": cells[4] if len(cells) > 4 else "", "computer_name": cells[5] if len(cells) > 5 else "", }) return rows, problems # -- status ------------------------------------------------------------- def status(self): state = self.license.state() return { "app": APP_NAME, "title": APP_TITLE, "version": APP_VERSION, "python": "%d.%d.%d" % sys.version_info[:3], "license": state.as_dict(), "dry_run_default": self.config.bool("DRY_RUN", True), "audit_log": self.config.path("AUDIT_LOG"), "catalyst_configured": self.catalyst.configured(), "catalyst_url": self.catalyst.base_url, "ise_configured": self.ise.configured(), "ise_url": self.ise.base_url, "ise_bulk_threshold": self.ise.bulk_threshold, "dhcp_configured": self.dhcp.configured(), "dhcp_host": self.dhcp.host, "device_count": len(self._devices), "config": self.config.redacted(), } # --------------------------------------------------------------------------- # Local web UI # --------------------------------------------------------------------------- PAGE_HTML = """ __TITLE__

__TITLE__

v__VERSION__ Lite

Status

Every call below goes from this machine straight to your controllers. The only thing that reaches ciscotools.dev is the license check: your API key and this app's version, nothing else.

Catalyst Center devices Lite · read only

Inventory from the Intent API. Pick a device to see its ports.

Port status Lite · read only

Interface state for the selected device.

Find a MAC Lite · read only

Any spelling: 0011.2233.4455, 00:11:22:33:44:55, 001122334455.

ISE MAB endpoints Pro · writes

Paste MACs in any format. They are normalised, deduped, and pushed create-or-update. At 20 or more the bulk path is used.

Windows DHCP reservations Pro · writes

One row per reservation: scope, IP, MAC, name, description. Every argument is checked against an allowlist before it reaches PowerShell.

""" class Handler(BaseHTTPRequestHandler): server_version = "%s/%s" % (APP_TITLE.replace(" ", ""), APP_VERSION) engine = None token = "" def log_message(self, fmt, *args): # quieter, and never logs a body sys.stderr.write("[%s] %s %s\n" % (time.strftime("%H:%M:%S"), self.command, self.path.split("?")[0])) def _host_is_local(self): """Refuse a request whose Host header is not loopback. This is the DNS-rebinding guard: a page on the internet can make your browser send requests to 127.0.0.1, but it cannot control the Host header it sends, so requiring loopback there keeps a random web page from driving your network tooling. """ host = (self.headers.get("Host") or "").rsplit(":", 1)[0].strip("[]") # An absent Host header is refused too: every browser and every curl # sends one, so a request without it is not a case worth accommodating. return host in ("127.0.0.1", "localhost", "::1") def _send(self, status, body, content_type="application/json", extra_headers=None): if isinstance(body, (dict, list)): body = json.dumps(body).encode("utf-8") elif isinstance(body, str): body = body.encode("utf-8") self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(body))) self.send_header("X-Content-Type-Options", "nosniff") self.send_header("Referrer-Policy", "no-referrer") self.send_header("Content-Security-Policy", "default-src 'none'; style-src 'unsafe-inline'; " "script-src 'unsafe-inline'; connect-src 'self'") for name, value in (extra_headers or {}).items(): self.send_header(name, value) self.end_headers() if self.command != "HEAD": self.wfile.write(body) def _body(self): try: length = int(self.headers.get("Content-Length") or 0) except ValueError: return {} if length <= 0 or length > 4 * 1024 * 1024: return {} try: return json.loads(self.rfile.read(length).decode("utf-8")) except (ValueError, UnicodeError): return {} def _authorised(self): given = self.headers.get("X-NetOps-Token", "") return bool(self.token) and secrets.compare_digest(given, self.token) def do_GET(self): if not self._host_is_local(): return self._send(403, {"ok": False, "error": "loopback only"}) path, _, query = self.path.partition("?") params = urllib.parse.parse_qs(query) if path == "/": page = (PAGE_HTML.replace("__TITLE__", APP_TITLE) .replace("__VERSION__", APP_VERSION) .replace("__TOKEN__", self.token)) return self._send(200, page, "text/html; charset=utf-8") if not self._authorised(): return self._send(403, {"ok": False, "error": "bad token"}) if path == "/api/status": return self._send(200, self.engine.status()) if path == "/api/sessions.zip": blob, info = self.engine.session_zip(params.get("site", [""])[0]) if blob is None: return self._send(200, info) return self._send(200, blob, "application/zip", { "Content-Disposition": 'attachment; filename="securecrt-sessions.zip"', "X-Device-Count": str(info.get("count", 0))}) return self._send(404, {"ok": False, "error": "not found"}) def do_POST(self): if not self._host_is_local(): return self._send(403, {"ok": False, "error": "loopback only"}) if not self._authorised(): return self._send(403, {"ok": False, "error": "bad token"}) path = self.path.partition("?")[0] body = self._body() engine = self.engine try: if path == "/api/license/refresh": ok, message = engine.license.refresh() return self._send(200, {"ok": ok, "message" if ok else "error": message}) if path == "/api/catalyst/devices": return self._send(200, engine.catalyst_devices()) if path == "/api/catalyst/ports": return self._send(200, engine.catalyst_ports(body.get("device_id", ""))) if path == "/api/catalyst/find-mac": return self._send(200, engine.catalyst_find_mac(body.get("mac", ""))) if path == "/api/ise/groups": return self._send(200, engine.ise_groups()) if path == "/api/ise/push": return self._send(200, engine.ise_push( body.get("macs", ""), body.get("group_id", ""), body.get("group_name", ""), body.get("description", ""), dry_run=body.get("dry_run"), confirm=bool(body.get("confirm")))) if path == "/api/dhcp/reservations": rows = body.get("rows") problems = [] if rows is None: rows, problems = engine.parse_reservation_csv(body.get("csv", "")) outcome = engine.dhcp_reservations( rows, dry_run=body.get("dry_run"), confirm=bool(body.get("confirm"))) if problems: outcome.setdefault("problems", []) outcome["problems"] = problems + list(outcome["problems"]) return self._send(200, outcome) except Exception as error: # noqa: BLE001 - never leak a traceback return self._send(200, {"ok": False, "error": "%s" % error}) return self._send(404, {"ok": False, "error": "not found"}) def serve(engine, host="127.0.0.1", port=8782, open_browser=True, token=None): # pragma: no cover - exercised by hand, not in CI token = token or secrets.token_urlsafe(24) handler = type("BoundHandler", (Handler,), {"engine": engine, "token": token}) if host not in ("127.0.0.1", "localhost", "::1"): print("Refusing to bind %s: this app is loopback only." % host) host = "127.0.0.1" for candidate in range(port, port + 20): try: httpd = ThreadingHTTPServer((host, candidate), handler) break except OSError: continue else: print("Could not find a free port near %d." % port) return 1 url = "http://%s:%d/" % (host, httpd.server_address[1]) print("%s %s" % (APP_TITLE, APP_VERSION)) print(" UI: %s" % url) print(" Dry run: %s" % ("on" if engine.config.bool("DRY_RUN", True) else "OFF")) print(" License: %s" % engine.license.state().reason) print(" Ctrl-C to stop. Nothing listens outside this machine.") if open_browser: threading.Timer(0.4, lambda: webbrowser.open(url)).start() try: httpd.serve_forever() except KeyboardInterrupt: print("\nStopped.") finally: httpd.server_close() return 0 # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def _print_json(payload): print(json.dumps(payload, indent=2, sort_keys=True)) def main(argv=None): # pragma: no cover - thin wrapper over tested pieces parser = argparse.ArgumentParser( prog=APP_NAME, description="%s %s - local widget dashboard. Runs on your machine." % (APP_TITLE, APP_VERSION)) parser.add_argument("--config", default="config.env") parser.add_argument("--no-browser", action="store_true") parser.add_argument("--version", action="store_true") sub = parser.add_subparsers(dest="command") sub.add_parser("status", help="print status as JSON and exit") sub.add_parser("license", help="refresh the license and exit") sub.add_parser("devices", help="print the Catalyst Center inventory and exit") mac_parser = sub.add_parser("find-mac", help="locate a MAC and exit") mac_parser.add_argument("mac") export_parser = sub.add_parser("export-sessions", help="write a SecureCRT sessions zip and exit") export_parser.add_argument("--out", default="securecrt-sessions.zip") args = parser.parse_args(argv) if args.version: print("%s %s" % (APP_TITLE, APP_VERSION)) return 0 config_path = args.config if not os.path.isabs(config_path): here = os.path.dirname(os.path.abspath(__file__)) candidate = os.path.join(here, config_path) config_path = candidate if os.path.exists(candidate) else os.path.abspath(config_path) config = Config.load(config_path) interactive = args.command is None and sys.stdin.isatty() config.prompt_for_missing_secrets(interactive=interactive) engine = Engine(config) if args.command == "status": _print_json(engine.status()) return 0 if args.command == "license": ok, message = engine.license.refresh() print(("OK: " if ok else "Not unlocked: ") + message) return 0 if ok else 1 if args.command == "devices": _print_json(engine.catalyst_devices()) return 0 if args.command == "find-mac": report = engine.find_mac_report(args.mac) # The human sentence goes to stderr so stdout stays pure JSON for # `| jq`; it is in the JSON too, under ise.message. ise = report.get("ise") or {} if ise.get("message") or ise.get("error"): sys.stderr.write("ISE: %s\n" % (ise.get("message") or ise["error"])) _print_json(report) return 0 if args.command == "export-sessions": blob, info = engine.session_zip() if blob is None: _print_json(info) return 1 with open(args.out, "wb") as handle: handle.write(blob) print("Wrote %s (%d device(s))." % (args.out, info.get("count", 0))) return 0 return serve(engine, config.get("BIND_HOST", "127.0.0.1"), config.int("BIND_PORT", 8782), open_browser=config.bool("OPEN_BROWSER", True) and not args.no_browser) if __name__ == "__main__": # pragma: no cover sys.exit(main())