commit 81f6afd69726ea17e56bb739854891bb7235bbfc Author: Outis Date: Mon Mar 23 08:52:01 2026 -0400 Add files via upload diff --git a/Avaya_5952_setup.py b/Avaya_5952_setup.py new file mode 100644 index 0000000..aeb6cd5 --- /dev/null +++ b/Avaya_5952_setup.py @@ -0,0 +1,1456 @@ +#!/usr/bin/env python3 +""" +Avaya / Extreme Avaya / Extreme ERS 5952 Switch Manager — One File Setup +───────────────────────────────────────── +Copy this single file to your always on management computer +(Raspberry Pi, HP T620 thin client, old computer or laptop) and run: + + python Avaya_5952_setup.py + +It will: + - Install Python dependencies + - Generate SSH keypair + - Guide you through loading the key on the switch (one manual step) + - Test SSH connectivity + - Pin the switch host key + - Set up TOTP authenticator + - Write switch_backend.py and build the React frontend + - Install and start the systemd service + - Open the browser + +Re-running is safe — skips steps already done. +───────────────────────────────────────── +""" + +import os, sys, time, shutil, subprocess, textwrap, socket +from pathlib import Path + +# ── Colour helpers ────────────────────────────────────────────────────────── +def c(t, code): return f"\033[{code}m{t}\033[0m" +def green(t): return c(t, "32") +def yellow(t): return c(t, "33") +def red(t): return c(t, "31") +def cyan(t): return c(t, "36") +def bold(t): return c(t, "1") +def dim(t): return c(t, "2") +def ok(m): print(f" {green(chr(10003))} {m}") +def warn(m): print(f" {yellow(chr(9888))} {m}") +def err(m): print(f" {red(chr(10007))} {m}") +def info(m): print(f" {cyan(chr(9672))} {m}") +def step(n,m): print(f"\n{bold(cyan(f'[{n}]'))} {bold(m)}") +def sep(): print(dim(" " + chr(8212)*58)) + +def ask(prompt, default=""): + try: + v = input(f" {cyan(chr(63))} {prompt}" + (f" [{default}]" if default else "") + ": ").strip() + return v if v else default + except KeyboardInterrupt: + print("\n\nAborted."); sys.exit(0) + +def ask_yn(prompt, default=True): + v = ask(f"{prompt} ({'Y/n' if default else 'y/N'})", "").lower() + return default if not v else v.startswith("y") + +def pause(m="Press Enter to continue..."): + try: input(f"\n {yellow(chr(9654))} {m}") + except KeyboardInterrupt: print("\n\nAborted."); sys.exit(0) + +def sudo(cmd): + prefix = "sudo " if os.geteuid() != 0 else "" + return subprocess.run(f"{prefix}{cmd}", shell=True, capture_output=True, text=True) + +def run(cmd, capture=True, check=False): + return subprocess.run(cmd, shell=True, capture_output=capture, text=True, check=check) + +def local_ip(): + try: + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + s.connect(("8.8.8.8", 80)) + ip = s.getsockname()[0]; s.close(); return ip + except: return "localhost" + +# ── Paths ──────────────────────────────────────────────────────────────────── +HERE = Path(__file__).parent.absolute() +CONF_DIR = Path("/etc/switch-manager") +KEY_PATH = CONF_DIR / "ers5952_key" +KNOWN_HOSTS = CONF_DIR / "known_hosts" +TOTP_FILE = CONF_DIR / "totp_secret" +BACKEND = HERE / "switch_backend.py" +FRONTEND = HERE / "frontend" +DIST = FRONTEND / "dist" +APP_JSX = HERE / "ers5952-manager.jsx" +SERVICE = Path("/etc/systemd/system/switch-manager.service") +PACKAGES = ["fastapi", "uvicorn", "paramiko", "pyotp"] +# ── Console software download links ─────────────────────────────────────────── +CONSOLE_SOFTWARE = { + "PuTTY (Windows)": "https://www.putty.org", + "TeraTerm (Windows)": "https://github.com/TeraTermProject/teraterm/releases", + "Serial (Mac)": "https://apps.apple.com/app/serial/id877615577", + "CoolTerm (Mac/Win)": "https://freeware.the-meiers.org", + "CP2102 USB driver": "https://www.silabs.com/developers/usb-to-uart-bridge-vcp-drivers", + "FTDI USB driver": "https://ftdichip.com/drivers/vcp-drivers/", +} + +CONSOLE_CABLE_SEARCH = "RJ45 console cable USB Cisco compatible" + + + +BACKEND_SRC = '"""\nERS 5952 Switch Manager — Backend v3\n────────────────────────────────────────────────────────────────────────\nChanges from v2:\n - Connection pool (30s lifetime, liveness check, auto-invalidate)\n - Visitor-aware polling — starts when someone hits the page,\n pauses when no active visitors, stops completely when idle\n - Hard-block danger patterns (no force override for lethal commands)\n - Session-based TOTP (one auth → session token → many pushes)\n - Per-command execution with stop-on-error\n - Config saved only on full success\n\nRequirements:\n pip install fastapi uvicorn paramiko pyotp\n\nRun:\n python switch_backend.py # start server\n python switch_backend.py --setup-totp # first-time TOTP setup\n────────────────────────────────────────────────────────────────────────\n"""\n\nimport re, sys, time, secrets, logging, threading, os\nfrom pathlib import Path\nfrom typing import Optional\n\nimport paramiko, pyotp\nfrom fastapi import FastAPI, HTTPException, Request\nfrom fastapi.middleware.cors import CORSMiddleware\nfrom fastapi.responses import JSONResponse\nfrom fastapi.staticfiles import StaticFiles\nfrom pydantic import BaseModel, field_validator\n\nlogging.basicConfig(level=logging.INFO,\n format="%(asctime)s [%(levelname)s] %(message)s")\nlog = logging.getLogger("switch-manager")\n\n# ══════════════════════════════════════════════════════════════════════\n# CONFIG\n# ══════════════════════════════════════════════════════════════════════\nSWITCH_HOST = "192.168.99.1"\nSWITCH_PORT = 22\nSWITCH_USER = "admin"\nKEY_PATH = "/etc/switch-manager/ers5952_key"\nKNOWN_HOSTS = "/etc/switch-manager/known_hosts"\nTOTP_FILE = "/etc/switch-manager/totp_secret"\nSTATIC_DIR = "./frontend/dist"\n\nPOLL_ACTIVE_S = 15 # poll interval when visitors present\nPOLL_BG_S = 60 # poll interval when tab backgrounded\nPOLL_IDLE_AFTER = 300 # stop polling after this many seconds with no visitors\nCONN_POOL_MAX_S = 25 # max connection age before refresh (< switch idle timeout)\nSESSION_TTL_S = 600 # TOTP session — 10 minutes, reset on activity\n\nALLOWED_ORIGINS = ["*"] # tighten to http://192.168.99.X in production\n\n# ══════════════════════════════════════════════════════════════════════\n# TOTP\n# ══════════════════════════════════════════════════════════════════════\n\ndef get_or_create_totp_secret() -> str:\n p = Path(TOTP_FILE)\n if p.exists():\n return p.read_text().strip()\n secret = pyotp.random_base32()\n p.parent.mkdir(parents=True, exist_ok=True)\n p.write_text(secret)\n p.chmod(0o600)\n log.info(f"New TOTP secret created at {TOTP_FILE}")\n return secret\n\ndef setup_totp():\n secret = get_or_create_totp_secret()\n uri = pyotp.TOTP(secret).provisioning_uri(\n name="ERS5952", issuer_name="SwitchManager")\n print("\\n══════════════════════════════════════════════════")\n print(" ERS 5952 Switch Manager — TOTP Setup")\n print("══════════════════════════════════════════════════")\n print(f"\\n Manual entry secret:\\n {secret}")\n print(f"\\n Provisioning URI (paste into authenticator app):\\n {uri}")\n print("\\n Or generate a QR code:")\n print(f" python -c \\"import qrcode; qrcode.make(\'{uri}\').show()\\"")\n print("\\n══════════════════════════════════════════════════\\n")\n\nTOTP_SECRET: str = ""\n\n# ── Session tokens (one TOTP → session, multiple pushes) ──────────────\n# { token: { expires: float, last_activity: float } }\n_sessions: dict[str, dict] = {}\n_sessions_lock = threading.Lock()\n\ndef create_session() -> str:\n token = secrets.token_hex(32)\n now = time.time()\n with _sessions_lock:\n _sessions[token] = {"expires": now + SESSION_TTL_S, "last_activity": now}\n return token\n\ndef validate_session(token: str) -> bool:\n """Returns True and refreshes activity timestamp if session is valid."""\n now = time.time()\n with _sessions_lock:\n # Prune expired\n expired = [t for t, s in _sessions.items() if s["expires"] < now]\n for t in expired:\n del _sessions[t]\n if token not in _sessions:\n return False\n # Refresh on activity\n _sessions[token]["last_activity"] = now\n _sessions[token]["expires"] = now + SESSION_TTL_S\n return True\n\ndef session_remaining(token: str) -> Optional[int]:\n with _sessions_lock:\n s = _sessions.get(token)\n if not s:\n return None\n return max(0, int(s["expires"] - time.time()))\n\ndef revoke_session(token: str):\n with _sessions_lock:\n _sessions.pop(token, None)\n\ndef require_session(token: str):\n if not validate_session(token):\n raise HTTPException(401, "Session expired or invalid — re-authenticate with TOTP")\n\n# ══════════════════════════════════════════════════════════════════════\n# DANGER DETECTION\n# Two tiers:\n# HARD_BLOCK — refused entirely, must run at console\n# WARN — flagged, push still offered with explicit override\n# ══════════════════════════════════════════════════════════════════════\n\nHARD_BLOCK_PATTERNS = [\n # Management VLAN removal — all variants\n (re.compile(r\'no\\s+vlan\\s+99\\b\', re.I), "Deletes management VLAN 99"),\n (re.compile(r\'vlan\\s+members\\s+remove\\b.*\\b99\\b\', re.I), "Removes VLAN 99 from a port — kills management trunk"),\n (re.compile(r\'no\\s+vlan\\s+tagging\\b.*\\b99\\b\', re.I), "Removes VLAN 99 tagging — kills management trunk"),\n (re.compile(r\'vlan\\s+pvid\\s+\\S+\\s+\\S+\', re.I), "Changes native VLAN — verify not management port"),\n # SSH / IP removal\n (re.compile(r\'no\\s+ip\\s+ssh\', re.I), "Disables SSH entirely — permanent lockout"),\n (re.compile(r\'no\\s+ip\\s+address\\b\', re.I), "Removes IP address — will lose connectivity"),\n # Management interface\n (re.compile(r\'interface\\s+vlan\\s+99\\b\', re.I), "Modifies management VLAN interface — run at console"),\n # Boot / factory\n (re.compile(r\'boot\\s+config\\s+flags\\s+factory\', re.I), "Factory reset — run at console"),\n]\n\nWARN_PATTERNS = [\n (re.compile(r\'\\bshutdown\\b\', re.I), "Shuts down an interface — confirm it is not your uplink"),\n (re.compile(r\'default\\s+interface\\b\', re.I), "Resets interface to defaults"),\n (re.compile(r\'no\\s+vlan\\s+\\d+\\b\', re.I), "Deletes a VLAN — confirm no active ports depend on it"),\n (re.compile(r\'spanning-tree\\s+.*\\s+disable\', re.I), "Disables spanning tree — loop risk"),\n]\n\ndef check_danger(commands: list[str]) -> dict:\n hard, warn = [], []\n for cmd in commands:\n for pat, reason in HARD_BLOCK_PATTERNS:\n if pat.search(cmd):\n hard.append({"command": cmd, "reason": reason})\n break\n else:\n for pat, reason in WARN_PATTERNS:\n if pat.search(cmd):\n warn.append({"command": cmd, "reason": reason})\n break\n return {\n "hard_blocked": hard,\n "warnings": warn,\n "has_hard_block": bool(hard),\n "has_warnings": bool(warn),\n }\n\n# ══════════════════════════════════════════════════════════════════════\n# SANITIZATION\n# ══════════════════════════════════════════════════════════════════════\n\n_BAD_CHARS = re.compile(r\'[;&|`$<>(){}\\\\"\\\']\')\n_BAD_PATS = [re.compile(p) for p in [r\'\\n\', r\'\\r\', r\'--\', r\'/\\*\']]\n_RE_VID = re.compile(r\'^\\d{1,4}$\')\n_RE_VNAME = re.compile(r\'^[a-zA-Z0-9\\-_]{1,32}$\')\n_RE_MODE = re.compile(r\'^(access|trunk|disabled)$\')\n_RE_ANAME = re.compile(r\'^[a-zA-Z0-9\\-_]{1,32}$\')\n_RE_DIR = re.compile(r\'^(in|out)$\')\n_RE_PROTO = re.compile(r\'^(ip|tcp|udp|icmp)$\')\n_RE_ACTION = re.compile(r\'^(permit|deny)$\')\n_RE_DESC = re.compile(r\'^[a-zA-Z0-9\\-_ ]{0,64}$\')\n\ndef _san(v: str, pat: re.Pattern, field: str) -> str:\n if _BAD_CHARS.search(v):\n raise ValueError(f"{field}: disallowed characters")\n for p in _BAD_PATS:\n if p.search(v):\n raise ValueError(f"{field}: injection pattern")\n if not pat.match(v):\n raise ValueError(f"{field}: invalid format — {repr(v)}")\n return v\n\ndef san_vid(v, field="vlan_id") -> int:\n _san(str(v), _RE_VID, field)\n vid = int(v)\n if not 1 <= vid <= 4094:\n raise ValueError(f"{field}: must be 1–4094")\n return vid\n\ndef san_port(v) -> int:\n p = int(v)\n if not 1 <= p <= 52:\n raise ValueError("port: must be 1–52")\n return p\n\n_ALLOWED_CMD_RE = [\n re.compile(r\'^vlan\\s+(create|members|tagging|pvid)\\s\'),\n re.compile(r\'^no\\s+vlan\\s+\\d+$\'),\n re.compile(r\'^interface\\s+(FastEthernet|GigabitEthernet|vlan)\\s\'),\n re.compile(r\'^\\s+(name|no\\s+shutdown|poe|speed|duplex|ip\\s+access-group|shutdown)\\b\'),\n re.compile(r\'^hostname\\s+\\S+$\'),\n re.compile(r\'^ip\\s+access-list\\s+extended\\s\'),\n re.compile(r\'^\\s+\\d+\\s+(permit|deny)\\s\'),\n re.compile(r\'^!\\s*\'), re.compile(r\'^\\s*$\'),\n]\n\ndef is_allowed(cmd: str) -> bool:\n return any(p.match(cmd) for p in _ALLOWED_CMD_RE)\n\n# ══════════════════════════════════════════════════════════════════════\n# CONNECTION POOL\n# ══════════════════════════════════════════════════════════════════════\n\nclass SwitchConnectionPool:\n """\n Keeps one SSH connection alive for up to CONN_POOL_MAX_S seconds.\n On liveness check failure or expiry → opens a fresh connection.\n Thread-safe.\n """\n def __init__(self):\n self._conn: Optional[paramiko.SSHClient] = None\n self._born: float = 0\n self._lock = threading.Lock()\n\n def get(self) -> paramiko.SSHClient:\n with self._lock:\n now = time.time()\n age = now - self._born\n if self._conn and age < CONN_POOL_MAX_S:\n try:\n transport = self._conn.get_transport()\n if transport and transport.is_active():\n transport.send_ignore() # lightweight liveness ping\n return self._conn\n except Exception:\n log.info("Pool: liveness check failed — opening fresh connection")\n self._close_unsafe()\n self._conn = _open_connection()\n self._born = time.time()\n log.info("Pool: new connection opened")\n return self._conn\n\n def invalidate(self):\n with self._lock:\n self._close_unsafe()\n\n def _close_unsafe(self):\n if self._conn:\n try:\n self._conn.close()\n except Exception:\n pass\n self._conn = None\n self._born = 0\n\n_pool = SwitchConnectionPool()\n\n\ndef _open_connection() -> paramiko.SSHClient:\n client = paramiko.SSHClient()\n known = Path(KNOWN_HOSTS)\n if known.exists():\n client.load_host_keys(str(known))\n client.set_missing_host_key_policy(paramiko.RejectPolicy())\n else:\n log.warning("No known_hosts — trust on first use")\n client.set_missing_host_key_policy(paramiko.AutoAddPolicy())\n try:\n client.connect(\n hostname=SWITCH_HOST, port=SWITCH_PORT, username=SWITCH_USER,\n key_filename=KEY_PATH, look_for_keys=False, allow_agent=False,\n timeout=10, banner_timeout=10,\n )\n except paramiko.AuthenticationException:\n raise HTTPException(503, "SSH auth failed — is the public key loaded on the switch?")\n except Exception as e:\n raise HTTPException(503, f"Cannot reach switch at {SWITCH_HOST}: {e}")\n if not known.exists():\n known.parent.mkdir(parents=True, exist_ok=True)\n client.save_host_keys(str(known))\n log.info("Host key pinned to known_hosts")\n return client\n\n\ndef read_cmd(cmd: str) -> str:\n """Run a read-only command via the pool. Invalidates pool on error."""\n try:\n conn = _pool.get()\n _, stdout, _ = conn.exec_command(cmd, timeout=10)\n return stdout.read().decode("utf-8", errors="replace")\n except HTTPException:\n raise\n except Exception as e:\n log.warning(f"read_cmd failed ({cmd}): {e} — invalidating pool")\n _pool.invalidate()\n raise HTTPException(503, f"Switch read failed: {e}")\n\n\n# ── Per-command interactive push ───────────────────────────────────────\n\n_SWITCH_ERR = re.compile(\n r\'%\\s*(invalid|error|unknown|bad|failed|cannot|not\\s+found|does\\s+not\\s+exist\'\n r\'|incomplete|ambiguous|out\\s+of\\s+range|already\\s+exists)\',\n re.I\n)\n\ndef _run_one(channel, cmd: str) -> tuple[str, bool]:\n channel.send(cmd + "\\n")\n time.sleep(0.35)\n out, deadline = "", time.time() + 6\n while time.time() < deadline:\n if channel.recv_ready():\n out += channel.recv(4096).decode("utf-8", errors="replace")\n time.sleep(0.1)\n else:\n if out:\n break\n time.sleep(0.1)\n return out.strip(), bool(_SWITCH_ERR.search(out))\n\n\ndef push_one_by_one(commands: list[str]) -> dict:\n """\n Opens a FRESH dedicated connection for push (not from pool —\n we don\'t want a push session to corrupt the pool\'s read connection).\n Runs commands one at a time, stops on first error.\n Saves config only on full success.\n """\n _pool.invalidate() # invalidate pool — switch will be busy during push\n client = _open_connection()\n results = []\n stopped_at = None\n try:\n ch = client.invoke_shell()\n ch.settimeout(10)\n time.sleep(0.5)\n if ch.recv_ready():\n ch.recv(4096) # drain banner\n\n for setup in ["enable", "configure terminal"]:\n out, err = _run_one(ch, setup)\n if err:\n return {"success": False, "saved": False,\n "error": f"Failed entering config mode: {out}",\n "results": [], "hint": "Check switch is reachable and credentials are correct"}\n\n for i, cmd in enumerate(commands):\n s = cmd.strip()\n if not s or s.startswith("!"):\n results.append({"index": i, "command": cmd,\n "output": "", "success": True, "skipped": True})\n continue\n log.info(f" [{i+1}/{len(commands)}] {s}")\n out, had_err = _run_one(ch, s)\n results.append({"index": i, "command": cmd,\n "output": out, "success": not had_err, "skipped": False})\n if had_err:\n stopped_at = i\n log.warning(f" Error at command {i+1}: {out}")\n break\n\n _run_one(ch, "end")\n saved = False\n if stopped_at is None:\n _, save_err = _run_one(ch, "copy running-config nvram:config.cfg")\n saved = not save_err\n if saved:\n log.info("Config saved to NVRAM")\n else:\n log.warning("Config save may have failed — verify at console")\n else:\n log.warning("Config NOT saved — push stopped on error")\n\n ch.close()\n return {\n "success": stopped_at is None,\n "saved": saved,\n "commands_total": len(commands),\n "commands_sent": len([r for r in results if not r.get("skipped")]),\n "stopped_at": stopped_at,\n "error": results[stopped_at]["output"] if stopped_at is not None else None,\n "hint": "Remaining commands must be run at the switch console" if stopped_at is not None else None,\n "results": results,\n }\n finally:\n client.close()\n # Pool will reopen on next poll naturally\n\n# ══════════════════════════════════════════════════════════════════════\n# VISITOR-AWARE POLLER\n# ══════════════════════════════════════════════════════════════════════\n\n_cache: dict = {\n "port_status": None, "poe_status": None,\n "vlan_members": None, "sys_info": None,\n "last_poll": 0, "poll_error": None,\n}\n_cache_lock = threading.Lock()\n\n# Visitor tracking\n_visitors: dict[str, float] = {} # { visitor_id: last_seen }\n_visitors_lock = threading.Lock()\n_poll_mode = "idle" # idle | active | background\n\ndef heartbeat(visitor_id: str, mode: str = "active"):\n """Called by frontend to indicate a visitor is present."""\n global _poll_mode\n with _visitors_lock:\n _visitors[visitor_id] = time.time()\n _poll_mode = mode\n\ndef prune_visitors():\n global _poll_mode\n with _visitors_lock:\n now = time.time()\n gone = [v for v, t in _visitors.items() if now - t > POLL_IDLE_AFTER]\n for v in gone:\n del _visitors[v]\n if not _visitors:\n _poll_mode = "idle"\n\ndef _poll_loop():\n log.info("Poller thread started")\n while True:\n prune_visitors()\n with _visitors_lock:\n mode = _poll_mode\n has_visitors = bool(_visitors)\n\n if not has_visitors:\n # No visitors — sleep and check again, don\'t poll switch\n time.sleep(30)\n continue\n\n # Poll the switch\n try:\n data = {\n "port_status": read_cmd("show interfaces"),\n "poe_status": read_cmd("show poe-port-status"),\n "vlan_members": read_cmd("show vlan members"),\n "sys_info": read_cmd("show sys-info"),\n }\n with _cache_lock:\n _cache.update(data)\n _cache["last_poll"] = time.time()\n _cache["poll_error"] = None\n except Exception as e:\n with _cache_lock:\n _cache["poll_error"] = str(e)\n log.warning(f"Poll error: {e}")\n\n interval = POLL_ACTIVE_S if mode == "active" else POLL_BG_S\n time.sleep(interval)\n\ndef start_poller():\n t = threading.Thread(target=_poll_loop, daemon=True)\n t.start()\n log.info("Poller started")\n\n# ══════════════════════════════════════════════════════════════════════\n# PYDANTIC MODELS\n# ══════════════════════════════════════════════════════════════════════\n\nclass TotpVerify(BaseModel):\n code: str\n\nclass SessionCheck(BaseModel):\n token: str\n\nclass SessionRevoke(BaseModel):\n token: str\n\nclass Heartbeat(BaseModel):\n visitor_id: str\n mode: str = "active" # active | background\n\nclass PushBatch(BaseModel):\n token: str\n commands: list[str]\n\nclass VlanCreate(BaseModel):\n token: str\n vlan_id: int\n name: str\n @field_validator("vlan_id")\n @classmethod\n def cv(cls, v): return san_vid(v)\n @field_validator("name")\n @classmethod\n def cn(cls, v): return _san(v, _RE_VNAME, "name")\n\nclass VlanDelete(BaseModel):\n token: str\n vlan_id: int\n @field_validator("vlan_id")\n @classmethod\n def cv(cls, v):\n vid = san_vid(v)\n if vid == 1: raise ValueError("Cannot delete VLAN 1")\n return vid\n\nclass AclRule(BaseModel):\n action: str\n proto: str\n src: Optional[str] = "any"\n src_mask: Optional[str] = "0.0.0.255"\n src_any: Optional[bool] = True\n dst: Optional[str] = "any"\n dst_mask: Optional[str] = "0.0.0.255"\n dst_any: Optional[bool] = True\n port: Optional[str] = ""\n @field_validator("action")\n @classmethod\n def ca(cls, v): return _san(v, _RE_ACTION, "action")\n @field_validator("proto")\n @classmethod\n def cp(cls, v): return _san(v, _RE_PROTO, "proto")\n\nclass AclCreate(BaseModel):\n token: str\n name: str\n apply_vlan: int\n direction: str\n rules: list[AclRule]\n @field_validator("name")\n @classmethod\n def cn(cls, v): return _san(v, _RE_ANAME, "name")\n @field_validator("direction")\n @classmethod\n def cd(cls, v): return _san(v, _RE_DIR, "direction")\n\nclass PortConfig(BaseModel):\n token: str\n port: int\n mode: str\n access_vlan: Optional[int] = 1\n tagged_vlans: Optional[list[int]] = []\n native_vlan: Optional[int] = 1\n poe: Optional[bool] = True\n poe_limit_mw: Optional[int] = 30000\n description: Optional[str] = ""\n @field_validator("port")\n @classmethod\n def cp(cls, v): return san_port(v)\n @field_validator("mode")\n @classmethod\n def cm(cls, v): return _san(v, _RE_MODE, "mode")\n @field_validator("poe_limit_mw")\n @classmethod\n def cpoe(cls, v):\n if v is not None and not 1000 <= v <= 30000:\n raise ValueError("PoE limit: 1000–30000 mW")\n return v\n\n# ══════════════════════════════════════════════════════════════════════\n# COMMAND BUILDERS\n# ══════════════════════════════════════════════════════════════════════\n\ndef build_port(cfg: PortConfig) -> list[str]:\n p = cfg.port\n iface = f"FastEthernet {p}" if p <= 48 else f"GigabitEthernet {p}"\n cmds = []\n if cfg.description:\n cmds += [f"interface {iface}", f\' name "{cfg.description}"\']\n if cfg.mode == "disabled":\n cmds += [f"interface {iface}", " shutdown"]\n elif cfg.mode == "access":\n vid = san_vid(cfg.access_vlan, "access_vlan")\n cmds += [f"vlan members add {vid} {p}", f"vlan pvid {p} {vid}"]\n elif cfg.mode == "trunk":\n native = san_vid(cfg.native_vlan, "native_vlan")\n tagged = [san_vid(v, f"tagged_{v}") for v in (cfg.tagged_vlans or [])]\n if tagged:\n ts = ",".join(str(v) for v in tagged)\n cmds += [f"vlan members add {ts} {p}", f"vlan tagging {ts} {p}"]\n cmds.append(f"vlan pvid {p} {native}")\n if p <= 48:\n cmds += [f"interface {iface}",\n " poe enable" if cfg.poe else " no poe enable"]\n if cfg.poe:\n cmds.append(f" poe poe-limit {cfg.poe_limit_mw}")\n return cmds\n\ndef build_acl(acl: AclCreate) -> list[str]:\n cmds = [f"ip access-list extended {acl.name}"]\n for i, r in enumerate(acl.rules):\n src = "any" if r.src_any else f"{r.src} {r.src_mask}"\n dst = "any" if r.dst_any else f"{r.dst} {r.dst_mask}"\n port_str = f" eq {r.port}" if r.port else ""\n cmds.append(f" {i+1} {r.action} {r.proto} {src} {dst}{port_str}")\n vid = san_vid(acl.apply_vlan, "apply_vlan")\n cmds += [f"interface vlan {vid}",\n f" ip access-group {acl.name} {acl.direction}"]\n return cmds\n\n# ══════════════════════════════════════════════════════════════════════\n# APP\n# ══════════════════════════════════════════════════════════════════════\n\napp = FastAPI(title="ERS 5952 Switch Manager", version="3.0.0",\n docs_url="/api/docs", redoc_url=None)\n\napp.add_middleware(CORSMiddleware, allow_origins=ALLOWED_ORIGINS,\n allow_methods=["GET","POST","DELETE"],\n allow_headers=["Content-Type"])\n\n@app.exception_handler(ValueError)\nasync def val_err(req: Request, exc: ValueError):\n return JSONResponse(400, {"detail": str(exc)})\n\n@app.on_event("startup")\ndef startup():\n global TOTP_SECRET\n TOTP_SECRET = get_or_create_totp_secret()\n log.info("TOTP secret loaded")\n start_poller()\n\n# ── Heartbeat (visitor tracking) ──────────────────────────────────────\n\n@app.post("/api/heartbeat")\ndef hb(body: Heartbeat):\n """\n Frontend calls this every 30s to indicate visitor is present.\n mode = \'active\' when tab is visible, \'background\' when hidden.\n Poller adjusts interval accordingly.\n """\n heartbeat(body.visitor_id, body.mode)\n with _cache_lock:\n last = _cache["last_poll"]\n err = _cache["poll_error"]\n return {\n "poll_mode": _poll_mode,\n "last_poll": last,\n "poll_age": round(time.time() - last, 1) if last else None,\n "poll_error": err,\n }\n\n# ── Auth ───────────────────────────────────────────────────────────────\n\n@app.post("/api/auth/verify")\ndef verify_totp(body: TotpVerify):\n """Verify TOTP code — returns session token for multiple pushes."""\n if not pyotp.TOTP(TOTP_SECRET).verify(body.code.strip(), valid_window=1):\n log.warning("TOTP verify failed")\n raise HTTPException(401, "Invalid TOTP code")\n token = create_session()\n log.info("TOTP OK — session created")\n return {\n "token": token,\n "expires_in": SESSION_TTL_S,\n "message": "Session active — re-authenticates on expiry or manual lock"\n }\n\n@app.post("/api/auth/check")\ndef check_session(body: SessionCheck):\n """Check if a session is still valid. Returns remaining seconds."""\n remaining = session_remaining(body.token)\n if remaining is None:\n raise HTTPException(401, "Session expired")\n return {"valid": True, "remaining": remaining}\n\n@app.post("/api/auth/revoke")\ndef revoke(body: SessionRevoke):\n """Manually lock — invalidates the session immediately."""\n revoke_session(body.token)\n log.info("Session manually revoked")\n return {"revoked": True}\n\n# ── Read endpoints (no auth) ───────────────────────────────────────────\n\n@app.get("/api/status")\ndef status():\n with _cache_lock:\n return {\n "backend": "online",\n "switch_host": SWITCH_HOST,\n "key_exists": Path(KEY_PATH).exists(),\n "known_hosts_pinned": Path(KNOWN_HOSTS).exists(),\n "last_poll": _cache["last_poll"],\n "poll_age": round(time.time() - _cache["last_poll"], 1)\n if _cache["last_poll"] else None,\n "poll_error": _cache["poll_error"],\n "poll_mode": _poll_mode,\n }\n\n@app.get("/api/live")\ndef live():\n """Cached live switch data — updated by background poller."""\n with _cache_lock:\n age = time.time() - _cache["last_poll"] if _cache["last_poll"] else None\n return {\n **_cache,\n "stale": age is None or age > max(POLL_ACTIVE_S, POLL_BG_S) * 3\n }\n\n@app.get("/api/switch/config")\ndef running_config():\n out = read_cmd("show running-config")\n return {"config": out, "lines": len(out.splitlines())}\n\n# ── Danger pre-flight (no auth — check before prompting TOTP) ─────────\n\n@app.post("/api/check/danger")\ndef danger_check(body: dict):\n cmds = body.get("commands", [])\n result = check_danger(cmds)\n rejected = [c for c in cmds if not is_allowed(c)]\n result["rejected_by_allowlist"] = rejected\n result["safe_to_push"] = (\n not result["has_hard_block"] and not rejected\n )\n return result\n\n# ── Push endpoints (require session token) ────────────────────────────\n\n@app.post("/api/switch/push")\ndef push(body: PushBatch):\n """\n Push CLI batch one command at a time.\n Requires valid session token.\n Hard-blocked commands are refused — no override.\n Warn-level commands proceed (user was already shown the warning).\n Stops on first switch error. Config saved only on full success.\n """\n require_session(body.token)\n\n danger = check_danger(body.commands)\n if danger["has_hard_block"]:\n raise HTTPException(400, {\n "message": "Hard-blocked commands detected — these must be run at the switch console",\n "blocked": danger["hard_blocked"],\n })\n\n rejected = [c for c in body.commands if not is_allowed(c)]\n if rejected:\n raise HTTPException(400, {\n "message": "Commands failed allowlist validation",\n "rejected": rejected[:10],\n })\n\n log.info(f"Push: {len(body.commands)} commands")\n return push_one_by_one(body.commands)\n\n@app.post("/api/switch/vlan")\ndef create_vlan(body: VlanCreate):\n require_session(body.token)\n return push_one_by_one(\n [f\'vlan create {body.vlan_id} name "{body.name}" type port\'])\n\n@app.delete("/api/switch/vlan/{vlan_id}")\ndef delete_vlan(vlan_id: int, token: str):\n require_session(token)\n return push_one_by_one([f"no vlan {san_vid(vlan_id)}"])\n\n@app.post("/api/switch/port")\ndef configure_port(body: PortConfig):\n require_session(body.token)\n return push_one_by_one(build_port(body))\n\n@app.post("/api/switch/acl")\ndef create_acl(body: AclCreate):\n require_session(body.token)\n return push_one_by_one(build_acl(body))\n\n# ── Serve React app ────────────────────────────────────────────────────\nif os.path.isdir(STATIC_DIR):\n app.mount("/", StaticFiles(directory=STATIC_DIR, html=True), name="frontend")\n\n# ══════════════════════════════════════════════════════════════════════\nif __name__ == "__main__":\n if "--setup-totp" in sys.argv:\n setup_totp()\n sys.exit(0)\n import uvicorn\n uvicorn.run("switch_backend:app", host="0.0.0.0", port=8765,\n log_level="info")\n\n# ══════════════════════════════════════════════════════════════════════\n# DEVICE ACCESS — DHCP LEASES, MAC RESERVATIONS, ACL PINHOLES\n# ══════════════════════════════════════════════════════════════════════\n\nimport json as _json\nfrom pathlib import Path as _Path\n\nDEVICES_FILE = _Path("/etc/switch-manager/devices.json")\n\ndef _load_devices() -> list:\n if DEVICES_FILE.exists():\n try: return _json.loads(DEVICES_FILE.read_text())\n except: pass\n return []\n\ndef _save_devices(devices: list):\n DEVICES_FILE.write_text(_json.dumps(devices, indent=2))\n\ndef _parse_dhcp_leases(raw: str) -> list:\n """Parse ERS 5952 \'show dhcp-server leases\' output."""\n import re\n leases = []\n for line in raw.splitlines():\n # Format: IP MAC State Remaining Hostname\n m = re.match(\n r\'\\s*(\\d+\\.\\d+\\.\\d+\\.\\d+)\\s+\'\n r\'([0-9a-fA-F]{2}[:\\-][0-9a-fA-F]{2}[:\\-][0-9a-fA-F]{2}\'\n r\'[:\\-][0-9a-fA-F]{2}[:\\-][0-9a-fA-F]{2}[:\\-][0-9a-fA-F]{2})\\s+\'\n r\'(\\S+)\\s+(\\S+)\\s*(.*)\', line)\n if m:\n leases.append({\n "ip": m.group(1),\n "mac": m.group(2).lower().replace(\'-\',\':\'),\n "state": m.group(3),\n "remaining": m.group(4),\n "hostname": m.group(5).strip() or "unknown",\n })\n return leases\n\ndef _parse_arp_table(raw: str) -> list:\n """Parse \'show arp\' for additional device discovery."""\n import re\n entries = []\n for line in raw.splitlines():\n m = re.match(\n r\'\\s*(\\d+\\.\\d+\\.\\d+\\.\\d+)\\s+\'\n r\'([0-9a-fA-F]{2}[:\\-][0-9a-fA-F]{2}[:\\-][0-9a-fA-F]{2}\'\n r\'[:\\-][0-9a-fA-F]{2}[:\\-][0-9a-fA-F]{2}[:\\-][0-9a-fA-F]{2})\',\n line)\n if m:\n entries.append({\n "ip": m.group(1),\n "mac": m.group(2).lower().replace(\'-\',\':\'),\n "state": "arp",\n "remaining": "-",\n "hostname": "",\n })\n return entries\n\nclass DeviceEntry(BaseModel):\n name: str\n mac: str\n ip: str\n vlan: int = 10\n management_access: bool = False\n static_ip: bool = False\n notes: Optional[str] = ""\n\nclass DeviceUpdate(BaseModel):\n token: str\n device: DeviceEntry\n\nclass DeviceDelete(BaseModel):\n token: str\n mac: str\n\nclass PinholeRequest(BaseModel):\n token: str\n mac: str\n allow: bool = True\n\ndef _build_dhcp_reservation_cmds(device: DeviceEntry) -> list:\n """Generate ERS 5952 CLI for DHCP reservation (static binding)."""\n mac_clean = device.mac.replace(\':\',\'-\').upper()\n return [\n f"ip dhcp-server static-binding {device.ip}",\n f" mac-address {mac_clean}",\n f" client-name \\"{device.name}\\"",\n ]\n\ndef _build_pinhole_acl_cmds(device: DeviceEntry, mgmt_ip: str, allow: bool) -> list:\n """Generate ACL commands to allow/deny a device IP to reach management."""\n acl_name = f"MGMT-ACCESS"\n if allow:\n return [\n f"ip access-list extended {acl_name}",\n f" permit tcp host {device.ip} host {mgmt_ip} eq 443",\n f" permit tcp host {device.ip} host {mgmt_ip} eq 8765",\n ]\n else:\n return [\n f"ip access-list extended {acl_name}",\n f" no permit tcp host {device.ip} host {mgmt_ip}",\n ]\n\n@app.get("/api/devices")\ndef get_devices():\n """Return saved device list plus live DHCP leases from switch."""\n saved = _load_devices()\n live_leases = []\n try:\n dhcp_raw = read_cmd("show dhcp-server leases")\n arp_raw = read_cmd("show arp")\n live_leases = _parse_dhcp_leases(dhcp_raw)\n # Merge ARP entries not already in leases\n arp = _parse_arp_table(arp_raw)\n lease_ips = {l["ip"] for l in live_leases}\n for entry in arp:\n if entry["ip"] not in lease_ips:\n live_leases.append(entry)\n except Exception as e:\n log.warning(f"Could not pull DHCP/ARP from switch: {e}")\n return {\n "saved": saved,\n "live": live_leases,\n "mgmt_ip": SWITCH_HOST,\n }\n\n@app.post("/api/devices/save")\ndef save_device(body: DeviceUpdate):\n require_session(body.token)\n devices = _load_devices()\n existing = next((i for i, d in enumerate(devices) if d["mac"] == body.device.mac), None)\n device_dict = body.device.dict()\n if existing is not None:\n devices[existing] = device_dict\n else:\n devices.append(device_dict)\n _save_devices(devices)\n log.info(f"Device saved: {body.device.name} ({body.device.mac})")\n return {"success": True, "devices": devices}\n\n@app.post("/api/devices/delete")\ndef delete_device(body: DeviceDelete):\n require_session(body.token)\n devices = [d for d in _load_devices() if d["mac"] != body.mac]\n _save_devices(devices)\n return {"success": True}\n\n@app.post("/api/devices/push-reservation")\ndef push_reservation(body: DeviceUpdate):\n """Push a DHCP static binding for this device to the switch."""\n require_session(body.token)\n cmds = _build_dhcp_reservation_cmds(body.device)\n danger = check_danger(cmds)\n if danger["has_hard_block"]:\n raise HTTPException(400, {"message": "Blocked", "blocked": danger["hard_blocked"]})\n log.info(f"Pushing DHCP reservation for {body.device.name}")\n return push_one_by_one(cmds)\n\n@app.post("/api/devices/push-pinhole")\ndef push_pinhole(body: PinholeRequest):\n """Add or remove an ACL pinhole for a device to reach management."""\n require_session(body.token)\n devices = _load_devices()\n device = next((DeviceEntry(**d) for d in devices if d["mac"] == body.mac), None)\n if not device:\n raise HTTPException(404, "Device not found — save it first")\n import socket\n try:\n mgmt_ip = socket.gethostbyname(socket.gethostname())\n except Exception:\n mgmt_ip = SWITCH_HOST.rsplit(\'.\',1)[0] + \'.50\'\n cmds = _build_pinhole_acl_cmds(device, mgmt_ip, body.allow)\n log.info(f"Pinhole {\'allow\' if body.allow else \'deny\'} for {device.name} ({device.ip})")\n return push_one_by_one(cmds)\n\n# ══════════════════════════════════════════════════════════════════════\n# WIREGUARD PEER MANAGEMENT\n# ══════════════════════════════════════════════════════════════════════\n\nWG_CLIENT_DIR_API = _Path("/etc/switch-manager/clients")\nWG_CONF_PATH = _Path("/etc/wireguard/wg0.conf")\n\nclass WGClientRequest(BaseModel):\n token: str\n name: str\n\nclass WGRevokeRequest(BaseModel):\n token: str\n name: str\n\ndef _wg_genkey_api():\n import subprocess as _sp\n priv = _sp.run(["wg","genkey"], capture_output=True, text=True).stdout.strip()\n pub = _sp.run(["wg","pubkey"], input=priv, capture_output=True, text=True).stdout.strip()\n return priv, pub\n\ndef _wg_status() -> dict:\n import subprocess as _sp\n try:\n raw = _sp.run(["wg","show"], capture_output=True, text=True).stdout\n peers = []\n current = {}\n for line in raw.splitlines():\n line = line.strip()\n if line.startswith("peer:"):\n if current: peers.append(current)\n current = {"public_key": line.split(":",1)[1].strip()}\n elif line.startswith("endpoint:"):\n current["endpoint"] = line.split(":",1)[1].strip()\n elif line.startswith("latest handshake:"):\n current["last_handshake"] = line.split(":",1)[1].strip()\n elif line.startswith("transfer:"):\n current["transfer"] = line.split(":",1)[1].strip()\n elif line.startswith("allowed ips:"):\n current["allowed_ips"] = line.split(":",1)[1].strip()\n if current: peers.append(current)\n\n # Match peers to named client files\n named = {}\n if WG_CLIENT_DIR_API.exists():\n for f in WG_CLIENT_DIR_API.glob("*.conf"):\n txt = f.read_text()\n import re\n m = re.search(r\'PublicKey\\s*=\\s*(\\S+)\', txt)\n if m: named[m.group(1)] = f.stem\n\n for p in peers:\n p["name"] = named.get(p.get("public_key",""), "unknown")\n\n return {"running": True, "peers": peers}\n except Exception as e:\n return {"running": False, "error": str(e), "peers": []}\n\n@app.get("/api/wireguard/status")\ndef wg_status():\n return _wg_status()\n\n@app.get("/api/wireguard/clients")\ndef wg_clients():\n clients = []\n if WG_CLIENT_DIR_API.exists():\n for f in sorted(WG_CLIENT_DIR_API.glob("*.conf")):\n clients.append({"name": f.stem, "file": str(f)})\n return {"clients": clients}\n\n@app.post("/api/wireguard/add-client")\ndef wg_add_client(body: WGClientRequest):\n require_session(body.token)\n if not WG_CONF_PATH.exists():\n raise HTTPException(503, "WireGuard not configured on this machine")\n\n import re, subprocess as _sp, socket\n\n # Next available IP\n conf_text = WG_CONF_PATH.read_text()\n used = set()\n for m in re.finditer(r\'AllowedIPs\\s*=\\s*(\\S+)\', conf_text):\n used.add(m.group(1).split(\'/\')[0])\n\n subnet = "10.99.0"\n num = 2\n while f"{subnet}.{num}" in used and num < 254: num += 1\n client_ip = f"{subnet}.{num}"\n\n # Server public key\n server_pub_path = _Path("/etc/switch-manager/wg_server_public")\n if not server_pub_path.exists():\n raise HTTPException(503, "Server public key not found")\n server_pub = server_pub_path.read_text().strip()\n\n # Client keys\n c_priv, c_pub = _wg_genkey_api()\n\n # Peer entry in server config\n peer = f"\\n[Peer]\\n# {body.name}\\nPublicKey = {c_pub}\\nAllowedIPs = {client_ip}/32\\n"\n with open(WG_CONF_PATH, \'a\') as f:\n f.write(peer)\n\n # Reload live\n _sp.run(["wg","addconf","wg0","/dev/stdin"],\n input=f"[Peer]\\nPublicKey = {c_pub}\\nAllowedIPs = {client_ip}/32\\n",\n capture_output=True, text=True)\n\n # Public IP for endpoint\n try:\n pub_ip = _sp.run(["curl","-s","--max-time","5","https://api.ipify.org"],\n capture_output=True, text=True).stdout.strip()\n except Exception:\n pub_ip = socket.gethostbyname(socket.gethostname())\n\n # Get management IP from switch config\n mgmt_subnet = \'.\'.join(SWITCH_HOST.split(\'.\')[:3]) + \'.0/24\'\n\n client_conf = (\n f"[Interface]\\nPrivateKey = {c_priv}\\nAddress = {client_ip}/24\\n"\n f"DNS = {subnet}.1\\n\\n"\n f"[Peer]\\nPublicKey = {server_pub}\\n"\n f"Endpoint = {pub_ip}:51820\\n"\n f"AllowedIPs = {mgmt_subnet}, {subnet}.0/24\\n"\n f"PersistentKeepalive = 25\\n"\n )\n\n WG_CLIENT_DIR_API.mkdir(exist_ok=True)\n client_file = WG_CLIENT_DIR_API / f"{body.name}.conf"\n client_file.write_text(client_conf)\n client_file.chmod(0o600)\n\n log.info(f"WireGuard client added: {body.name} → {client_ip}")\n return {\n "success": True,\n "name": body.name,\n "client_ip": client_ip,\n "config": client_conf,\n "file": str(client_file),\n }\n\n@app.post("/api/wireguard/revoke-client")\ndef wg_revoke_client(body: WGRevokeRequest):\n require_session(body.token)\n import re, subprocess as _sp\n\n client_file = WG_CLIENT_DIR_API / f"{body.name}.conf"\n if not client_file.exists():\n raise HTTPException(404, f"Client \'{body.name}\' not found")\n\n # Get client public key\n txt = client_file.read_text()\n m = re.search(r\'\\[Peer\\].*?PublicKey\\s*=\\s*(\\S+)\', txt, re.DOTALL)\n client_pub = m.group(1) if m else None\n\n # Remove from server config\n if WG_CONF_PATH.exists():\n conf = WG_CONF_PATH.read_text()\n # Remove the [Peer] block for this client\n cleaned = re.sub(\n rf\'\\n\\[Peer\\]\\n# {re.escape(body.name)}\\n.*?(?=\\n\\[Peer\\]|\\Z)\',\n \'\', conf, flags=re.DOTALL\n )\n WG_CONF_PATH.write_text(cleaned)\n\n # Remove live peer\n if client_pub:\n _sp.run(["wg","set","wg0","peer",client_pub,"remove"],\n capture_output=True, text=True)\n\n # Delete client file\n client_file.unlink()\n log.info(f"WireGuard client revoked: {body.name}")\n return {"success": True}\n\n@app.get("/api/wireguard/client-qr/{name}")\ndef wg_client_qr(name: str):\n """Return client config as text for QR generation in frontend."""\n client_file = WG_CLIENT_DIR_API / f"{name}.conf"\n if not client_file.exists():\n raise HTTPException(404, f"Client \'{name}\' not found")\n return {"name": name, "config": client_file.read_text()}\n\n# ══════════════════════════════════════════════════════════════════════\n# DHCP MANAGEMENT — SWITCH + OPNSENSE UNIFIED VIEW\n# ══════════════════════════════════════════════════════════════════════\n\nimport urllib.request as _urlreq\nimport urllib.error as _urlerr\nimport ssl as _ssl\nimport base64 as _b64\n\nOPNSENSE_FILE = _Path("/etc/switch-manager/opnsense.json")\n\ndef _load_opnsense_cfg() -> dict:\n if OPNSENSE_FILE.exists():\n try: return _json.loads(OPNSENSE_FILE.read_text())\n except: pass\n return {}\n\ndef _save_opnsense_cfg(cfg: dict):\n OPNSENSE_FILE.write_text(_json.dumps(cfg, indent=2))\n OPNSENSE_FILE.chmod(0o600)\n\ndef _opnsense_request(cfg: dict, path: str, method="GET", body=None) -> dict:\n """Make an authenticated request to the OPNsense API."""\n host = cfg.get("host","")\n key = cfg.get("key","")\n secret = cfg.get("secret","")\n if not host or not key or not secret:\n raise ValueError("OPNsense not configured")\n url = f"https://{host}/api/{path}"\n creds = _b64.b64encode(f"{key}:{secret}".encode()).decode()\n ctx = _ssl.create_default_context()\n ctx.check_hostname = False\n ctx.verify_mode = _ssl.CERT_NONE\n headers = {\n "Authorization": f"Basic {creds}",\n "Content-Type": "application/json",\n }\n data = _json.dumps(body).encode() if body else None\n req = _urlreq.Request(url, data=data, headers=headers, method=method)\n try:\n with _urlreq.urlopen(req, timeout=5, context=ctx) as r:\n return _json.loads(r.read().decode())\n except _urlerr.HTTPError as e:\n raise ValueError(f"OPNsense API error {e.code}: {e.reason}")\n except Exception as e:\n raise ValueError(f"OPNsense unreachable: {e}")\n\ndef _detect_opnsense_host(gateway_ip: str) -> str | None:\n """Try to reach OPNsense API at the gateway IP."""\n try:\n ctx = _ssl.create_default_context()\n ctx.check_hostname = False\n ctx.verify_mode = _ssl.CERT_NONE\n url = f"https://{gateway_ip}/api/core/firmware/status"\n req = _urlreq.Request(url, headers={"User-Agent":"switch-manager/1"})\n _urlreq.urlopen(req, timeout=3, context=ctx)\n return gateway_ip\n except Exception:\n return None\n\ndef _get_opnsense_reservations(cfg: dict) -> list:\n """Fetch DHCP static mappings from OPNsense."""\n try:\n data = _opnsense_request(cfg, "dhcpv4/leases/searchReservation")\n rows = data.get("rows", [])\n return [\n {\n "ip": r.get("ipaddr",""),\n "mac": r.get("mac","").lower(),\n "hostname": r.get("hostname",""),\n "descr": r.get("descr",""),\n "if": r.get("if",""),\n "source": "opnsense",\n "uuid": r.get("uuid",""),\n }\n for r in rows if r.get("mac")\n ]\n except Exception as e:\n log.warning(f"OPNsense reservations fetch failed: {e}")\n return []\n\ndef _get_opnsense_leases(cfg: dict) -> list:\n """Fetch active DHCP leases from OPNsense."""\n try:\n data = _opnsense_request(cfg, "dhcpv4/leases/searchLease")\n rows = data.get("rows", [])\n return [\n {\n "ip": r.get("address",""),\n "mac": r.get("mac","").lower(),\n "hostname": r.get("hostname",""),\n "state": r.get("state",""),\n "if": r.get("if",""),\n "source": "opnsense_lease",\n }\n for r in rows if r.get("mac")\n ]\n except Exception as e:\n log.warning(f"OPNsense leases fetch failed: {e}")\n return []\n\ndef _get_switch_reservations() -> list:\n """Fetch DHCP static bindings from ERS 5952."""\n import re as _re\n try:\n raw = read_cmd("show dhcp-server static-binding")\n bindings = []\n current = {}\n for line in raw.splitlines():\n m = _re.match(r\'\\s*IP Address:\\s*(\\S+)\', line)\n if m:\n if current: bindings.append(current)\n current = {"ip": m.group(1), "mac":"", "hostname":"", "source":"switch"}\n m2 = _re.match(r\'\\s*MAC Address:\\s*(\\S+)\', line)\n if m2 and current:\n current["mac"] = m2.group(1).lower().replace(\'-\',\':\')\n m3 = _re.match(r\'\\s*Client Name:\\s*(\\S+)\', line)\n if m3 and current:\n current["hostname"] = m3.group(1)\n if current and current.get("ip"):\n bindings.append(current)\n return bindings\n except Exception as e:\n log.warning(f"Switch DHCP reservation fetch failed: {e}")\n return []\n\ndef _get_switch_dhcp_status() -> dict:\n """Check if switch DHCP server is running and which VLANs it serves."""\n import re as _re\n try:\n raw = read_cmd("show dhcp-server")\n running = "enabled" in raw.lower() or "active" in raw.lower()\n vlans = _re.findall(r\'VLAN\\s+(\\d+)\', raw, _re.I)\n return {"running": running, "vlans": list(set(vlans))}\n except Exception:\n return {"running": False, "vlans": []}\n\ndef _find_conflicts(switch_res: list, opnsense_res: list) -> list:\n """\n Find same MAC in both switch and OPNsense.\n Flag if IPs differ (conflict) or same (duplicate — harmless but messy).\n """\n switch_by_mac = {r["mac"]: r for r in switch_res if r.get("mac")}\n conflicts = []\n for r in opnsense_res:\n mac = r.get("mac","")\n if mac and mac in switch_by_mac:\n sw = switch_by_mac[mac]\n conflicts.append({\n "mac": mac,\n "hostname": r.get("hostname") or sw.get("hostname",""),\n "switch_ip": sw["ip"],\n "opnsense_ip": r["ip"],\n "ip_conflict": sw["ip"] != r["ip"],\n "opnsense_uuid": r.get("uuid",""),\n })\n return conflicts\n\n# ── OPNsense config models ─────────────────────────────────────────────────\n\nclass OPNsenseConfig(BaseModel):\n host: str\n key: str\n secret: str\n\nclass OPNsenseReservationPush(BaseModel):\n token: str\n mac: str\n ip: str\n hostname: str\n descr: Optional[str] = ""\n iface: Optional[str] = "lan"\n\nclass SyncRequest(BaseModel):\n token: str\n mac: str\n direction: str # "to_switch" | "to_opnsense" | "remove_switch" | "remove_opnsense"\n\n# ── DHCP endpoints ─────────────────────────────────────────────────────────\n\n@app.get("/api/dhcp/overview")\ndef dhcp_overview():\n """\n Unified DHCP view:\n - Switch static bindings + active leases\n - OPNsense reservations + leases (if configured)\n - Conflicts (same MAC, different IP)\n - Which DHCP server is active per VLAN\n """\n switch_res = _get_switch_reservations()\n switch_leases = []\n switch_status = _get_switch_dhcp_status()\n\n # Also pull ARP for discovery\n try:\n arp_raw = read_cmd("show arp")\n dhcp_raw = read_cmd("show dhcp-server leases")\n switch_leases = _parse_dhcp_leases(dhcp_raw) + _parse_arp_table(arp_raw)\n # Deduplicate by IP\n seen_ips = set()\n unique_leases = []\n for l in switch_leases:\n if l["ip"] not in seen_ips:\n seen_ips.add(l["ip"])\n unique_leases.append(l)\n switch_leases = unique_leases\n except Exception as e:\n log.warning(f"Switch lease fetch failed: {e}")\n\n cfg = _load_opnsense_cfg()\n opnsense_res = _get_opnsense_reservations(cfg) if cfg.get("key") else []\n opnsense_leases = _get_opnsense_leases(cfg) if cfg.get("key") else []\n conflicts = _find_conflicts(switch_res, opnsense_res)\n\n # Which VLANs have switch DHCP vs OPNsense\n # Switch DHCP VLANs from status, OPNsense VLANs inferred from interface names\n opnsense_ifaces = list({r.get("if","") for r in opnsense_res + opnsense_leases if r.get("if")})\n\n return {\n "switch": {\n "status": switch_status,\n "reservations": switch_res,\n "leases": switch_leases,\n },\n "opnsense": {\n "configured": bool(cfg.get("key")),\n "host": cfg.get("host",""),\n "reservations": opnsense_res,\n "leases": opnsense_leases,\n "interfaces": opnsense_ifaces,\n },\n "conflicts": conflicts,\n "has_conflicts": len(conflicts) > 0,\n }\n\n@app.get("/api/dhcp/detect-opnsense")\ndef detect_opnsense_endpoint():\n """Auto-detect OPNsense at the gateway IP."""\n import re as _re\n try:\n route = read_cmd("show ip route default")\n m = _re.search(r\'(\\d+\\.\\d+\\.\\d+\\.\\d+)\', route)\n gateway = m.group(1) if m else None\n except Exception:\n gateway = None\n\n if not gateway:\n # Try from switch management IP\n parts = SWITCH_HOST.split(\'.\')\n parts[-1] = \'1\'\n gateway = \'.\'.join(parts)\n\n host = _detect_opnsense_host(gateway)\n return {\n "detected": host is not None,\n "host": host,\n "gateway": gateway,\n "configured": bool(_load_opnsense_cfg().get("key")),\n }\n\n@app.post("/api/dhcp/configure-opnsense")\ndef configure_opnsense(body: OPNsenseConfig):\n """Save OPNsense API credentials. Tests connectivity first."""\n cfg = {"host": body.host, "key": body.key, "secret": body.secret}\n try:\n result = _opnsense_request(cfg, "core/firmware/status")\n _save_opnsense_cfg(cfg)\n log.info(f"OPNsense configured: {body.host}")\n return {"success": True, "version": result.get("product_version","unknown")}\n except ValueError as e:\n raise HTTPException(400, str(e))\n\n@app.delete("/api/dhcp/configure-opnsense")\ndef remove_opnsense_config():\n """Remove OPNsense API credentials."""\n if OPNSENSE_FILE.exists():\n OPNSENSE_FILE.unlink()\n return {"success": True}\n\n@app.post("/api/dhcp/push-to-opnsense")\ndef push_reservation_to_opnsense(body: OPNsenseReservationPush):\n """Create a DHCP static mapping in OPNsense."""\n require_session(body.token)\n cfg = _load_opnsense_cfg()\n if not cfg.get("key"):\n raise HTTPException(503, "OPNsense not configured")\n try:\n result = _opnsense_request(cfg, "dhcpv4/reservations/addReservation", "POST", {\n "reservation": {\n "interface": body.iface,\n "mac": body.mac,\n "ipaddr": body.ip,\n "hostname": body.hostname,\n "descr": body.descr or f"Added by switch-manager",\n }\n })\n # Apply changes\n _opnsense_request(cfg, "dhcpv4/service/reconfigure", "POST")\n log.info(f"OPNsense reservation pushed: {body.mac} → {body.ip}")\n return {"success": True, "result": result}\n except ValueError as e:\n raise HTTPException(400, str(e))\n\n@app.post("/api/dhcp/sync")\ndef sync_reservation(body: SyncRequest):\n """\n Sync a reservation between switch and OPNsense.\n Directions:\n to_switch — copy OPNsense reservation to switch\n to_opnsense — copy switch reservation to OPNsense\n remove_switch — remove from switch only\n remove_opnsense — remove from OPNsense only\n """\n require_session(body.token)\n cfg = _load_opnsense_cfg()\n overview = dhcp_overview()\n\n # Find the device in both sources\n sw_res = next((r for r in overview["switch"]["reservations"] if r["mac"]==body.mac), None)\n ops_res = next((r for r in overview["opnsense"]["reservations"] if r["mac"]==body.mac), None)\n\n if body.direction == "to_switch":\n if not ops_res:\n raise HTTPException(404, "OPNsense reservation not found")\n cmds = _build_dhcp_reservation_cmds(type(\'D\',(),{\n "ip": ops_res["ip"], "mac": ops_res["mac"], "name": ops_res.get("hostname","")\n })())\n return push_one_by_one(cmds)\n\n elif body.direction == "to_opnsense":\n if not sw_res:\n raise HTTPException(404, "Switch reservation not found")\n if not cfg.get("key"):\n raise HTTPException(503, "OPNsense not configured")\n result = _opnsense_request(cfg, "dhcpv4/reservations/addReservation", "POST", {\n "reservation": {\n "mac": sw_res["mac"], "ipaddr": sw_res["ip"],\n "hostname": sw_res.get("hostname",""), "descr": "Synced from switch",\n "interface": "lan",\n }\n })\n _opnsense_request(cfg, "dhcpv4/service/reconfigure", "POST")\n return {"success": True, "result": result}\n\n elif body.direction == "remove_switch":\n if not sw_res:\n raise HTTPException(404, "Switch reservation not found")\n return push_one_by_one([f"no ip dhcp-server static-binding {sw_res[\'ip\']}"])\n\n elif body.direction == "remove_opnsense":\n if not ops_res or not ops_res.get("uuid"):\n raise HTTPException(404, "OPNsense reservation not found or missing UUID")\n if not cfg.get("key"):\n raise HTTPException(503, "OPNsense not configured")\n _opnsense_request(cfg, f"dhcpv4/reservations/delReservation/{ops_res[\'uuid\']}", "DELETE")\n _opnsense_request(cfg, "dhcpv4/service/reconfigure", "POST")\n return {"success": True}\n\n raise HTTPException(400, f"Unknown direction: {body.direction}")\n\n# ══════════════════════════════════════════════════════════════════════\n# CONTROL D / ctrld DNS MANAGEMENT\n# ══════════════════════════════════════════════════════════════════════\n\nCTRLD_FILE = _Path("/etc/switch-manager/ctrld.json")\nCTRLD_BIN = _Path("/usr/local/bin/ctrld")\n\ndef _load_ctrld_cfg() -> dict:\n if CTRLD_FILE.exists():\n try: return _json.loads(CTRLD_FILE.read_text())\n except: pass\n return {}\n\ndef _save_ctrld_cfg(cfg: dict):\n CTRLD_FILE.write_text(_json.dumps(cfg, indent=2))\n CTRLD_FILE.chmod(0o600)\n\ndef _ctrld_status() -> dict:\n """Check if ctrld is running and which mode."""\n import subprocess as _sp\n try:\n r = _sp.run([str(CTRLD_BIN), "status"],\n capture_output=True, text=True, timeout=5)\n running = r.returncode == 0\n return {\n "installed": CTRLD_BIN.exists(),\n "running": running,\n "output": r.stdout.strip() or r.stderr.strip(),\n }\n except Exception as e:\n return {\n "installed": CTRLD_BIN.exists(),\n "running": False,\n "output": str(e),\n }\n\ndef _ctrld_config_path() -> _Path:\n """Find ctrld config file location."""\n candidates = [\n _Path("/etc/controld/ctrld.toml"),\n _Path("/usr/local/etc/controld/ctrld.toml"),\n _Path.home() / ".config" / "controld" / "ctrld.toml",\n ]\n for p in candidates:\n if p.exists(): return p\n return candidates[0] # default for new install\n\ndef _build_ctrld_toml(vlan_profiles: list) -> str:\n """\n Build a ctrld.toml config for per-VLAN DNS filtering.\n Each VLAN gets its own upstream pointing to its Control D Resolver ID.\n Source IP routing directs traffic to the correct profile automatically.\n\n vlan_profiles: list of { vlan_id, name, subnet, resolver_id }\n """\n lines = [\n "# ctrld configuration — generated by Avaya 5952 Switch Manager",\n "# https://github.com/Control-D-Inc/ctrld",\n "",\n "[service]",\n \'name = "ctrld"\',\n "",\n "# Single listener on port 53 — all VLANs send DNS here",\n "[[listener]]",\n \'ip = "0.0.0.0"\',\n "port = 53",\n \'tag = "all-vlans"\',\n "",\n ]\n\n # One upstream per VLAN\n for vp in vlan_profiles:\n rid = vp.get("resolver_id","").strip()\n if not rid:\n continue\n tag = f"vlan{vp[\'vlan_id\']}"\n lines += [\n f"# VLAN {vp[\'vlan_id\']} — {vp[\'name\']}",\n f"[[upstream]]",\n f\'id = "{tag}"\',\n f\'type = "doh3"\',\n f\'endpoint = "https://dns.controld.com/{rid}"\',\n f\'tag = "{tag}"\',\n "",\n ]\n\n # Routing rules — match source subnet to upstream\n lines += ["# Route each VLAN subnet to its profile"]\n for vp in vlan_profiles:\n rid = vp.get("resolver_id","").strip()\n if not rid:\n continue\n subnet = vp.get("subnet", f"192.168.{vp[\'vlan_id\']}.0/24")\n tag = f"vlan{vp[\'vlan_id\']}"\n lines += [\n f"[[rule]]",\n f\'listener = "all-vlans"\',\n f\'source_ip = "{subnet}"\',\n f\'upstream = "{tag}"\',\n "",\n ]\n\n # Fallback upstream (first valid profile or safe default)\n first_valid = next((vp for vp in vlan_profiles if vp.get("resolver_id")), None)\n if first_valid:\n lines += [\n "# Fallback for unmatched source IPs",\n "[[upstream]]",\n f\'id = "fallback"\',\n f\'type = "doh3"\',\n f\'endpoint = "https://dns.controld.com/{first_valid["resolver_id"]}"\',\n f\'tag = "fallback"\',\n "",\n ]\n\n return "\\n".join(lines)\n\n# ── ctrld API models ────────────────────────────────────────────────────────\n\nclass CtrldVlanProfile(BaseModel):\n vlan_id: int\n name: str\n subnet: str\n resolver_id: str\n\nclass CtrldConfig(BaseModel):\n mode: str # "local" | "opnsense" | "manual"\n vlan_profiles: list[CtrldVlanProfile]\n opnsense_host: Optional[str] = ""\n\nclass CtrldInstallRequest(BaseModel):\n token: str\n config: CtrldConfig\n\nclass CtrldUpdateProfile(BaseModel):\n token: str\n vlan_profiles: list[CtrldVlanProfile]\n\n# ── ctrld endpoints ─────────────────────────────────────────────────────────\n\n@app.get("/api/ctrld/status")\ndef ctrld_status():\n """Return ctrld installation and running status."""\n cfg = _load_ctrld_cfg()\n status = _ctrld_status()\n return {\n **status,\n "configured": bool(cfg.get("mode")),\n "mode": cfg.get("mode",""),\n "vlan_profiles": cfg.get("vlan_profiles",[]),\n "opnsense_host": cfg.get("opnsense_host",""),\n "config_path": str(_ctrld_config_path()),\n "docs_url": "https://docs.controld.com/docs/ctrld",\n }\n\n@app.get("/api/ctrld/toml-preview")\ndef ctrld_toml_preview():\n """Generate and return the ctrld.toml without installing it."""\n cfg = _load_ctrld_cfg()\n profiles = cfg.get("vlan_profiles",[])\n if not profiles:\n raise HTTPException(400, "No VLAN profiles configured yet")\n toml = _build_ctrld_toml(profiles)\n return {"toml": toml}\n\n@app.post("/api/ctrld/save-config")\ndef ctrld_save_config(body: CtrldInstallRequest):\n """\n Save ctrld configuration.\n For \'local\' mode: installs ctrld on this machine, writes config, starts service.\n For \'opnsense\' mode: generates the SSH install command.\n For \'manual\' mode: saves config for reference, generates toml only.\n """\n require_session(body.token)\n\n cfg_dict = {\n "mode": body.config.mode,\n "vlan_profiles": [p.dict() for p in body.config.vlan_profiles],\n "opnsense_host": body.config.opnsense_host,\n }\n _save_ctrld_cfg(cfg_dict)\n\n profiles = [p.dict() for p in body.config.vlan_profiles]\n toml = _build_ctrld_toml(profiles)\n\n if body.config.mode == "local":\n return _ctrld_install_local(toml, profiles)\n elif body.config.mode == "opnsense":\n return _ctrld_generate_opnsense_cmd(body.config.opnsense_host, profiles)\n else:\n # Manual — just return the toml and instructions\n return {\n "success": True,\n "mode": "manual",\n "toml": toml,\n "message": "Config saved. Install ctrld manually and use the toml below.",\n "install_cmd": "sh -c \'sh -c \\"$(curl -sL https://api.controld.com/dl)\\"\'",\n "config_path": str(_ctrld_config_path()),\n }\n\ndef _ctrld_install_local(toml: str, profiles: list) -> dict:\n """Download and install ctrld on this machine, write config, start service."""\n import subprocess as _sp, platform as _platform\n\n # Detect architecture\n arch = _platform.machine().lower()\n os_name = _platform.system().lower()\n\n if os_name != "linux":\n return {\n "success": False,\n "mode": "local",\n "message": f"Auto-install only supported on Linux. "\n f"Download ctrld from https://github.com/Control-D-Inc/ctrld/releases",\n "toml": toml,\n }\n\n # Use ctrld\'s own installer with the first profile\'s resolver ID\n first_rid = next((p["resolver_id"] for p in profiles if p.get("resolver_id")), None)\n if not first_rid:\n return {"success": False, "message": "No Resolver ID provided"}\n\n # Download the binary directly (more reliable than the shell installer for service control)\n log.info("Installing ctrld...")\n\n # Use the official installer\n install_result = _sp.run(\n f\'sh -c \\\'sh -c "$(curl -sL https://api.controld.com/dl)" -s {first_rid} forced\\\'\',\n shell=True, capture_output=True, text=True, timeout=120\n )\n\n if install_result.returncode != 0 and not CTRLD_BIN.exists():\n return {\n "success": False,\n "mode": "local",\n "message": f"Install failed: {install_result.stderr or install_result.stdout}",\n "toml": toml,\n }\n\n # Write our multi-VLAN config (overrides the default single-profile config)\n cfg_path = _ctrld_config_path()\n cfg_path.parent.mkdir(parents=True, exist_ok=True)\n cfg_path.write_text(toml)\n log.info(f"ctrld config written to {cfg_path}")\n\n # Restart ctrld to pick up new config\n _sp.run([str(CTRLD_BIN), "stop"], capture_output=True)\n _sp.run([str(CTRLD_BIN), "start"], capture_output=True)\n\n # Update DHCP on switch — set DNS option 6 per VLAN pool to this machine\'s IP\n import socket as _sock\n try:\n mgmt_ip = _sock.gethostbyname(_sock.gethostname())\n except Exception:\n mgmt_ip = "192.168.99.50"\n\n status = _ctrld_status()\n return {\n "success": status["running"],\n "mode": "local",\n "message": "ctrld installed and running" if status["running"] else "ctrld installed but may not be running — check logs",\n "dns_ip": mgmt_ip,\n "dhcp_action": f"Set DNS (option 6) to {mgmt_ip} on each VLAN pool in the DHCP tab",\n "toml": toml,\n "config_path": str(cfg_path),\n "docs": "https://docs.controld.com/docs/ctrld",\n }\n\ndef _ctrld_generate_opnsense_cmd(opnsense_host: str, profiles: list) -> dict:\n """\n Generate the SSH command to install ctrld on OPNsense.\n User runs this in OPNsense shell.\n """\n first_rid = next((p["resolver_id"] for p in profiles if p.get("resolver_id")), None)\n if not first_rid:\n return {"success": False, "message": "No Resolver ID provided"}\n\n install_cmd = (\n f"sh -c \'sh -c \\"$(curl -sL https://api.controld.com/dl)\\" "\n f"-s {first_rid} forced\'"\n )\n\n toml = _build_ctrld_toml(profiles)\n\n # For OPNsense the config path is different\n opnsense_cfg = "/usr/local/etc/controld/ctrld.toml"\n\n return {\n "success": True,\n "mode": "opnsense",\n "message": "Run the install command in OPNsense shell (SSH or console)",\n "install_cmd": install_cmd,\n "ssh_cmd": f"ssh root@{opnsense_host or \'your-opnsense-ip\'} \'{install_cmd}\'",\n "toml": toml,\n "config_path": opnsense_cfg,\n "step2": f"After install, replace {opnsense_cfg} with the toml config shown below",\n "step3": "Run: ctrld restart",\n "step4": f"Set DNS (option 6) to {opnsense_host or \'OPNsense IP\'} on each VLAN pool",\n "docs": "https://docs.controld.com/docs/routers-platform",\n }\n\n@app.post("/api/ctrld/update-profiles")\ndef ctrld_update_profiles(body: CtrldUpdateProfile):\n """Update VLAN profiles and regenerate/reload ctrld config."""\n require_session(body.token)\n cfg = _load_ctrld_cfg()\n cfg["vlan_profiles"] = [p.dict() for p in body.vlan_profiles]\n _save_ctrld_cfg(cfg)\n\n toml = _build_ctrld_toml(cfg["vlan_profiles"])\n cfg_path = _ctrld_config_path()\n\n if cfg.get("mode") == "local" and cfg_path.exists():\n cfg_path.write_text(toml)\n import subprocess as _sp\n _sp.run([str(CTRLD_BIN), "stop"], capture_output=True)\n _sp.run([str(CTRLD_BIN), "start"], capture_output=True)\n return {"success": True, "message": "Profiles updated and ctrld reloaded", "toml": toml}\n\n return {"success": True, "message": "Profiles saved", "toml": toml,\n "note": "Restart ctrld manually to apply changes" if cfg.get("mode") == "opnsense" else ""}\n\n@app.delete("/api/ctrld/uninstall")\ndef ctrld_uninstall(token: str):\n require_session(token)\n import subprocess as _sp\n if CTRLD_BIN.exists():\n _sp.run([str(CTRLD_BIN), "uninstall", "--cleanup"], capture_output=True)\n if CTRLD_FILE.exists():\n CTRLD_FILE.unlink()\n return {"success": True}\n' + +JSX_SRC = 'import { useState, useCallback, useEffect, useRef } from "react";\n\nconst FONT = `@import url(\'https://fonts.googleapis.com/css2?family=Share+Tech+Mono&family=Barlow:wght@300;400;500;600;700&display=swap\');`;\nconst VLAN_COLORS = ["#00e5ff","#00e676","#ffea00","#ff6d00","#d500f9","#ff1744","#76ff03","#2979ff","#ff4081","#18ffff"];\n\nconst DEFAULT_VLANS = [\n { id: 1, name: "Management", color: VLAN_COLORS[0] },\n { id: 10, name: "Staff", color: VLAN_COLORS[1] },\n { id: 20, name: "Servers", color: VLAN_COLORS[2] },\n { id: 30, name: "IoT", color: VLAN_COLORS[3] },\n { id: 40, name: "Guest", color: VLAN_COLORS[4] },\n { id: 50, name: "Cameras", color: VLAN_COLORS[5] },\n];\n\nconst mkPort = n => ({\n id: n, mode: "access", accessVlan: 1, taggedVlans: [],\n nativeVlan: 1, poe: true, poeLimit: 30000, description: "",\n});\nconst DEFAULT_PORTS = Array.from({ length: 52 }, (_, i) => mkPort(i + 1));\nDEFAULT_PORTS.forEach((p, i) => {\n if (i >= 48) {\n p.mode = "trunk"; p.taggedVlans = [1,10,20,30,40,50];\n p.poe = false; p.description = `SFP+ Uplink ${i-47}`;\n }\n});\n\nconst VISITOR_ID = Math.random().toString(36).slice(2);\n\n// ── CLI Generator with explanations ────────────────────────────────────────\n// Each entry: { cmd: string, explain: string, group: string }\nfunction generateAnnotatedCLI({ ports, vlans, acls, hostname }) {\n const entries = [];\n const h = hostname || "ERS-5952";\n\n const add = (cmd, explain, group) => entries.push({ cmd, explain, group });\n\n add(`hostname ${h}`, `Set switch hostname to "${h}"`, "System");\n\n vlans.forEach(v => {\n add(`vlan create ${v.id} name "${v.name}" type port`,\n `Create VLAN ${v.id} named "${v.name}" — port-based VLAN`, "VLANs");\n });\n\n ports.forEach(p => {\n const iface = p.id <= 48 ? `FastEthernet ${p.id}` : `GigabitEthernet ${p.id}`;\n const portLabel = p.description ? `port ${p.id} (${p.description})` : `port ${p.id}`;\n const vlanName = vlans.find(v => v.id === p.accessVlan)?.name || `VLAN ${p.accessVlan}`;\n const group = `Port ${p.id}${p.description ? ` — ${p.description}` : ""}`;\n\n if (p.description) {\n add(`interface ${iface}`, `Select ${portLabel} for configuration`, group);\n add(` name "${p.description}"`, `Label ${portLabel} as "${p.description}"`, group);\n }\n\n if (p.mode === "disabled") {\n add(`interface ${iface}`, `Select ${portLabel}`, group);\n add(` shutdown`, `Disable ${portLabel} — no traffic will pass`, group);\n } else if (p.mode === "access") {\n add(`vlan members add ${p.accessVlan} ${p.id}`,\n `Assign ${portLabel} to ${vlanName} (VLAN ${p.accessVlan})`, group);\n add(`vlan pvid ${p.id} ${p.accessVlan}`,\n `Set ${portLabel} untagged VLAN to ${vlanName} — devices here join ${vlanName}`, group);\n } else if (p.mode === "trunk") {\n const tagged = p.taggedVlans;\n const names = tagged.map(id => vlans.find(v=>v.id===id)?.name || `VLAN ${id}`).join(", ");\n if (tagged.length) {\n add(`vlan members add ${tagged.join(",")} ${p.id}`,\n `Add ${portLabel} to VLANs: ${names}`, group);\n add(`vlan tagging ${tagged.join(",")} ${p.id}`,\n `Tag traffic on ${portLabel} for VLANs: ${names} — used for uplinks and inter-switch connections`, group);\n }\n const nativeName = vlans.find(v=>v.id===p.nativeVlan)?.name || `VLAN ${p.nativeVlan}`;\n add(`vlan pvid ${p.id} ${p.nativeVlan}`,\n `Set ${portLabel} native (untagged) VLAN to ${nativeName}`, group);\n }\n\n if (p.id <= 48) {\n add(`interface ${iface}`, `Select ${portLabel} for PoE configuration`, group);\n if (!p.poe) {\n add(` no poe enable`, `Disable Power over Ethernet on ${portLabel}`, group);\n } else {\n add(` poe enable`, `Enable Power over Ethernet on ${portLabel}`, group);\n add(` poe poe-limit ${p.poeLimit}`,\n `Limit PoE draw on ${portLabel} to ${(p.poeLimit/1000).toFixed(1)}W — protects switch power budget`, group);\n }\n }\n });\n\n acls.forEach(acl => {\n const vlanName = vlans.find(v=>v.id===acl.applyVlan)?.name || `VLAN ${acl.applyVlan}`;\n const group = `ACL: ${acl.name}`;\n add(`ip access-list extended ${acl.name}`,\n `Create access control list named "${acl.name}"`, group);\n acl.rules.forEach((r, i) => {\n const src = r.srcAny ? "any source" : `source ${r.src}`;\n const dst = r.dstAny ? "any destination" : `destination ${r.dst}`;\n const port = r.port ? ` on port ${r.port}` : "";\n add(` ${i+1} ${r.action} ${r.proto} ${r.srcAny?"any":`${r.src} ${r.srcMask||"0.0.0.255"}`} ${r.dstAny?"any":`${r.dst} ${r.dstMask||"0.0.0.255"}`}${r.port?` eq ${r.port}`:""}`,\n `Rule ${i+1}: ${r.action.toUpperCase()} ${r.proto.toUpperCase()} from ${src} to ${dst}${port}`, group);\n });\n add(`interface vlan ${acl.applyVlan}`,\n `Select ${vlanName} interface to apply the ACL`, group);\n add(` ip access-group ${acl.name} ${acl.direction}`,\n `Apply "${acl.name}" to ${vlanName} — filter traffic going ${acl.direction === "in" ? "INTO" : "OUT OF"} this VLAN`, group);\n });\n\n return entries;\n}\n\nfunction entriesToCLI(entries, hostname) {\n const lines = [\n `! ERS 5952 — ${hostname || "ERS-5952"}`,\n "enable", "configure terminal", "",\n ];\n entries.forEach(e => lines.push(e.cmd));\n lines.push("", "end", "copy running-config nvram:config.cfg");\n return lines.join("\\n");\n}\n\n// ── API ─────────────────────────────────────────────────────────────────────\nconst API = async (path, opts = {}) => {\n const r = await fetch(`/api${path}`, {\n headers: { "Content-Type": "application/json" },\n ...opts,\n body: opts.body ? JSON.stringify(opts.body) : undefined,\n });\n const data = await r.json().catch(() => ({}));\n if (!r.ok) throw new Error(data.detail?.message || data.detail || `HTTP ${r.status}`);\n return data;\n};\n\n// ══════════════════════════════════════════════════════════════════════════════\n// CSS\n// ══════════════════════════════════════════════════════════════════════════════\nconst css = `\n${FONT}\n*,*::before,*::after{box-sizing:border-box;margin:0;padding:0}\n:root{\n --bg:#0a0c10;--sf:#111318;--b1:#1e2230;--b2:#2a2f42;\n --ac:#00e5ff;--ac2:#0077ff;--tx:#c8d0e0;--dm:#5a6070;\n --err:#ff1744;--ok:#00e676;--warn:#ffea00;--warn2:#ff6d00;\n --mono:\'Share Tech Mono\',monospace;--sans:\'Barlow\',sans-serif;\n}\nbody{background:var(--bg);color:var(--tx);font-family:var(--sans);font-size:13px;line-height:1.5;min-height:100vh}\n.app{display:flex;flex-direction:column;min-height:100vh}\n\n/* banner */\n.conn-banner{display:flex;align-items:center;gap:10px;padding:8px 20px;font-size:12px;font-family:var(--mono);border-bottom:1px solid;transition:all .4s}\n.conn-banner.connecting{background:rgba(255,234,0,.06);border-color:rgba(255,234,0,.2);color:var(--warn)}\n.conn-banner.connected{background:rgba(0,230,118,.06);border-color:rgba(0,230,118,.2);color:var(--ok)}\n.conn-banner.error{background:rgba(255,23,68,.06);border-color:rgba(255,23,68,.2);color:var(--err)}\n.conn-banner.hidden{display:none}\n.conn-spin{width:12px;height:12px;border:2px solid currentColor;border-top-color:transparent;border-radius:50%;animation:spin .8s linear infinite;flex-shrink:0}\n@keyframes spin{to{transform:rotate(360deg)}}\n.conn-dot{width:8px;height:8px;border-radius:50%;background:currentColor;flex-shrink:0}\n\n/* topbar */\n.topbar{display:flex;align-items:center;gap:14px;padding:10px 20px;background:var(--sf);border-bottom:1px solid var(--b1);position:sticky;top:0;z-index:100;flex-wrap:wrap}\n.logo{font-family:var(--mono);font-size:14px;color:var(--ac);letter-spacing:2px}\n.logo-sub{font-size:10px;color:var(--dm);font-family:var(--mono)}\n.sp{flex:1}\n.ti{background:var(--bg);border:1px solid var(--b2);color:var(--tx);padding:4px 8px;font-family:var(--mono);font-size:12px;border-radius:3px;width:130px}\n.ti:focus{outline:none;border-color:var(--ac)}\n.tl{font-size:11px;color:var(--dm)}\n\n/* session */\n.sess-btn{display:flex;align-items:center;gap:8px;padding:5px 12px;border-radius:4px;border:1px solid;font-size:11px;font-weight:700;letter-spacing:1px;text-transform:uppercase;cursor:pointer;font-family:var(--sans);transition:all .15s}\n.sess-btn.locked{border-color:var(--b2);color:var(--dm);background:transparent}\n.sess-btn.locked:hover{border-color:var(--ac);color:var(--ac)}\n.sess-btn.unlocked{border-color:var(--ok);color:var(--ok);background:rgba(0,230,118,.08)}\n.sess-btn.unlocked:hover{background:rgba(255,23,68,.1);border-color:var(--err);color:var(--err)}\n.sess-timer{font-family:var(--mono);font-size:12px;letter-spacing:1px}\n\n/* settings icon */\n.settings-btn{background:none;border:1px solid var(--b2);color:var(--dm);border-radius:4px;padding:5px 9px;cursor:pointer;font-size:14px;transition:all .15s;line-height:1}\n.settings-btn:hover{border-color:var(--ac);color:var(--ac)}\n\n/* settings panel */\n.settings-overlay{position:fixed;inset:0;background:rgba(0,0,0,.6);z-index:150;display:flex;align-items:flex-start;justify-content:flex-end}\n.settings-panel{background:var(--sf);border-left:1px solid var(--b2);width:320px;min-height:100vh;padding:20px}\n.settings-panel h2{font-family:var(--mono);font-size:13px;color:var(--ac);letter-spacing:2px;margin-bottom:16px}\n.settings-section{margin-bottom:20px;padding-bottom:20px;border-bottom:1px solid var(--b1)}\n.settings-section:last-child{border-bottom:none}\n.settings-section h3{font-size:10px;font-weight:700;letter-spacing:2px;text-transform:uppercase;color:var(--dm);margin-bottom:10px}\n.setting-row{display:flex;align-items:center;justify-content:space-between;padding:6px 0}\n.setting-label{font-size:12px}\n.setting-sub{font-size:11px;color:var(--dm);margin-top:2px}\n\n/* poll pill */\n.poll-pill{display:flex;align-items:center;gap:5px;font-size:10px;color:var(--dm);font-family:var(--mono);padding:3px 8px;border:1px solid var(--b1);border-radius:20px}\n.dot{width:6px;height:6px;border-radius:50%}\n.dot.ok{background:var(--ok)}.dot.warn{background:var(--warn)}.dot.err{background:var(--err)}.dot.idle{background:var(--dm)}\n\n/* tabs */\n.tabs{display:flex;gap:2px;padding:0 20px;background:var(--sf);border-bottom:1px solid var(--b1)}\n.tab{padding:10px 18px;font-size:12px;font-weight:600;letter-spacing:1px;text-transform:uppercase;cursor:pointer;border-bottom:2px solid transparent;color:var(--dm);transition:all .15s;background:none;border-top:none;border-left:none;border-right:none;font-family:var(--sans)}\n.tab:hover{color:var(--tx)}.tab.active{color:var(--ac);border-bottom-color:var(--ac)}\n\n/* main */\n.main{flex:1;padding:20px;display:flex;gap:16px;align-items:flex-start}\n\n/* panel */\n.panel{background:var(--sf);border:1px solid var(--b1);border-radius:6px;overflow:hidden}\n.ph{padding:10px 14px;border-bottom:1px solid var(--b1);font-size:11px;font-weight:700;letter-spacing:2px;text-transform:uppercase;color:var(--ac);font-family:var(--mono);display:flex;align-items:center;gap:8px}\n.pb{padding:14px}\n\n/* chassis */\n.chassis{background:#0d0f14;border:1px solid var(--b2);border-radius:8px;padding:16px;margin-bottom:12px}\n.cl{font-family:var(--mono);font-size:10px;color:var(--dm);letter-spacing:3px;text-transform:uppercase;margin-bottom:10px}\n.pgrid{display:grid;grid-template-columns:repeat(24,1fr);gap:4px}\n.pgrid-sfp{display:grid;grid-template-columns:repeat(4,1fr);gap:4px;margin-top:8px;padding-top:8px;border-top:1px solid var(--b1);width:calc(4*(100%/24)+3*4px)}\n.port{aspect-ratio:1;border-radius:3px;border:1px solid transparent;cursor:pointer;display:flex;align-items:center;justify-content:center;font-family:var(--mono);font-size:8px;font-weight:700;transition:all .1s;position:relative}\n.port:hover{filter:brightness(1.3);transform:scale(1.1);z-index:2}\n.port.sel{border-color:white!important;box-shadow:0 0 0 2px white;z-index:3}\n.poe-dot{position:absolute;top:1px;right:1px;width:4px;height:4px;border-radius:50%;background:var(--ok)}\n.legend{display:flex;flex-wrap:wrap;gap:8px;padding:10px 14px;border-top:1px solid var(--b1)}\n.li{display:flex;align-items:center;gap:5px;font-size:11px;color:var(--dm)}\n.ld{width:10px;height:10px;border-radius:2px}\n\n/* fields */\n.field{margin-bottom:12px}\n.field label{display:block;font-size:10px;font-weight:700;letter-spacing:1.5px;text-transform:uppercase;color:var(--dm);margin-bottom:4px}\n.field input,.field select{width:100%;background:var(--bg);border:1px solid var(--b2);color:var(--tx);padding:6px 8px;font-family:var(--mono);font-size:12px;border-radius:3px}\n.field input:focus,.field select:focus{outline:none;border-color:var(--ac)}\n.field select option{background:var(--bg)}\n.rgrp{display:flex;gap:6px}\n.rbtn{flex:1;padding:5px;text-align:center;border:1px solid var(--b2);border-radius:3px;cursor:pointer;font-size:11px;font-weight:600;letter-spacing:1px;text-transform:uppercase;transition:all .1s;background:none;color:var(--dm);font-family:var(--sans)}\n.rbtn:hover{border-color:var(--ac);color:var(--ac)}\n.rbtn.ra{background:var(--ac);border-color:var(--ac);color:#000}\n.rbtn.rt{background:var(--ac2);border-color:var(--ac2);color:#fff}\n.rbtn.rd{background:var(--err);border-color:var(--err);color:#fff}\n.trow{display:flex;align-items:center;justify-content:space-between;padding:6px 0}\n.tog{width:36px;height:20px;background:var(--b2);border-radius:10px;cursor:pointer;position:relative;transition:background .2s;border:none}\n.tog.on{background:var(--ok)}\n.tog::after{content:\'\';position:absolute;top:3px;left:3px;width:14px;height:14px;background:#fff;border-radius:50%;transition:transform .2s}\n.tog.on::after{transform:translateX(16px)}\n.vtags{display:flex;flex-wrap:wrap;gap:4px;padding:6px;background:var(--bg);border:1px solid var(--b2);border-radius:3px;min-height:34px}\n.vtag{padding:2px 7px;border-radius:2px;font-family:var(--mono);font-size:11px;cursor:pointer;font-weight:700;transition:opacity .1s}\n.vtag.von{opacity:1}.vtag.voff{opacity:.25}\n\n/* buttons */\n.btn{padding:7px 14px;border-radius:3px;border:none;cursor:pointer;font-size:11px;font-weight:700;letter-spacing:1.5px;text-transform:uppercase;font-family:var(--sans);transition:all .1s;white-space:nowrap}\n.bp{background:var(--ac);color:#000}.bp:hover{background:#33ecff}\n.bd{background:var(--err);color:#fff}.bd:hover{background:#ff4569}\n.bg{background:transparent;border:1px solid var(--b2);color:var(--tx)}.bg:hover{border-color:var(--ac);color:var(--ac)}\n.bs{background:var(--ok);color:#000}.bs:hover{background:#33eb91}\n.bw{background:var(--warn);color:#000}.bw:hover{filter:brightness(1.1)}\n.btn:disabled{opacity:.4;cursor:not-allowed}\n.btn-row{display:flex;gap:6px;margin-top:12px;flex-wrap:wrap}\n\n/* vlan table */\n.vtbl{width:100%;border-collapse:collapse}\n.vtbl th{text-align:left;padding:6px 10px;font-size:10px;letter-spacing:2px;text-transform:uppercase;color:var(--dm);border-bottom:1px solid var(--b1)}\n.vtbl td{padding:8px 10px;border-bottom:1px solid var(--b1);font-family:var(--mono);font-size:12px}\n.vtbl tr:last-child td{border-bottom:none}\n.vtbl tr:hover td{background:rgba(255,255,255,.02)}\n.sw{width:16px;height:16px;border-radius:3px;display:inline-block;vertical-align:middle}\n.ii{background:var(--bg);border:1px solid var(--b2);color:var(--tx);padding:3px 6px;font-family:var(--mono);font-size:12px;border-radius:3px;width:100%}\n.ii:focus{outline:none;border-color:var(--ac)}\n\n/* acl */\n.acl-card{background:var(--bg);border:1px solid var(--b2);border-radius:5px;overflow:hidden;margin-bottom:10px}\n.acl-hd{padding:8px 12px;display:flex;align-items:center;gap:8px;background:rgba(255,255,255,.03);border-bottom:1px solid var(--b1)}\n.acl-nm{font-family:var(--mono);font-size:13px;color:var(--ac);flex:1}\n.acl-rules{padding:8px 12px}\n.acl-rule{display:grid;grid-template-columns:60px 50px 1fr 1fr 60px 24px;gap:6px;align-items:center;margin-bottom:6px;font-size:11px}\n.acl-rule select,.acl-rule input{background:var(--sf);border:1px solid var(--b2);color:var(--tx);padding:3px 5px;font-family:var(--mono);font-size:11px;border-radius:3px;width:100%}\n\n/* ── ANNOTATED CLI TABLE ── */\n.cli-table{width:100%;border-collapse:collapse;font-size:12px;margin-bottom:12px}\n.cli-table th{text-align:left;padding:6px 10px;font-size:10px;letter-spacing:2px;text-transform:uppercase;color:var(--dm);border-bottom:1px solid var(--b1);font-family:var(--sans)}\n.cli-group-header td{padding:8px 10px 4px;font-size:10px;font-weight:700;letter-spacing:2px;text-transform:uppercase;color:var(--ac);font-family:var(--mono);border-top:1px solid var(--b1);background:rgba(0,229,255,.04)}\n.cli-row{transition:background .1s}\n.cli-row:hover{background:rgba(255,255,255,.03)}\n.cli-row td{padding:5px 10px;border-bottom:1px solid rgba(255,255,255,.03);vertical-align:top}\n.cli-cmd{font-family:var(--mono);color:var(--tx);white-space:pre}\n.cli-explain{color:var(--dm);font-size:11px;padding-left:8px}\n.cli-row.pending td{opacity:.5}\n.cli-row.running td .cli-cmd{color:var(--warn)}\n.cli-row.running td .cli-explain{color:var(--warn)}\n.cli-row.success td .cli-cmd{color:var(--ok)}\n.cli-row.success td .cli-explain{color:#3a6040}\n.cli-row.failed td .cli-cmd{color:var(--err)}\n.cli-row.failed td .cli-explain{color:#6a2030}\n.cli-row.waiting td{opacity:.35}\n.cli-status{width:20px;text-align:center;font-size:13px}\n.cli-err-out{font-family:var(--mono);font-size:10px;color:var(--warn);padding:3px 10px 5px;background:rgba(255,23,68,.05);border-bottom:1px solid var(--b1)}\n\n/* raw cli textarea (cli mode) */\n.cli-raw{width:100%;background:#060809;border:1px solid var(--b1);color:#a0b0c0;padding:14px;font-family:var(--mono);font-size:12px;line-height:1.7;border-radius:4px;min-height:300px;resize:vertical}\n.cli-raw:focus{outline:none;border-color:var(--ac)}\n\n/* totp modal */\n.modal-bg{position:fixed;inset:0;background:rgba(0,0,0,.75);z-index:200;display:flex;align-items:center;justify-content:center}\n.modal{background:var(--sf);border:1px solid var(--b2);border-radius:8px;padding:28px;width:360px;text-align:center}\n.modal h2{font-family:var(--mono);color:var(--ac);font-size:14px;letter-spacing:2px;margin-bottom:8px}\n.modal p{font-size:12px;color:var(--dm);margin-bottom:20px;line-height:1.7}\n.totp-in{width:100%;background:var(--bg);border:2px solid var(--b2);color:var(--tx);padding:12px;font-family:var(--mono);font-size:28px;letter-spacing:10px;border-radius:4px;text-align:center;margin-bottom:12px}\n.totp-in:focus{outline:none;border-color:var(--ac)}\n\n/* danger */\n.hard-box{background:rgba(255,23,68,.07);border:1px solid rgba(255,23,68,.3);border-radius:4px;padding:10px 14px;margin-bottom:10px}\n.hard-box h4{color:var(--err);font-size:11px;font-weight:700;letter-spacing:1px;text-transform:uppercase;margin-bottom:6px}\n.warn-box{background:rgba(255,234,0,.07);border:1px solid rgba(255,234,0,.3);border-radius:4px;padding:10px 14px;margin-bottom:10px}\n.warn-box h4{color:var(--warn);font-size:11px;font-weight:700;letter-spacing:1px;text-transform:uppercase;margin-bottom:6px}\n.danger-item{font-size:11px;margin-bottom:5px}\n.danger-cmd{font-family:var(--mono);color:var(--tx)}\n.danger-why{font-size:10px;color:var(--dm);margin-top:1px}\n\n/* push mode choice */\n.push-choice{display:flex;gap:10px;margin-bottom:14px}\n.push-choice-btn{flex:1;padding:10px 12px;border-radius:5px;border:1px solid var(--b2);background:var(--bg);color:var(--tx);cursor:pointer;text-align:left;transition:all .15s;font-family:var(--sans)}\n.push-choice-btn:hover{border-color:var(--ac)}\n.push-choice-btn.chosen{border-color:var(--ac);background:rgba(0,229,255,.06)}\n.push-choice-btn h4{font-size:12px;font-weight:700;margin-bottom:3px;color:var(--ac)}\n.push-choice-btn p{font-size:11px;color:var(--dm);line-height:1.4}\n\n/* push summary bar */\n.push-summary{display:flex;align-items:center;gap:12px;padding:8px 12px;border-radius:4px;font-size:12px;font-family:var(--mono);margin-bottom:10px}\n.push-summary.ok{background:rgba(0,230,118,.08);border:1px solid rgba(0,230,118,.2);color:var(--ok)}\n.push-summary.fail{background:rgba(255,23,68,.08);border:1px solid rgba(255,23,68,.2);color:var(--err)}\n.push-summary.running{background:rgba(255,234,0,.05);border:1px solid rgba(255,234,0,.15);color:var(--warn)}\n\n\n/* ── Device Access ── */\n.device-card{background:var(--bg);border:1px solid var(--b2);border-radius:4px;padding:10px 12px;margin-bottom:8px}\n.device-badges{display:flex;gap:6px;margin-top:6px;flex-wrap:wrap}\n/* misc */\n.sect{font-size:10px;font-weight:700;letter-spacing:2px;text-transform:uppercase;color:var(--dm);margin-bottom:10px;padding-bottom:6px;border-bottom:1px solid var(--b1)}\n.empty{text-align:center;padding:40px 20px;color:var(--dm);font-size:12px}\n.badge{padding:1px 6px;border-radius:10px;font-size:10px;font-family:var(--mono);font-weight:700}\n.poe-bar-wrap{height:4px;background:var(--b2);border-radius:2px;margin-top:4px;overflow:hidden}\n.poe-bar{height:100%;border-radius:2px;background:var(--ok);transition:width .3s}\n`;\n\n// ══════════════════════════════════════════════════════════════════════════════\n// CONNECTION BANNER\n// ══════════════════════════════════════════════════════════════════════════════\nfunction ConnBanner({ state, info }) {\n if (state === "hidden") return null;\n return (\n
\n {state === "connecting" && }\n {state !== "connecting" && }\n \n {state === "connecting" && "Connecting to switch..."}\n {state === "connected" && (info || "Connected")}\n {state === "error" && (info || "Switch unreachable — retrying...")}\n \n
\n );\n}\n\n// ══════════════════════════════════════════════════════════════════════════════\n// TOTP MODAL\n// ══════════════════════════════════════════════════════════════════════════════\nfunction TotpModal({ onSuccess, onCancel, commandCount }) {\n const [code, setCode] = useState("");\n const [error, setError] = useState("");\n const [loading, setLoading] = useState(false);\n\n const verify = async () => {\n if (code.length !== 6) return;\n setLoading(true); setError("");\n try {\n const data = await API("/auth/verify", { method: "POST", body: { code } });\n onSuccess(data.token);\n } catch(e) {\n setError(e.message); setCode("");\n } finally { setLoading(false); }\n };\n\n return (\n
\n
e.stopPropagation()}>\n

◈ AUTHENTICATE

\n

\n Enter your 6-digit TOTP code to authorize this push.
\n {commandCount} command{commandCount!==1?"s":""} ready to send.
\n Session locks automatically when push completes.\n

\n setCode(e.target.value.replace(/\\D/g,"").slice(0,6))}\n onKeyDown={e => e.key==="Enter" && verify()}\n placeholder="000000" autoFocus maxLength={6}/>\n {error &&
{error}
}\n
\n \n \n
\n
\n
\n );\n}\n\n// ══════════════════════════════════════════════════════════════════════════════\n// SESSION BUTTON — shows "Authenticate to Push" only, no timer\n// Session is scoped to one push batch, lock happens automatically after push\n// ══════════════════════════════════════════════════════════════════════════════\nfunction SessionBtn({ session, onUnlock }) {\n if (session) {\n return (\n
\n 🔓 Session Active\n
\n );\n }\n return (\n \n );\n}\n\n// ══════════════════════════════════════════════════════════════════════════════\n// SETTINGS PANEL\n// ══════════════════════════════════════════════════════════════════════════════\nfunction SettingsPanel({ settings, setSettings, onClose }) {\n return (\n
\n
e.stopPropagation()}>\n
\n

◈ SETTINGS

\n \n
\n\n
\n

Interface

\n
\n
\n
CLI Mode
\n
Advanced — type raw ERS commands directly
\n
\n
\n
\n\n
\n

Push Behaviour

\n
\n
\n
Default push mode
\n
Batch or step-by-step
\n
\n \n
\n
\n\n
\n

About

\n
\n ERS 5952 Switch Manager
\n Runs on your management computer on VLAN 99.
\n SSH key lives on the management computer only.
\n You never write CLI commands.
\n README.md has full documentation.\n
\n
\n
\n
\n );\n}\n\n// ══════════════════════════════════════════════════════════════════════════════\n// ANNOTATED CLI TABLE — the main inspect view\n// ══════════════════════════════════════════════════════════════════════════════\nfunction AnnotatedCLITable({ entries, pushState, currentIdx, pushMode, onConfirmOne }) {\n // Group entries by group field\n const groups = [];\n let lastGroup = null;\n entries.forEach((e, i) => {\n if (e.group !== lastGroup) {\n groups.push({ label: e.group, entries: [] });\n lastGroup = e.group;\n }\n groups[groups.length-1].entries.push({ ...e, globalIdx: i });\n });\n\n const getRowState = (globalIdx) => {\n if (!pushState) return "pending";\n if (globalIdx < currentIdx) return "success";\n if (globalIdx === currentIdx) return "running";\n return "waiting";\n };\n\n const getRowResult = (globalIdx) => {\n if (!pushState?.results) return null;\n return pushState.results.find(r => r.index === globalIdx);\n };\n\n return (\n \n \n \n \n \n \n {pushMode === "step" && !pushState?.done && }\n \n \n \n {groups.map((grp, gi) => (\n <>\n \n \n \n {grp.entries.map((e, ei) => {\n const rowState = getRowState(e.globalIdx);\n const result = getRowResult(e.globalIdx);\n const isCurrentStep = pushMode === "step" && pushState && !pushState.done && e.globalIdx === currentIdx;\n return (\n <>\n \n \n \n \n {pushMode === "step" && !pushState?.done && (\n \n )}\n \n {result && !result.success && result.output && (\n \n \n \n )}\n \n );\n })}\n \n ))}\n \n
CommandWhat it does
{grp.label}
\n {rowState === "success" && "✓"}\n {rowState === "failed" && "✗"}\n {rowState === "running" && }\n {(rowState === "pending" || rowState === "waiting") && "·"}\n {e.cmd}{e.explain}\n {isCurrentStep && (\n \n )}\n
\n Switch error: {result.output}\n
\n );\n}\n\n// ══════════════════════════════════════════════════════════════════════════════\n// CLI & PUSH TAB\n// ══════════════════════════════════════════════════════════════════════════════\nfunction CliTab({ ports, vlans, acls, hostname, session, setSession, onNeedAuth, backendOk, settings }) {\n const annotated = generateAnnotatedCLI({ ports, vlans, acls, hostname });\n const pushCommands = annotated.map(e => e.cmd).filter(c => {\n const t = c.trim();\n return t && !t.startsWith("!");\n });\n\n const [stage, setStage] = useState("idle"); // idle|checking|danger|choosePushMode|pushing|done\n const [dangerResult, setDangerResult] = useState(null);\n const [pushMode, setPushMode] = useState(settings.defaultPushMode || "batch");\n const [pushState, setPushState] = useState(null);\n const [currentIdx, setCurrentIdx] = useState(0);\n const [rawCli, setRawCli] = useState("");\n\n // Reset when ports/vlans/acls change\n useEffect(() => { setStage("idle"); setPushState(null); setCurrentIdx(0); }, [ports, vlans, acls]);\n\n const handleReviewPush = async () => {\n setStage("checking");\n try {\n const check = await API("/check/danger", { method:"POST", body:{ commands: pushCommands } });\n setDangerResult(check);\n setStage("danger");\n } catch(e) {\n setDangerResult({ error: e.message }); setStage("danger");\n }\n };\n\n const proceedToChoose = () => setStage("choosePushMode");\n\n const startPush = async (mode, token) => {\n const t = token || session?.token;\n if (!t) { onNeedAuth(); return; }\n setPushMode(mode);\n setStage("pushing");\n setCurrentIdx(0);\n setPushState({ results: [], done: false });\n\n if (mode === "batch") {\n await executeBatch(t);\n }\n // step mode is driven by onConfirmOne\n };\n\n const executeBatch = async (token) => {\n try {\n const result = await API("/switch/push", {\n method:"POST",\n body: { token, commands: pushCommands }\n });\n setPushState({ ...result, done: true });\n setCurrentIdx(result.commands_sent || pushCommands.length);\n setStage("done");\n setSession(null); // lock session after push completes\n } catch(e) {\n setPushState({ success:false, error:e.message, results:[], commands_sent:0, done:true });\n setStage("done");\n setSession(null);\n }\n };\n\n const onConfirmOne = async (idx) => {\n if (!session?.token) { onNeedAuth(); return; }\n const cmd = pushCommands[idx];\n try {\n const result = await API("/switch/push", {\n method:"POST",\n body: { token: session.token, commands: [cmd] }\n });\n const r = result.results?.[0] || { index: idx, command: cmd, success: result.success, output: result.error || "", skipped: false };\n r.index = idx;\n setPushState(prev => {\n const results = [...(prev?.results||[]), r];\n if (!result.success) {\n setStage("done");\n setSession(null);\n return { ...prev, results, done: true, success: false, error: result.error };\n }\n const nextIdx = idx + 1;\n setCurrentIdx(nextIdx);\n if (nextIdx >= pushCommands.length) {\n // Save config\n API("/switch/push", { method:"POST", body:{ token: session.token, commands:["copy running-config nvram:config.cfg"] } })\n .catch(() => {});\n setStage("done");\n setSession(null);\n return { ...prev, results, done:true, success:true, saved:true };\n }\n return { ...prev, results };\n });\n } catch(e) {\n setStage("done"); setSession(null);\n setPushState(prev => ({ ...prev, done:true, success:false, error:e.message }));\n }\n };\n\n const handleRawPush = async () => {\n if (!session?.token) { onNeedAuth(); return; }\n const cmds = rawCli.split("\\n").filter(l => l.trim() && !l.trim().startsWith("!"));\n try {\n setStage("pushing");\n const result = await API("/switch/push", { method:"POST", body:{ token:session.token, commands:cmds } });\n setPushState({ ...result, done:true });\n setStage("done");\n setSession(null);\n } catch(e) {\n setPushState({ success:false, error:e.message, results:[], done:true });\n setStage("done"); setSession(null);\n }\n };\n\n const fullCLI = entriesToCLI(annotated, hostname);\n const cmdCount = pushCommands.length;\n\n // ── CLI Mode (raw text) ──\n if (settings.cliMode) {\n return (\n
\n
\n
◈ CLI Mode — Advanced\n Danger check and allowlist still apply\n
\n
\n
\n Type or paste ERS CLI commands. Do not include enable, configure terminal, end, or copy — these are added automatically.\n
\n