""" ERS 5952 Switch Manager — Backend v3 ──────────────────────────────────────────────────────────────────────── Changes from v2: - Connection pool (30s lifetime, liveness check, auto-invalidate) - Visitor-aware polling — starts when someone hits the page, pauses when no active visitors, stops completely when idle - Hard-block danger patterns (no force override for lethal commands) - Session-based TOTP (one auth → session token → many pushes) - Per-command execution with stop-on-error - Config saved only on full success Requirements: pip install fastapi uvicorn paramiko pyotp Run: python switch_backend.py # start server python switch_backend.py --setup-totp # first-time TOTP setup ──────────────────────────────────────────────────────────────────────── """ import re, sys, time, secrets, logging, threading, os from pathlib import Path from typing import Optional import paramiko, pyotp from fastapi import FastAPI, HTTPException, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel, field_validator logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") log = logging.getLogger("switch-manager") # ══════════════════════════════════════════════════════════════════════ # CONFIG # ══════════════════════════════════════════════════════════════════════ SWITCH_HOST = "192.168.99.1" SWITCH_PORT = 22 SWITCH_USER = "admin" KEY_PATH = "/etc/switch-manager/ers5952_key" KNOWN_HOSTS = "/etc/switch-manager/known_hosts" TOTP_FILE = "/etc/switch-manager/totp_secret" STATIC_DIR = "./frontend/dist" POLL_ACTIVE_S = 15 # poll interval when visitors present POLL_BG_S = 60 # poll interval when tab backgrounded POLL_IDLE_AFTER = 300 # stop polling after this many seconds with no visitors CONN_POOL_MAX_S = 25 # max connection age before refresh (< switch idle timeout) SESSION_TTL_S = 600 # TOTP session — 10 minutes, reset on activity ALLOWED_ORIGINS = ["*"] # tighten to http://192.168.99.X in production # ══════════════════════════════════════════════════════════════════════ # TOTP # ══════════════════════════════════════════════════════════════════════ def get_or_create_totp_secret() -> str: p = Path(TOTP_FILE) if p.exists(): return p.read_text().strip() secret = pyotp.random_base32() p.parent.mkdir(parents=True, exist_ok=True) p.write_text(secret) p.chmod(0o600) log.info(f"New TOTP secret created at {TOTP_FILE}") return secret def setup_totp(): secret = get_or_create_totp_secret() uri = pyotp.TOTP(secret).provisioning_uri( name="ERS5952", issuer_name="SwitchManager") print("\n══════════════════════════════════════════════════") print(" ERS 5952 Switch Manager — TOTP Setup") print("══════════════════════════════════════════════════") print(f"\n Manual entry secret:\n {secret}") print(f"\n Provisioning URI (paste into authenticator app):\n {uri}") print("\n Or generate a QR code:") print(f" python -c \"import qrcode; qrcode.make('{uri}').show()\"") print("\n══════════════════════════════════════════════════\n") TOTP_SECRET: str = "" # ── Session tokens (one TOTP → session, multiple pushes) ────────────── # { token: { expires: float, last_activity: float } } _sessions: dict[str, dict] = {} _sessions_lock = threading.Lock() def create_session() -> str: token = secrets.token_hex(32) now = time.time() with _sessions_lock: _sessions[token] = {"expires": now + SESSION_TTL_S, "last_activity": now} return token def validate_session(token: str) -> bool: """Returns True and refreshes activity timestamp if session is valid.""" now = time.time() with _sessions_lock: # Prune expired expired = [t for t, s in _sessions.items() if s["expires"] < now] for t in expired: del _sessions[t] if token not in _sessions: return False # Refresh on activity _sessions[token]["last_activity"] = now _sessions[token]["expires"] = now + SESSION_TTL_S return True def session_remaining(token: str) -> Optional[int]: with _sessions_lock: s = _sessions.get(token) if not s: return None return max(0, int(s["expires"] - time.time())) def revoke_session(token: str): with _sessions_lock: _sessions.pop(token, None) def require_session(token: str): if not validate_session(token): raise HTTPException(401, "Session expired or invalid — re-authenticate with TOTP") # ══════════════════════════════════════════════════════════════════════ # DANGER DETECTION # Two tiers: # HARD_BLOCK — refused entirely, must run at console # WARN — flagged, push still offered with explicit override # ══════════════════════════════════════════════════════════════════════ HARD_BLOCK_PATTERNS = [ # Management VLAN removal — all variants (re.compile(r'no\s+vlan\s+99\b', re.I), "Deletes management VLAN 99"), (re.compile(r'vlan\s+members\s+remove\b.*\b99\b', re.I), "Removes VLAN 99 from a port — kills management trunk"), (re.compile(r'no\s+vlan\s+tagging\b.*\b99\b', re.I), "Removes VLAN 99 tagging — kills management trunk"), (re.compile(r'vlan\s+pvid\s+\S+\s+\S+', re.I), "Changes native VLAN — verify not management port"), # SSH / IP removal (re.compile(r'no\s+ip\s+ssh', re.I), "Disables SSH entirely — permanent lockout"), (re.compile(r'no\s+ip\s+address\b', re.I), "Removes IP address — will lose connectivity"), # Management interface (re.compile(r'interface\s+vlan\s+99\b', re.I), "Modifies management VLAN interface — run at console"), # Boot / factory (re.compile(r'boot\s+config\s+flags\s+factory', re.I), "Factory reset — run at console"), ] WARN_PATTERNS = [ (re.compile(r'\bshutdown\b', re.I), "Shuts down an interface — confirm it is not your uplink"), (re.compile(r'default\s+interface\b', re.I), "Resets interface to defaults"), (re.compile(r'no\s+vlan\s+\d+\b', re.I), "Deletes a VLAN — confirm no active ports depend on it"), (re.compile(r'spanning-tree\s+.*\s+disable', re.I), "Disables spanning tree — loop risk"), ] def check_danger(commands: list[str]) -> dict: """ Scan a list of CLI commands for dangerous patterns. Returns a dict with: hard_blocked — commands that are refused entirely (e.g. no vlan 99, no ip ssh) warnings — commands that are allowed but flagged (e.g. shutdown) has_hard_block, has_warnings — convenience booleans """ hard, warn = [], [] for cmd in commands: for pat, reason in HARD_BLOCK_PATTERNS: if pat.search(cmd): hard.append({"command": cmd, "reason": reason}) break else: for pat, reason in WARN_PATTERNS: if pat.search(cmd): warn.append({"command": cmd, "reason": reason}) break return { "hard_blocked": hard, "warnings": warn, "has_hard_block": bool(hard), "has_warnings": bool(warn), } # ══════════════════════════════════════════════════════════════════════ # SANITIZATION # ══════════════════════════════════════════════════════════════════════ _BAD_CHARS = re.compile(r'[;&|`$<>(){}\\"\']') _BAD_PATS = [re.compile(p) for p in [r'\n', r'\r', r'--', r'/\*']] _RE_VID = re.compile(r'^\d{1,4}$') _RE_VNAME = re.compile(r'^[a-zA-Z0-9\-_]{1,32}$') _RE_MODE = re.compile(r'^(access|trunk|disabled)$') _RE_ANAME = re.compile(r'^[a-zA-Z0-9\-_]{1,32}$') _RE_DIR = re.compile(r'^(in|out)$') _RE_PROTO = re.compile(r'^(ip|tcp|udp|icmp)$') _RE_ACTION = re.compile(r'^(permit|deny)$') _RE_DESC = re.compile(r'^[a-zA-Z0-9\-_ ]{0,64}$') _RE_IP = re.compile(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$') def _san(v: str, pat: re.Pattern, field: str) -> str: """Reject shell-injection characters and check value against an allow-list regex.""" if _BAD_CHARS.search(v): raise ValueError(f"{field}: disallowed characters") for p in _BAD_PATS: if p.search(v): raise ValueError(f"{field}: injection pattern") if not pat.match(v): raise ValueError(f"{field}: invalid format — {repr(v)}") return v def san_vid(v, field="vlan_id") -> int: """Validate and return a VLAN ID integer (1–4094).""" _san(str(v), _RE_VID, field) vid = int(v) if not 1 <= vid <= 4094: raise ValueError(f"{field}: must be 1–4094") return vid def san_port(v) -> int: """Validate and return a port number (1–52 for the ERS 5952).""" p = int(v) if not 1 <= p <= 52: raise ValueError("port: must be 1–52") return p _ALLOWED_CMD_RE = [ re.compile(r'^vlan\s+(create|members|tagging|pvid)\s'), re.compile(r'^no\s+vlan\s+\d+$'), re.compile(r'^interface\s+(FastEthernet|GigabitEthernet|vlan)\s'), re.compile(r'^\s+(name|no\s+shutdown|poe|speed|duplex|ip\s+access-group|shutdown)\b'), re.compile(r'^hostname\s+\S+$'), re.compile(r'^ip\s+access-list\s+extended\s'), re.compile(r'^\s+\d+\s+(permit|deny)\s'), re.compile(r'^!\s*'), re.compile(r'^\s*$'), ] def is_allowed(cmd: str) -> bool: """Return True if cmd matches the CLI allow-list (whitelist of safe command patterns).""" return any(p.match(cmd) for p in _ALLOWED_CMD_RE) # ══════════════════════════════════════════════════════════════════════ # CONNECTION POOL # ══════════════════════════════════════════════════════════════════════ class SwitchConnectionPool: """ Keeps one SSH connection alive for up to CONN_POOL_MAX_S seconds. On liveness check failure or expiry → opens a fresh connection. Thread-safe. """ def __init__(self): self._conn: Optional[paramiko.SSHClient] = None self._born: float = 0 self._lock = threading.Lock() def get(self) -> paramiko.SSHClient: with self._lock: now = time.time() age = now - self._born if self._conn and age < CONN_POOL_MAX_S: try: transport = self._conn.get_transport() if transport and transport.is_active(): transport.send_ignore() # lightweight liveness ping return self._conn except Exception: log.info("Pool: liveness check failed — opening fresh connection") self._close_unsafe() self._conn = _open_connection() self._born = time.time() log.info("Pool: new connection opened") return self._conn def invalidate(self): with self._lock: self._close_unsafe() def _close_unsafe(self): if self._conn: try: self._conn.close() except Exception: pass self._conn = None self._born = 0 _pool = SwitchConnectionPool() def _open_connection() -> paramiko.SSHClient: client = paramiko.SSHClient() known = Path(KNOWN_HOSTS) if known.exists(): client.load_host_keys(str(known)) client.set_missing_host_key_policy(paramiko.RejectPolicy()) else: log.warning("No known_hosts — trust on first use") client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: client.connect( hostname=SWITCH_HOST, port=SWITCH_PORT, username=SWITCH_USER, key_filename=KEY_PATH, look_for_keys=False, allow_agent=False, timeout=10, banner_timeout=10, ) except paramiko.AuthenticationException: raise HTTPException(503, "SSH auth failed — is the public key loaded on the switch?") except Exception as e: raise HTTPException(503, f"Cannot reach switch at {SWITCH_HOST}: {e}") if not known.exists(): known.parent.mkdir(parents=True, exist_ok=True) client.save_host_keys(str(known)) log.info("Host key pinned to known_hosts") return client def read_cmd(cmd: str) -> str: """Run a read-only command via the pool. Invalidates pool on error.""" try: conn = _pool.get() _, stdout, _ = conn.exec_command(cmd, timeout=10) return stdout.read().decode("utf-8", errors="replace") except HTTPException: raise except Exception as e: log.warning(f"read_cmd failed ({cmd}): {e} — invalidating pool") _pool.invalidate() raise HTTPException(503, f"Switch read failed: {e}") # ── Per-command interactive push ─────────────────────────────────────── _SWITCH_ERR = re.compile( r'%\s*(invalid|error|unknown|bad|failed|cannot|not\s+found|does\s+not\s+exist' r'|incomplete|ambiguous|out\s+of\s+range|already\s+exists)', re.I ) def _run_one(channel, cmd: str) -> tuple[str, bool]: channel.send(cmd + "\n") time.sleep(0.35) out, deadline = "", time.time() + 6 while time.time() < deadline: if channel.recv_ready(): out += channel.recv(4096).decode("utf-8", errors="replace") time.sleep(0.1) else: if out: break time.sleep(0.1) return out.strip(), bool(_SWITCH_ERR.search(out)) def push_one_by_one(commands: list[str]) -> dict: """ Opens a FRESH dedicated connection for push (not from pool — we don't want a push session to corrupt the pool's read connection). Runs commands one at a time, stops on first error. Saves config only on full success. """ _pool.invalidate() # invalidate pool — switch will be busy during push client = _open_connection() results = [] stopped_at = None try: ch = client.invoke_shell() ch.settimeout(10) time.sleep(0.5) if ch.recv_ready(): ch.recv(4096) # drain banner for setup in ["enable", "configure terminal"]: out, err = _run_one(ch, setup) if err: return {"success": False, "saved": False, "error": f"Failed entering config mode: {out}", "results": [], "hint": "Check switch is reachable and credentials are correct"} for i, cmd in enumerate(commands): s = cmd.strip() if not s or s.startswith("!"): results.append({"index": i, "command": cmd, "output": "", "success": True, "skipped": True}) continue log.info(f" [{i+1}/{len(commands)}] {s}") out, had_err = _run_one(ch, s) results.append({"index": i, "command": cmd, "output": out, "success": not had_err, "skipped": False}) if had_err: stopped_at = i log.warning(f" Error at command {i+1}: {out}") break _run_one(ch, "end") saved = False if stopped_at is None: _, save_err = _run_one(ch, "copy running-config nvram:config.cfg") saved = not save_err if saved: log.info("Config saved to NVRAM") else: log.warning("Config save may have failed — verify at console") else: log.warning("Config NOT saved — push stopped on error") ch.close() return { "success": stopped_at is None, "saved": saved, "commands_total": len(commands), "commands_sent": len([r for r in results if not r.get("skipped")]), "stopped_at": stopped_at, "error": results[stopped_at]["output"] if stopped_at is not None else None, "hint": "Remaining commands must be run at the switch console" if stopped_at is not None else None, "results": results, } finally: client.close() # Pool will reopen on next poll naturally # ══════════════════════════════════════════════════════════════════════ # VISITOR-AWARE POLLER # ══════════════════════════════════════════════════════════════════════ _cache: dict = { "port_status": None, "poe_status": None, "vlan_members": None, "sys_info": None, "last_poll": 0, "poll_error": None, } _cache_lock = threading.Lock() # Visitor tracking _visitors: dict[str, float] = {} # { visitor_id: last_seen } _visitors_lock = threading.Lock() _poll_mode = "idle" # idle | active | background def heartbeat(visitor_id: str, mode: str = "active"): """Called by frontend to indicate a visitor is present.""" global _poll_mode with _visitors_lock: _visitors[visitor_id] = time.time() _poll_mode = mode def prune_visitors(): """Remove visitors not seen for POLL_IDLE_AFTER seconds and set mode to idle if none remain.""" global _poll_mode with _visitors_lock: now = time.time() gone = [v for v, t in _visitors.items() if now - t > POLL_IDLE_AFTER] for v in gone: del _visitors[v] if not _visitors: _poll_mode = "idle" def _poll_loop(): """ Background thread: polls the switch for live status at a visitor-adaptive interval. When visitors are active: polls every POLL_ACTIVE_S seconds. When visitors have the tab backgrounded: polls every POLL_BG_S seconds. When no visitors for POLL_IDLE_AFTER seconds: sleeps without polling. Results cached in _cache; poll_error set on SSH failure. """ log.info("Poller thread started") while True: prune_visitors() with _visitors_lock: mode = _poll_mode has_visitors = bool(_visitors) if not has_visitors: # No visitors — sleep and check again, don't poll switch time.sleep(30) continue # Poll the switch try: data = { "port_status": read_cmd("show interfaces"), "poe_status": read_cmd("show poe-port-status"), "vlan_members": read_cmd("show vlan members"), "sys_info": read_cmd("show sys-info"), } with _cache_lock: _cache.update(data) _cache["last_poll"] = time.time() _cache["poll_error"] = None except Exception as e: with _cache_lock: _cache["poll_error"] = str(e) log.warning(f"Poll error: {e}") interval = POLL_ACTIVE_S if mode == "active" else POLL_BG_S time.sleep(interval) def start_poller(): """Launch the background polling thread as a daemon (exits when main process exits).""" t = threading.Thread(target=_poll_loop, daemon=True) t.start() log.info("Poller started") # ══════════════════════════════════════════════════════════════════════ # PYDANTIC MODELS # ══════════════════════════════════════════════════════════════════════ class TotpVerify(BaseModel): code: str class SessionCheck(BaseModel): token: str class SessionRevoke(BaseModel): token: str class Heartbeat(BaseModel): visitor_id: str mode: str = "active" # active | background class PushBatch(BaseModel): token: str commands: list[str] class VlanCreate(BaseModel): token: str vlan_id: int name: str @field_validator("vlan_id") @classmethod def cv(cls, v): return san_vid(v) @field_validator("name") @classmethod def cn(cls, v): return _san(v, _RE_VNAME, "name") class VlanDelete(BaseModel): token: str vlan_id: int @field_validator("vlan_id") @classmethod def cv(cls, v): vid = san_vid(v) if vid == 1: raise ValueError("Cannot delete VLAN 1") return vid class AclRule(BaseModel): action: str proto: str src: Optional[str] = "any" src_mask: Optional[str] = "0.0.0.255" src_any: Optional[bool] = True dst: Optional[str] = "any" dst_mask: Optional[str] = "0.0.0.255" dst_any: Optional[bool] = True port: Optional[str] = "" port_end: Optional[str] = "" # when set, generates "range port port_end" @field_validator("action") @classmethod def ca(cls, v): return _san(v, _RE_ACTION, "action") @field_validator("proto") @classmethod def cp(cls, v): return _san(v, _RE_PROTO, "proto") class AclCreate(BaseModel): token: str name: str apply_vlan: int direction: str rules: list[AclRule] @field_validator("name") @classmethod def cn(cls, v): return _san(v, _RE_ANAME, "name") @field_validator("direction") @classmethod def cd(cls, v): return _san(v, _RE_DIR, "direction") class PortConfig(BaseModel): token: str port: int mode: str access_vlan: Optional[int] = 1 tagged_vlans: Optional[list[int]] = [] native_vlan: Optional[int] = 1 poe: Optional[bool] = True poe_limit_mw: Optional[int] = 30000 description: Optional[str] = "" @field_validator("port") @classmethod def cp(cls, v): return san_port(v) @field_validator("mode") @classmethod def cm(cls, v): return _san(v, _RE_MODE, "mode") @field_validator("poe_limit_mw") @classmethod def cpoe(cls, v): if v is not None and not 1000 <= v <= 30000: raise ValueError("PoE limit: 1000–30000 mW") return v # ══════════════════════════════════════════════════════════════════════ # COMMAND BUILDERS # ══════════════════════════════════════════════════════════════════════ def build_port(cfg: PortConfig) -> list[str]: """ Generate ERS 5952 CLI commands for a port configuration change. Port 1–48 are FastEthernet; ports 49–52 are GigabitEthernet SFP uplinks. PoE is only available on ports 1–48. Returns a list of CLI command strings ready for push_one_by_one(). """ p = cfg.port iface = f"FastEthernet {p}" if p <= 48 else f"GigabitEthernet {p}" cmds = [] if cfg.description: cmds += [f"interface {iface}", f' name "{cfg.description}"'] if cfg.mode == "disabled": cmds += [f"interface {iface}", " shutdown"] elif cfg.mode == "access": vid = san_vid(cfg.access_vlan, "access_vlan") cmds += [f"vlan members add {vid} {p}", f"vlan pvid {p} {vid}"] elif cfg.mode == "trunk": native = san_vid(cfg.native_vlan, "native_vlan") tagged = [san_vid(v, f"tagged_{v}") for v in (cfg.tagged_vlans or [])] if tagged: ts = ",".join(str(v) for v in tagged) cmds += [f"vlan members add {ts} {p}", f"vlan tagging {ts} {p}"] cmds.append(f"vlan pvid {p} {native}") if p <= 48: cmds += [f"interface {iface}", " poe enable" if cfg.poe else " no poe enable"] if cfg.poe: cmds.append(f" poe poe-limit {cfg.poe_limit_mw}") return cmds def build_acl(acl: AclCreate) -> list[str]: """ Generate ERS 5952 CLI commands to create an extended IP ACL and apply it to a VLAN interface. Rules are numbered sequentially starting from 1. The ACL is applied to the VLAN's Layer 3 interface in the specified direction (in/out). """ cmds = [f"ip access-list extended {acl.name}"] for i, r in enumerate(acl.rules): src = "any" if r.src_any else f"{r.src} {r.src_mask}" dst = "any" if r.dst_any else f"{r.dst} {r.dst_mask}" if r.port and r.port_end: port_str = f" range {r.port} {r.port_end}" elif r.port: port_str = f" eq {r.port}" else: port_str = "" cmds.append(f" {i+1} {r.action} {r.proto} {src} {dst}{port_str}") vid = san_vid(acl.apply_vlan, "apply_vlan") cmds += [f"interface vlan {vid}", f" ip access-group {acl.name} {acl.direction}"] return cmds # ══════════════════════════════════════════════════════════════════════ # APP # ══════════════════════════════════════════════════════════════════════ app = FastAPI(title="ERS 5952 Switch Manager", version="3.0.0", docs_url="/api/docs", redoc_url=None) app.add_middleware(CORSMiddleware, allow_origins=ALLOWED_ORIGINS, allow_methods=["GET","POST","DELETE"], allow_headers=["Content-Type"]) @app.exception_handler(ValueError) async def val_err(req: Request, exc: ValueError): return JSONResponse(400, {"detail": str(exc)}) @app.on_event("startup") def startup(): global TOTP_SECRET TOTP_SECRET = get_or_create_totp_secret() log.info("TOTP secret loaded") start_poller() # ── Heartbeat (visitor tracking) ────────────────────────────────────── @app.post("/api/heartbeat") def hb(body: Heartbeat): """ Frontend calls this every 30s to indicate visitor is present. mode = 'active' when tab is visible, 'background' when hidden. Poller adjusts interval accordingly. """ heartbeat(body.visitor_id, body.mode) with _cache_lock: last = _cache["last_poll"] err = _cache["poll_error"] return { "poll_mode": _poll_mode, "last_poll": last, "poll_age": round(time.time() - last, 1) if last else None, "poll_error": err, } # ── Auth ─────────────────────────────────────────────────────────────── @app.post("/api/auth/verify") def verify_totp(body: TotpVerify): """Verify TOTP code — returns session token for multiple pushes.""" if not pyotp.TOTP(TOTP_SECRET).verify(body.code.strip(), valid_window=1): log.warning("TOTP verify failed") raise HTTPException(401, "Invalid TOTP code") token = create_session() log.info("TOTP OK — session created") return { "token": token, "expires_in": SESSION_TTL_S, "message": "Session active — re-authenticates on expiry or manual lock" } @app.post("/api/auth/check") def check_session(body: SessionCheck): """Check if a session is still valid. Returns remaining seconds.""" remaining = session_remaining(body.token) if remaining is None: raise HTTPException(401, "Session expired") return {"valid": True, "remaining": remaining} @app.post("/api/auth/revoke") def revoke(body: SessionRevoke): """Manually lock — invalidates the session immediately.""" revoke_session(body.token) log.info("Session manually revoked") return {"revoked": True} # ── Read endpoints (no auth) ─────────────────────────────────────────── @app.get("/api/status") def status(): """Backend and switch connectivity summary (no auth required).""" with _cache_lock: return { "backend": "online", "switch_host": SWITCH_HOST, "key_exists": Path(KEY_PATH).exists(), "known_hosts_pinned": Path(KNOWN_HOSTS).exists(), "last_poll": _cache["last_poll"], "poll_age": round(time.time() - _cache["last_poll"], 1) if _cache["last_poll"] else None, "poll_error": _cache["poll_error"], "poll_mode": _poll_mode, } @app.get("/api/live") def live(): """Cached live switch data — updated by background poller.""" with _cache_lock: age = time.time() - _cache["last_poll"] if _cache["last_poll"] else None return { **_cache, "stale": age is None or age > max(POLL_ACTIVE_S, POLL_BG_S) * 3 } @app.get("/api/switch/config") def running_config(): """Fetch and return the full switch running config (read-only, no auth).""" out = read_cmd("show running-config") return {"config": out, "lines": len(out.splitlines())} # ── Danger pre-flight (no auth — check before prompting TOTP) ───────── @app.post("/api/check/danger") def danger_check(body: dict): """ Pre-flight danger check — call this before showing TOTP prompt. Returns hard_blocked, warnings, and safe_to_push flag. No auth required so the user sees danger info before authenticating. """ cmds = body.get("commands", []) result = check_danger(cmds) rejected = [c for c in cmds if not is_allowed(c)] result["rejected_by_allowlist"] = rejected result["safe_to_push"] = ( not result["has_hard_block"] and not rejected ) return result # ── Push endpoints (require session token) ──────────────────────────── @app.post("/api/switch/push") def push(body: PushBatch): """ Push CLI batch one command at a time. Requires valid session token. Hard-blocked commands are refused — no override. Warn-level commands proceed (user was already shown the warning). Stops on first switch error. Config saved only on full success. """ require_session(body.token) danger = check_danger(body.commands) if danger["has_hard_block"]: raise HTTPException(400, { "message": "Hard-blocked commands detected — these must be run at the switch console", "blocked": danger["hard_blocked"], }) rejected = [c for c in body.commands if not is_allowed(c)] if rejected: raise HTTPException(400, { "message": "Commands failed allowlist validation", "rejected": rejected[:10], }) log.info(f"Push: {len(body.commands)} commands") return push_one_by_one(body.commands) @app.post("/api/switch/vlan") def create_vlan(body: VlanCreate): """Create a new VLAN on the switch (type port = standard Layer 2 VLAN).""" require_session(body.token) return push_one_by_one( [f'vlan create {body.vlan_id} name "{body.name}" type port']) @app.delete("/api/switch/vlan/{vlan_id}") def delete_vlan(vlan_id: int, token: str): """Delete a VLAN by ID. VLAN 1 is blocked at model level; VLAN 99 is blocked by danger check.""" require_session(token) return push_one_by_one([f"no vlan {san_vid(vlan_id)}"]) @app.post("/api/switch/port") def configure_port(body: PortConfig): """Apply port configuration: mode (access/trunk/disabled), VLAN, PoE, description.""" require_session(body.token) return push_one_by_one(build_port(body)) @app.post("/api/switch/acl") def create_acl(body: AclCreate): """Create an extended IP ACL and apply it to a VLAN interface.""" require_session(body.token) return push_one_by_one(build_acl(body)) # ── Serve React app ──────────────────────────────────────────────────── if os.path.isdir(STATIC_DIR): app.mount("/", StaticFiles(directory=STATIC_DIR, html=True), name="frontend") # ══════════════════════════════════════════════════════════════════════ if __name__ == "__main__": if "--setup-totp" in sys.argv: setup_totp() sys.exit(0) import uvicorn uvicorn.run("switch_backend:app", host="0.0.0.0", port=8765, log_level="info") # ══════════════════════════════════════════════════════════════════════ # DEVICE ACCESS — DHCP LEASES, MAC RESERVATIONS, ACL PINHOLES # ══════════════════════════════════════════════════════════════════════ import json as _json from pathlib import Path as _Path DEVICES_FILE = _Path("/etc/switch-manager/devices.json") def _load_devices() -> list: """Load the saved device list from devices.json, returning [] on missing or corrupt file.""" if DEVICES_FILE.exists(): try: return _json.loads(DEVICES_FILE.read_text()) except: pass return [] def _save_devices(devices: list): """Persist the device list to devices.json with 2-space indentation.""" DEVICES_FILE.write_text(_json.dumps(devices, indent=2)) def _parse_dhcp_leases(raw: str) -> list: """Parse ERS 5952 'show dhcp-server leases' output.""" import re leases = [] for line in raw.splitlines(): # Format: IP MAC State Remaining Hostname m = re.match( r'\s*(\d+\.\d+\.\d+\.\d+)\s+' r'([0-9a-fA-F]{2}[:\-][0-9a-fA-F]{2}[:\-][0-9a-fA-F]{2}' r'[:\-][0-9a-fA-F]{2}[:\-][0-9a-fA-F]{2}[:\-][0-9a-fA-F]{2})\s+' r'(\S+)\s+(\S+)\s*(.*)', line) if m: leases.append({ "ip": m.group(1), "mac": m.group(2).lower().replace('-',':'), "state": m.group(3), "remaining": m.group(4), "hostname": m.group(5).strip() or "unknown", }) return leases def _parse_arp_table(raw: str) -> list: """Parse 'show arp' for additional device discovery.""" import re entries = [] for line in raw.splitlines(): m = re.match( r'\s*(\d+\.\d+\.\d+\.\d+)\s+' r'([0-9a-fA-F]{2}[:\-][0-9a-fA-F]{2}[:\-][0-9a-fA-F]{2}' r'[:\-][0-9a-fA-F]{2}[:\-][0-9a-fA-F]{2}[:\-][0-9a-fA-F]{2})', line) if m: entries.append({ "ip": m.group(1), "mac": m.group(2).lower().replace('-',':'), "state": "arp", "remaining": "-", "hostname": "", }) return entries class DeviceEntry(BaseModel): name: str mac: str ip: str vlan: int = 10 management_access: bool = False static_ip: bool = False notes: Optional[str] = "" class DeviceUpdate(BaseModel): token: str device: DeviceEntry class DeviceDelete(BaseModel): token: str mac: str class PinholeRequest(BaseModel): token: str mac: str allow: bool = True def _build_dhcp_reservation_cmds(device: DeviceEntry) -> list: """Generate ERS 5952 CLI for DHCP reservation (static binding).""" mac_clean = device.mac.replace(':','-').upper() return [ f"ip dhcp-server static-binding {device.ip}", f" mac-address {mac_clean}", f" client-name \"{device.name}\"", ] def _build_pinhole_acl_cmds(device: DeviceEntry, mgmt_ip: str, allow: bool) -> list: """Generate ACL commands to allow/deny a device IP to reach management.""" acl_name = f"MGMT-ACCESS" if allow: return [ f"ip access-list extended {acl_name}", f" permit tcp host {device.ip} host {mgmt_ip} eq 443", f" permit tcp host {device.ip} host {mgmt_ip} eq 8765", ] else: return [ f"ip access-list extended {acl_name}", f" no permit tcp host {device.ip} host {mgmt_ip}", ] @app.get("/api/devices") def get_devices(): """Return saved device list plus live DHCP leases from switch.""" saved = _load_devices() live_leases = [] try: dhcp_raw = read_cmd("show dhcp-server leases") arp_raw = read_cmd("show arp") live_leases = _parse_dhcp_leases(dhcp_raw) # Merge ARP entries not already in leases arp = _parse_arp_table(arp_raw) lease_ips = {l["ip"] for l in live_leases} for entry in arp: if entry["ip"] not in lease_ips: live_leases.append(entry) except Exception as e: log.warning(f"Could not pull DHCP/ARP from switch: {e}") return { "saved": saved, "live": live_leases, "mgmt_ip": SWITCH_HOST, } @app.post("/api/devices/save") def save_device(body: DeviceUpdate): """Save or update a device entry (upsert by MAC address).""" require_session(body.token) devices = _load_devices() existing = next((i for i, d in enumerate(devices) if d["mac"] == body.device.mac), None) device_dict = body.device.dict() if existing is not None: devices[existing] = device_dict else: devices.append(device_dict) _save_devices(devices) log.info(f"Device saved: {body.device.name} ({body.device.mac})") return {"success": True, "devices": devices} @app.post("/api/devices/delete") def delete_device(body: DeviceDelete): """Remove a device from the saved list by MAC address.""" require_session(body.token) devices = [d for d in _load_devices() if d["mac"] != body.mac] _save_devices(devices) return {"success": True} @app.post("/api/devices/push-reservation") def push_reservation(body: DeviceUpdate): """Push a DHCP static binding for this device to the switch.""" require_session(body.token) cmds = _build_dhcp_reservation_cmds(body.device) danger = check_danger(cmds) if danger["has_hard_block"]: raise HTTPException(400, {"message": "Blocked", "blocked": danger["hard_blocked"]}) log.info(f"Pushing DHCP reservation for {body.device.name}") return push_one_by_one(cmds) @app.post("/api/devices/push-pinhole") def push_pinhole(body: PinholeRequest): """Add or remove an ACL pinhole for a device to reach management.""" require_session(body.token) devices = _load_devices() device = next((DeviceEntry(**d) for d in devices if d["mac"] == body.mac), None) if not device: raise HTTPException(404, "Device not found — save it first") import socket try: mgmt_ip = socket.gethostbyname(socket.gethostname()) except Exception: mgmt_ip = SWITCH_HOST.rsplit('.',1)[0] + '.50' cmds = _build_pinhole_acl_cmds(device, mgmt_ip, body.allow) log.info(f"Pinhole {'allow' if body.allow else 'deny'} for {device.name} ({device.ip})") return push_one_by_one(cmds) # ══════════════════════════════════════════════════════════════════════ # WIREGUARD PEER MANAGEMENT # ══════════════════════════════════════════════════════════════════════ WG_CLIENT_DIR_API = _Path("/etc/switch-manager/clients") WG_CONF_PATH = _Path("/etc/wireguard/wg0.conf") class WGClientRequest(BaseModel): token: str name: str class WGRevokeRequest(BaseModel): token: str name: str def _wg_genkey_api(): """Generate a WireGuard private/public keypair using the system wg tool.""" import subprocess as _sp priv = _sp.run(["wg","genkey"], capture_output=True, text=True).stdout.strip() pub = _sp.run(["wg","pubkey"], input=priv, capture_output=True, text=True).stdout.strip() return priv, pub def _wg_status() -> dict: """Return parsed WireGuard interface status including connected peers.""" import subprocess as _sp try: raw = _sp.run(["wg","show"], capture_output=True, text=True).stdout peers = [] current = {} for line in raw.splitlines(): line = line.strip() if line.startswith("peer:"): if current: peers.append(current) current = {"public_key": line.split(":",1)[1].strip()} elif line.startswith("endpoint:"): current["endpoint"] = line.split(":",1)[1].strip() elif line.startswith("latest handshake:"): current["last_handshake"] = line.split(":",1)[1].strip() elif line.startswith("transfer:"): current["transfer"] = line.split(":",1)[1].strip() elif line.startswith("allowed ips:"): current["allowed_ips"] = line.split(":",1)[1].strip() if current: peers.append(current) # Match peers to named client files named = {} if WG_CLIENT_DIR_API.exists(): for f in WG_CLIENT_DIR_API.glob("*.conf"): txt = f.read_text() import re m = re.search(r'PublicKey\s*=\s*(\S+)', txt) if m: named[m.group(1)] = f.stem for p in peers: p["name"] = named.get(p.get("public_key",""), "unknown") return {"running": True, "peers": peers} except Exception as e: return {"running": False, "error": str(e), "peers": []} @app.get("/api/wireguard/status") def wg_status(): return _wg_status() @app.get("/api/wireguard/clients") def wg_clients(): clients = [] if WG_CLIENT_DIR_API.exists(): for f in sorted(WG_CLIENT_DIR_API.glob("*.conf")): clients.append({"name": f.stem, "file": str(f)}) return {"clients": clients} @app.post("/api/wireguard/add-client") def wg_add_client(body: WGClientRequest): require_session(body.token) if not WG_CONF_PATH.exists(): raise HTTPException(503, "WireGuard not configured on this machine") import re, subprocess as _sp, socket # Next available IP conf_text = WG_CONF_PATH.read_text() used = set() for m in re.finditer(r'AllowedIPs\s*=\s*(\S+)', conf_text): used.add(m.group(1).split('/')[0]) subnet = "10.99.0" num = 2 while f"{subnet}.{num}" in used and num < 254: num += 1 client_ip = f"{subnet}.{num}" # Server public key server_pub_path = _Path("/etc/switch-manager/wg_server_public") if not server_pub_path.exists(): raise HTTPException(503, "Server public key not found") server_pub = server_pub_path.read_text().strip() # Client keys c_priv, c_pub = _wg_genkey_api() # Peer entry in server config peer = f"\n[Peer]\n# {body.name}\nPublicKey = {c_pub}\nAllowedIPs = {client_ip}/32\n" with open(WG_CONF_PATH, 'a') as f: f.write(peer) # Reload live _sp.run(["wg","addconf","wg0","/dev/stdin"], input=f"[Peer]\nPublicKey = {c_pub}\nAllowedIPs = {client_ip}/32\n", capture_output=True, text=True) # Public IP for endpoint try: pub_ip = _sp.run(["curl","-s","--max-time","5","https://api.ipify.org"], capture_output=True, text=True).stdout.strip() except Exception: pub_ip = socket.gethostbyname(socket.gethostname()) # Get management IP from switch config mgmt_subnet = '.'.join(SWITCH_HOST.split('.')[:3]) + '.0/24' client_conf = ( f"[Interface]\nPrivateKey = {c_priv}\nAddress = {client_ip}/24\n" f"DNS = {subnet}.1\n\n" f"[Peer]\nPublicKey = {server_pub}\n" f"Endpoint = {pub_ip}:51820\n" f"AllowedIPs = {mgmt_subnet}, {subnet}.0/24\n" f"PersistentKeepalive = 25\n" ) WG_CLIENT_DIR_API.mkdir(exist_ok=True) client_file = WG_CLIENT_DIR_API / f"{body.name}.conf" client_file.write_text(client_conf) client_file.chmod(0o600) log.info(f"WireGuard client added: {body.name} → {client_ip}") return { "success": True, "name": body.name, "client_ip": client_ip, "config": client_conf, "file": str(client_file), } @app.post("/api/wireguard/revoke-client") def wg_revoke_client(body: WGRevokeRequest): require_session(body.token) import re, subprocess as _sp client_file = WG_CLIENT_DIR_API / f"{body.name}.conf" if not client_file.exists(): raise HTTPException(404, f"Client '{body.name}' not found") # Get client public key txt = client_file.read_text() m = re.search(r'\[Peer\].*?PublicKey\s*=\s*(\S+)', txt, re.DOTALL) client_pub = m.group(1) if m else None # Remove from server config if WG_CONF_PATH.exists(): conf = WG_CONF_PATH.read_text() # Remove the [Peer] block for this client cleaned = re.sub( rf'\n\[Peer\]\n# {re.escape(body.name)}\n.*?(?=\n\[Peer\]|\Z)', '', conf, flags=re.DOTALL ) WG_CONF_PATH.write_text(cleaned) # Remove live peer if client_pub: _sp.run(["wg","set","wg0","peer",client_pub,"remove"], capture_output=True, text=True) # Delete client file client_file.unlink() log.info(f"WireGuard client revoked: {body.name}") return {"success": True} @app.get("/api/wireguard/client-qr/{name}") def wg_client_qr(name: str): """Return client config as text for QR generation in frontend.""" client_file = WG_CLIENT_DIR_API / f"{name}.conf" if not client_file.exists(): raise HTTPException(404, f"Client '{name}' not found") return {"name": name, "config": client_file.read_text()} # ══════════════════════════════════════════════════════════════════════ # DHCP MANAGEMENT — SWITCH + OPNSENSE UNIFIED VIEW # ══════════════════════════════════════════════════════════════════════ import urllib.request as _urlreq import urllib.error as _urlerr import ssl as _ssl import base64 as _b64 OPNSENSE_FILE = _Path("/etc/switch-manager/opnsense.json") def _load_opnsense_cfg() -> dict: """Load saved OPNsense API credentials from opnsense.json, returning {} if absent.""" if OPNSENSE_FILE.exists(): try: return _json.loads(OPNSENSE_FILE.read_text()) except: pass return {} def _save_opnsense_cfg(cfg: dict): """Persist OPNsense API credentials to opnsense.json (chmod 600 — contains secrets).""" OPNSENSE_FILE.write_text(_json.dumps(cfg, indent=2)) OPNSENSE_FILE.chmod(0o600) def _opnsense_request(cfg: dict, path: str, method="GET", body=None) -> dict: """Make an authenticated request to the OPNsense API.""" host = cfg.get("host","") key = cfg.get("key","") secret = cfg.get("secret","") if not host or not key or not secret: raise ValueError("OPNsense not configured") url = f"https://{host}/api/{path}" creds = _b64.b64encode(f"{key}:{secret}".encode()).decode() ctx = _ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = _ssl.CERT_NONE headers = { "Authorization": f"Basic {creds}", "Content-Type": "application/json", } data = _json.dumps(body).encode() if body else None req = _urlreq.Request(url, data=data, headers=headers, method=method) try: with _urlreq.urlopen(req, timeout=5, context=ctx) as r: return _json.loads(r.read().decode()) except _urlerr.HTTPError as e: raise ValueError(f"OPNsense API error {e.code}: {e.reason}") except Exception as e: raise ValueError(f"OPNsense unreachable: {e}") def _detect_opnsense_host(gateway_ip: str) -> str | None: """Try to reach OPNsense API at the gateway IP.""" try: ctx = _ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = _ssl.CERT_NONE url = f"https://{gateway_ip}/api/core/firmware/status" req = _urlreq.Request(url, headers={"User-Agent":"switch-manager/1"}) _urlreq.urlopen(req, timeout=3, context=ctx) return gateway_ip except Exception: return None def _get_opnsense_reservations(cfg: dict) -> list: """Fetch DHCP static mappings from OPNsense.""" try: data = _opnsense_request(cfg, "dhcpv4/leases/searchReservation") rows = data.get("rows", []) return [ { "ip": r.get("ipaddr",""), "mac": r.get("mac","").lower(), "hostname": r.get("hostname",""), "descr": r.get("descr",""), "if": r.get("if",""), "source": "opnsense", "uuid": r.get("uuid",""), } for r in rows if r.get("mac") ] except Exception as e: log.warning(f"OPNsense reservations fetch failed: {e}") return [] def _get_opnsense_leases(cfg: dict) -> list: """Fetch active DHCP leases from OPNsense.""" try: data = _opnsense_request(cfg, "dhcpv4/leases/searchLease") rows = data.get("rows", []) return [ { "ip": r.get("address",""), "mac": r.get("mac","").lower(), "hostname": r.get("hostname",""), "state": r.get("state",""), "if": r.get("if",""), "source": "opnsense_lease", } for r in rows if r.get("mac") ] except Exception as e: log.warning(f"OPNsense leases fetch failed: {e}") return [] def _get_switch_reservations() -> list: """Fetch DHCP static bindings from ERS 5952.""" import re as _re try: raw = read_cmd("show dhcp-server static-binding") bindings = [] current = {} for line in raw.splitlines(): m = _re.match(r'\s*IP Address:\s*(\S+)', line) if m: if current: bindings.append(current) current = {"ip": m.group(1), "mac":"", "hostname":"", "source":"switch"} m2 = _re.match(r'\s*MAC Address:\s*(\S+)', line) if m2 and current: current["mac"] = m2.group(1).lower().replace('-',':') m3 = _re.match(r'\s*Client Name:\s*(\S+)', line) if m3 and current: current["hostname"] = m3.group(1) if current and current.get("ip"): bindings.append(current) return bindings except Exception as e: log.warning(f"Switch DHCP reservation fetch failed: {e}") return [] def _get_switch_dhcp_status() -> dict: """Check if switch DHCP server is running and which VLANs it serves.""" import re as _re try: raw = read_cmd("show dhcp-server") running = "enabled" in raw.lower() or "active" in raw.lower() vlans = _re.findall(r'VLAN\s+(\d+)', raw, _re.I) return {"running": running, "vlans": list(set(vlans))} except Exception: return {"running": False, "vlans": []} def _get_relay_status() -> dict: """Read current DHCP relay (ip helper-address) config from each VLAN interface.""" import re as _re try: raw = read_cmd("show ip helper-address") configured = {} for line in raw.splitlines(): # Typical output: " 10 192.168.99.1" m = _re.match(r'\s*(\d+)\s+(\d+\.\d+\.\d+\.\d+)', line) if m: configured[int(m.group(1))] = m.group(2) return {"vlans": configured, "ok": True} except Exception as e: log.warning(f"Relay status fetch failed: {e}") return {"vlans": {}, "ok": False} def _build_relay_cmds(opnsense_ip: str, vlan_ids: list) -> list: """Generate ERS 5952 CLI to set ip helper-address on the specified VLANs.""" cmds = [] for vid in vlan_ids: cmds += [ f"interface vlan {vid}", f" ip helper-address {opnsense_ip}", ] return cmds def _find_conflicts(switch_res: list, opnsense_res: list) -> list: """ Find same MAC in both switch and OPNsense. Flag if IPs differ (conflict) or same (duplicate — harmless but messy). """ switch_by_mac = {r["mac"]: r for r in switch_res if r.get("mac")} conflicts = [] for r in opnsense_res: mac = r.get("mac","") if mac and mac in switch_by_mac: sw = switch_by_mac[mac] conflicts.append({ "mac": mac, "hostname": r.get("hostname") or sw.get("hostname",""), "switch_ip": sw["ip"], "opnsense_ip": r["ip"], "ip_conflict": sw["ip"] != r["ip"], "opnsense_uuid": r.get("uuid",""), }) return conflicts # ── OPNsense config models ───────────────────────────────────────────────── class OPNsenseConfig(BaseModel): host: str key: str secret: str class OPNsenseReservationPush(BaseModel): token: str mac: str ip: str hostname: str descr: Optional[str] = "" iface: Optional[str] = "lan" class SyncRequest(BaseModel): token: str mac: str direction: str # "to_switch" | "to_opnsense" | "remove_switch" | "remove_opnsense" class RelayConfig(BaseModel): token: str opnsense_ip: str vlans: list = [10, 20, 30, 40, 50] # VLANs to relay; 99 is always local @field_validator("opnsense_ip") @classmethod def cip(cls, v): return _san(v, _RE_IP, "opnsense_ip") # ── DHCP endpoints ───────────────────────────────────────────────────────── @app.get("/api/dhcp/overview") def dhcp_overview(): """ Unified DHCP view: - Switch static bindings + active leases - OPNsense reservations + leases (if configured) - Conflicts (same MAC, different IP) - Which DHCP server is active per VLAN """ switch_res = _get_switch_reservations() switch_leases = [] switch_status = _get_switch_dhcp_status() # Also pull ARP for discovery try: arp_raw = read_cmd("show arp") dhcp_raw = read_cmd("show dhcp-server leases") switch_leases = _parse_dhcp_leases(dhcp_raw) + _parse_arp_table(arp_raw) # Deduplicate by IP seen_ips = set() unique_leases = [] for l in switch_leases: if l["ip"] not in seen_ips: seen_ips.add(l["ip"]) unique_leases.append(l) switch_leases = unique_leases except Exception as e: log.warning(f"Switch lease fetch failed: {e}") cfg = _load_opnsense_cfg() opnsense_res = _get_opnsense_reservations(cfg) if cfg.get("key") else [] opnsense_leases = _get_opnsense_leases(cfg) if cfg.get("key") else [] conflicts = _find_conflicts(switch_res, opnsense_res) relay_status = _get_relay_status() # Which VLANs have switch DHCP vs OPNsense # Switch DHCP VLANs from status, OPNsense VLANs inferred from interface names opnsense_ifaces = list({r.get("if","") for r in opnsense_res + opnsense_leases if r.get("if")}) return { "switch": { "status": switch_status, "reservations": switch_res, "leases": switch_leases, }, "opnsense": { "configured": bool(cfg.get("key")), "host": cfg.get("host",""), "reservations": opnsense_res, "leases": opnsense_leases, "interfaces": opnsense_ifaces, }, "relay": relay_status, "conflicts": conflicts, "has_conflicts": len(conflicts) > 0, } @app.get("/api/dhcp/detect-opnsense") def detect_opnsense_endpoint(): """Auto-detect OPNsense at the gateway IP.""" import re as _re try: route = read_cmd("show ip route default") m = _re.search(r'(\d+\.\d+\.\d+\.\d+)', route) gateway = m.group(1) if m else None except Exception: gateway = None if not gateway: # Try from switch management IP parts = SWITCH_HOST.split('.') parts[-1] = '1' gateway = '.'.join(parts) host = _detect_opnsense_host(gateway) return { "detected": host is not None, "host": host, "gateway": gateway, "configured": bool(_load_opnsense_cfg().get("key")), } @app.post("/api/dhcp/configure-opnsense") def configure_opnsense(body: OPNsenseConfig): """Save OPNsense API credentials. Tests connectivity first.""" cfg = {"host": body.host, "key": body.key, "secret": body.secret} try: result = _opnsense_request(cfg, "core/firmware/status") _save_opnsense_cfg(cfg) log.info(f"OPNsense configured: {body.host}") return {"success": True, "version": result.get("product_version","unknown")} except ValueError as e: raise HTTPException(400, str(e)) @app.delete("/api/dhcp/configure-opnsense") def remove_opnsense_config(): """Remove OPNsense API credentials.""" if OPNSENSE_FILE.exists(): OPNSENSE_FILE.unlink() return {"success": True} @app.post("/api/dhcp/push-to-opnsense") def push_reservation_to_opnsense(body: OPNsenseReservationPush): """Create a DHCP static mapping in OPNsense.""" require_session(body.token) cfg = _load_opnsense_cfg() if not cfg.get("key"): raise HTTPException(503, "OPNsense not configured") try: result = _opnsense_request(cfg, "dhcpv4/reservations/addReservation", "POST", { "reservation": { "interface": body.iface, "mac": body.mac, "ipaddr": body.ip, "hostname": body.hostname, "descr": body.descr or f"Added by switch-manager", } }) # Apply changes _opnsense_request(cfg, "dhcpv4/service/reconfigure", "POST") log.info(f"OPNsense reservation pushed: {body.mac} → {body.ip}") return {"success": True, "result": result} except ValueError as e: raise HTTPException(400, str(e)) @app.post("/api/dhcp/sync") def sync_reservation(body: SyncRequest): """ Sync a reservation between switch and OPNsense. Directions: to_switch — copy OPNsense reservation to switch to_opnsense — copy switch reservation to OPNsense remove_switch — remove from switch only remove_opnsense — remove from OPNsense only """ require_session(body.token) cfg = _load_opnsense_cfg() overview = dhcp_overview() # Find the device in both sources sw_res = next((r for r in overview["switch"]["reservations"] if r["mac"]==body.mac), None) ops_res = next((r for r in overview["opnsense"]["reservations"] if r["mac"]==body.mac), None) if body.direction == "to_switch": if not ops_res: raise HTTPException(404, "OPNsense reservation not found") cmds = _build_dhcp_reservation_cmds(type('D',(),{ "ip": ops_res["ip"], "mac": ops_res["mac"], "name": ops_res.get("hostname","") })()) return push_one_by_one(cmds) elif body.direction == "to_opnsense": if not sw_res: raise HTTPException(404, "Switch reservation not found") if not cfg.get("key"): raise HTTPException(503, "OPNsense not configured") result = _opnsense_request(cfg, "dhcpv4/reservations/addReservation", "POST", { "reservation": { "mac": sw_res["mac"], "ipaddr": sw_res["ip"], "hostname": sw_res.get("hostname",""), "descr": "Synced from switch", "interface": "lan", } }) _opnsense_request(cfg, "dhcpv4/service/reconfigure", "POST") return {"success": True, "result": result} elif body.direction == "remove_switch": if not sw_res: raise HTTPException(404, "Switch reservation not found") return push_one_by_one([f"no ip dhcp-server static-binding {sw_res['ip']}"]) elif body.direction == "remove_opnsense": if not ops_res or not ops_res.get("uuid"): raise HTTPException(404, "OPNsense reservation not found or missing UUID") if not cfg.get("key"): raise HTTPException(503, "OPNsense not configured") _opnsense_request(cfg, f"dhcpv4/reservations/delReservation/{ops_res['uuid']}", "DELETE") _opnsense_request(cfg, "dhcpv4/service/reconfigure", "POST") return {"success": True} raise HTTPException(400, f"Unknown direction: {body.direction}") @app.get("/api/dhcp/relay/status") def relay_status_endpoint(): """Return current ip helper-address config from the switch per VLAN.""" return _get_relay_status() @app.post("/api/dhcp/relay/configure") def configure_relay(body: RelayConfig): """ Push ip helper-address to each non-management VLAN so the switch relays DHCP requests to OPNsense. VLAN 99 is never relayed — it stays local as the management / recovery path. """ require_session(body.token) safe_vlans = [int(v) for v in body.vlans if int(v) != 99] if not safe_vlans: raise HTTPException(400, "No VLANs to configure (VLAN 99 is excluded)") cmds = _build_relay_cmds(body.opnsense_ip, safe_vlans) return push_one_by_one(cmds) # ══════════════════════════════════════════════════════════════════════ # CONTROL D / ctrld DNS MANAGEMENT # ══════════════════════════════════════════════════════════════════════ CTRLD_FILE = _Path("/etc/switch-manager/ctrld.json") # NOTE: /usr/local/bin/ctrld is the Linux default path. # On OPNsense (FreeBSD) ctrld installs to /usr/local/sbin/ctrld. # For local-mode installs this path is checked at runtime, so it's fine. # For OPNsense mode the binary runs on the router, not here — the path is irrelevant. CTRLD_BIN = _Path("/usr/local/bin/ctrld") def _load_ctrld_cfg() -> dict: """Load saved ctrld configuration (mode, vlan_profiles) from ctrld.json.""" if CTRLD_FILE.exists(): try: return _json.loads(CTRLD_FILE.read_text()) except: pass return {} def _save_ctrld_cfg(cfg: dict): """Persist ctrld configuration to ctrld.json (chmod 600 — contains Resolver IDs).""" CTRLD_FILE.write_text(_json.dumps(cfg, indent=2)) CTRLD_FILE.chmod(0o600) def _ctrld_status() -> dict: """Check if ctrld is running and which mode.""" import subprocess as _sp try: r = _sp.run([str(CTRLD_BIN), "status"], capture_output=True, text=True, timeout=5) running = r.returncode == 0 return { "installed": CTRLD_BIN.exists(), "running": running, "output": r.stdout.strip() or r.stderr.strip(), } except Exception as e: return { "installed": CTRLD_BIN.exists(), "running": False, "output": str(e), } def _ctrld_config_path() -> _Path: """Find ctrld config file location.""" candidates = [ _Path("/etc/controld/ctrld.toml"), _Path("/usr/local/etc/controld/ctrld.toml"), _Path.home() / ".config" / "controld" / "ctrld.toml", ] for p in candidates: if p.exists(): return p return candidates[0] # default for new install def _build_ctrld_toml(vlan_profiles: list, local_domain: str = "lan", local_resolver: str = "") -> str: """ Build a ctrld.toml in the correct format — table notation, not TOML arrays. The correct ctrld format uses [listener.0], [network.N], [upstream.N] table sections, NOT [[listener]] / [[upstream]] / [[rule]] array tables. Source VLAN routing is done via [network.N] sections (CIDR-based) referenced in the [listener.0.policy].networks array. Domain-specific overrides go in .rules. vlan_profiles: list of { vlan_id, name, subnet, resolver_id } local_domain: suffix for internal hostnames (default 'lan') local_resolver: if set (e.g. '127.0.0.1:5353'), adds split-horizon upstream and rules so *.lan / *.local go to the local resolver instead of Control D — keeps .lan names working for all VLAN clients. Control D bootstrap IP 76.76.2.0 is used for cold-start before DoH is up. """ BOOTSTRAP = "76.76.2.0" # Control D anycast — required for cold-start active = [vp for vp in vlan_profiles if vp.get("resolver_id", "").strip()] lines = [ "# ctrld configuration — generated by Avaya 5952 Switch Manager", "# Documentation: https://docs.controld.com/docs/ctrld", "", "[service]", " log_level = 'info'", " log_path = '/tmp/ctrld.log'", "", ] # ── Listener with per-VLAN policy ──────────────────────────────────────── # The policy.networks array maps each [network.N] to one [upstream.N]. # This is how ctrld routes different VLAN subnets to different profiles. lines += [ "[listener]", " [listener.0]", " ip = '0.0.0.0'", " port = 53", " [listener.0.policy]", ] if active: net_entries = [f" " + "{ " + f"'network.{i}' = ['upstream.{i}']" + " }," for i in range(len(active))] lines += [" networks = ["] + net_entries + [" ]"] else: lines += [" networks = []"] # Domain-specific rules (split-horizon for .lan / .local → local resolver) if local_resolver: domain_suffix = local_domain.strip(".") lines += [ " rules = [", f" " + "{ " + f"'*.{domain_suffix}' = ['upstream.local']" + " },", " " + "{ " + "'*.local' = ['upstream.local']" + " },", " ]", ] else: lines += [" rules = []"] lines += [""] # ── Network sections — one per VLAN ────────────────────────────────────── if active: lines += ["[network]"] for i, vp in enumerate(active): vid = vp["vlan_id"] name = vp.get("name", f"VLAN{vid}") subnet = vp.get("subnet", f"192.168.{vid}.0/24") lines += [ f" # VLAN {vid} — {name}", f" [network.{i}]", f" name = '{name}'", f" cidrs = ['{subnet}']", "", ] # ── Upstream sections — one per VLAN plus optional local ────────────────── lines += ["[upstream]"] for i, vp in enumerate(active): vid = vp["vlan_id"] rid = vp["resolver_id"].strip() lines += [ f" # VLAN {vid} — {vp.get('name', '')}", f" [upstream.{i}]", f" type = 'doh'", f" endpoint = 'https://dns.controld.com/{rid}'", f" bootstrap_ip = '{BOOTSTRAP}'", f" timeout = 5000", "", ] # Optional local resolver for split-horizon .lan resolution (dnsmasq/Unbound) if local_resolver: lines += [ f" # Local resolver — handles *.{local_domain} and *.local", f" # dnsmasq on port 5353 (Docker) or Unbound on 127.0.0.1:5353 (OPNsense)", f" [upstream.local]", f" type = 'legacy'", f" endpoint = '{local_resolver}'", f" timeout = 2000", "", ] return "\n".join(lines) # ── ctrld API models ──────────────────────────────────────────────────────── class CtrldVlanProfile(BaseModel): vlan_id: int name: str subnet: str resolver_id: str class CtrldConfig(BaseModel): mode: str # "local" | "opnsense" | "manual" vlan_profiles: list[CtrldVlanProfile] opnsense_host: Optional[str] = "" class CtrldInstallRequest(BaseModel): token: str config: CtrldConfig class CtrldUpdateProfile(BaseModel): token: str vlan_profiles: list[CtrldVlanProfile] # ── ctrld endpoints ───────────────────────────────────────────────────────── @app.get("/api/ctrld/status") def ctrld_status(): """Return ctrld installation and running status.""" cfg = _load_ctrld_cfg() status = _ctrld_status() return { **status, "configured": bool(cfg.get("mode")), "mode": cfg.get("mode",""), "vlan_profiles": cfg.get("vlan_profiles",[]), "opnsense_host": cfg.get("opnsense_host",""), "config_path": str(_ctrld_config_path()), "docs_url": "https://docs.controld.com/docs/ctrld", } @app.get("/api/ctrld/toml-preview") def ctrld_toml_preview(): """Generate and return the ctrld.toml without installing it.""" cfg = _load_ctrld_cfg() profiles = cfg.get("vlan_profiles",[]) if not profiles: raise HTTPException(400, "No VLAN profiles configured yet") toml = _build_ctrld_toml(profiles) return {"toml": toml} @app.post("/api/ctrld/save-config") def ctrld_save_config(body: CtrldInstallRequest): """ Save ctrld configuration. For 'local' mode: installs ctrld on this machine, writes config, starts service. For 'opnsense' mode: generates the SSH install command. For 'manual' mode: saves config for reference, generates toml only. """ require_session(body.token) cfg_dict = { "mode": body.config.mode, "vlan_profiles": [p.dict() for p in body.config.vlan_profiles], "opnsense_host": body.config.opnsense_host, } _save_ctrld_cfg(cfg_dict) profiles = [p.dict() for p in body.config.vlan_profiles] toml = _build_ctrld_toml(profiles) if body.config.mode == "local": return _ctrld_install_local(toml, profiles) elif body.config.mode == "opnsense": return _ctrld_generate_opnsense_cmd(body.config.opnsense_host, profiles) else: # Manual — just return the toml and instructions return { "success": True, "mode": "manual", "toml": toml, "message": "Config saved. Install ctrld manually and use the toml below.", "install_cmd": "sh -c 'sh -c \"$(curl -sL https://api.controld.com/dl)\"'", "config_path": str(_ctrld_config_path()), } def _fix_port53_conflict() -> dict: """ Detect and fix systemd-resolved holding port 53 (common on Ubuntu/Debian). systemd-resolved's stub listener binds 127.0.0.53:53 and sometimes 0.0.0.0:53, which blocks ctrld from binding port 53. The right fix is to disable only the stub listener — NOT the service itself (the service still handles /etc/resolv.conf and local hostname resolution). Returns a dict with keys: needed (bool), fixed (bool), message (str). """ import subprocess as _sp # Check if systemd-resolved is running and holding port 53 try: ss_out = _sp.run( ["ss", "-tlnp", "sport", "=", ":53"], capture_output=True, text=True, timeout=5 ).stdout if "systemd-resolve" not in ss_out and "resolved" not in ss_out: return {"needed": False, "fixed": False, "message": "No port 53 conflict detected"} except Exception: return {"needed": False, "fixed": False, "message": "Could not check port 53 status (ss not available)"} log.info("systemd-resolved is holding port 53 — disabling stub listener") resolved_conf = _Path("/etc/systemd/resolved.conf") try: current = resolved_conf.read_text() if resolved_conf.exists() else "" except Exception as e: return {"needed": True, "fixed": False, "message": f"Cannot read {resolved_conf}: {e}"} # Already fixed? if "DNSStubListener=no" in current: _sp.run(["systemctl", "restart", "systemd-resolved"], capture_output=True) return {"needed": True, "fixed": True, "message": "DNSStubListener=no already present — restarted systemd-resolved"} # Add the setting under [Resolve], creating the section if needed if "[Resolve]" in current: new_conf = current.rstrip() + "\nDNSStubListener=no\n" else: new_conf = current.rstrip() + "\n[Resolve]\nDNSStubListener=no\n" try: resolved_conf.write_text(new_conf) except PermissionError: return {"needed": True, "fixed": False, "message": "Permission denied writing /etc/systemd/resolved.conf — run backend as root or with sudo"} restart = _sp.run(["systemctl", "restart", "systemd-resolved"], capture_output=True, text=True) if restart.returncode != 0: return {"needed": True, "fixed": False, "message": f"Added DNSStubListener=no but systemd-resolved restart failed: {restart.stderr}"} log.info("Port 53 conflict resolved — systemd-resolved stub listener disabled") return {"needed": True, "fixed": True, "message": "Disabled systemd-resolved stub listener (DNSStubListener=no) and restarted service"} def _ctrld_install_local(toml: str, profiles: list) -> dict: """Download and install ctrld on this machine, write config, start service.""" import subprocess as _sp, platform as _platform # Detect architecture arch = _platform.machine().lower() os_name = _platform.system().lower() if os_name != "linux": return { "success": False, "mode": "local", "message": f"Auto-install only supported on Linux. " f"Download ctrld from https://github.com/Control-D-Inc/ctrld/releases", "toml": toml, } # Use ctrld's own installer with the first profile's resolver ID first_rid = next((p["resolver_id"] for p in profiles if p.get("resolver_id")), None) if not first_rid: return {"success": False, "message": "No Resolver ID provided"} # Fix port 53 conflict BEFORE installing ctrld — on Ubuntu/Debian, systemd-resolved # holds port 53 and ctrld cannot bind. Disabling the stub listener is safe: # systemd-resolved keeps running for /etc/resolv.conf management. port53_fix = _fix_port53_conflict() log.info(f"Port 53 check: {port53_fix['message']}") # Download the binary directly (more reliable than the shell installer for service control) log.info("Installing ctrld...") # Use the official installer install_result = _sp.run( f'sh -c \'sh -c "$(curl -sL https://api.controld.com/dl)" -s {first_rid} forced\'', shell=True, capture_output=True, text=True, timeout=120 ) if install_result.returncode != 0 and not CTRLD_BIN.exists(): return { "success": False, "mode": "local", "message": f"Install failed: {install_result.stderr or install_result.stdout}", "toml": toml, "port53": port53_fix, } # Write our multi-VLAN config (overrides the default single-profile config) cfg_path = _ctrld_config_path() cfg_path.parent.mkdir(parents=True, exist_ok=True) cfg_path.write_text(toml) log.info(f"ctrld config written to {cfg_path}") # Restart ctrld to pick up new config _sp.run([str(CTRLD_BIN), "stop"], capture_output=True) _sp.run([str(CTRLD_BIN), "start"], capture_output=True) # Update DHCP on switch — set DNS option 6 per VLAN pool to this machine's IP import socket as _sock try: mgmt_ip = _sock.gethostbyname(_sock.gethostname()) except Exception: mgmt_ip = "192.168.99.50" status = _ctrld_status() return { "success": status["running"], "mode": "local", "message": "ctrld installed and running" if status["running"] else "ctrld installed but may not be running — check logs", "dns_ip": mgmt_ip, "dhcp_action": f"Set DNS (option 6) to {mgmt_ip} on each VLAN pool in the DHCP tab", "toml": toml, "config_path": str(cfg_path), "docs": "https://docs.controld.com/docs/ctrld", "port53": port53_fix, } def _ctrld_generate_opnsense_cmd(opnsense_host: str, profiles: list) -> dict: """ Generate the SSH command to install ctrld on OPNsense. User runs this in OPNsense shell. """ first_rid = next((p["resolver_id"] for p in profiles if p.get("resolver_id")), None) if not first_rid: return {"success": False, "message": "No Resolver ID provided"} install_cmd = ( f"sh -c 'sh -c \"$(curl -sL https://api.controld.com/dl)\" " f"-s {first_rid} forced'" ) toml = _build_ctrld_toml(profiles) # For OPNsense the config path is different opnsense_cfg = "/usr/local/etc/controld/ctrld.toml" return { "success": True, "mode": "opnsense", "message": "Run the install command in OPNsense shell (SSH or console)", "install_cmd": install_cmd, "ssh_cmd": f"ssh root@{opnsense_host or 'your-opnsense-ip'} '{install_cmd}'", "toml": toml, "config_path": opnsense_cfg, "step2": f"After install, replace {opnsense_cfg} with the toml config shown below", "step3": "Run: ctrld restart", "step4": f"Set DNS (option 6) to {opnsense_host or 'OPNsense IP'} on each VLAN pool", "docs": "https://docs.controld.com/docs/routers-platform", } @app.post("/api/ctrld/update-profiles") def ctrld_update_profiles(body: CtrldUpdateProfile): """Update VLAN profiles and regenerate/reload ctrld config.""" require_session(body.token) cfg = _load_ctrld_cfg() cfg["vlan_profiles"] = [p.dict() for p in body.vlan_profiles] _save_ctrld_cfg(cfg) toml = _build_ctrld_toml(cfg["vlan_profiles"]) cfg_path = _ctrld_config_path() if cfg.get("mode") == "local" and cfg_path.exists(): cfg_path.write_text(toml) import subprocess as _sp _sp.run([str(CTRLD_BIN), "stop"], capture_output=True) _sp.run([str(CTRLD_BIN), "start"], capture_output=True) return {"success": True, "message": "Profiles updated and ctrld reloaded", "toml": toml} return {"success": True, "message": "Profiles saved", "toml": toml, "note": "Restart ctrld manually to apply changes" if cfg.get("mode") == "opnsense" else ""} @app.delete("/api/ctrld/uninstall") def ctrld_uninstall(token: str): """Stop ctrld, remove its binary, and delete saved config.""" require_session(token) import subprocess as _sp if CTRLD_BIN.exists(): _sp.run([str(CTRLD_BIN), "uninstall", "--cleanup"], capture_output=True) if CTRLD_FILE.exists(): CTRLD_FILE.unlink() return {"success": True} # ── DNS enforcement ACL generation ────────────────────────────────────────── class DnsEnforceRequest(BaseModel): """Request body for generating DNS enforcement ACLs.""" token: str ctrld_ip: str # IP of the machine running ctrld (becomes the only allowed DNS target) vlan_ids: list[int] # VLANs to enforce (excludes VLAN 99 management) def _build_dns_enforce_acls(ctrld_ip: str, vlans_info: list[dict]) -> list[str]: """ Generate CLI commands for DNS enforcement ACLs on each VLAN interface. For each VLAN the ACL: - Permits UDP/TCP port 53 to ctrld_ip (allows DHCP-assigned DNS) - Denies UDP/TCP port 53 to anywhere (blocks direct DNS bypass e.g. 8.8.8.8) - Denies TCP port 853 to anywhere (blocks DNS-over-TLS bypass) - Permits everything else (internet still works) Without these rules a device can ignore DHCP-assigned DNS and use 8.8.8.8 directly, bypassing all ctrld filtering entirely. vlans_info: list of { vlan_id: int, subnet: str } e.g. { vlan_id: 10, subnet: "192.168.10.0/24" } """ # Validate ctrld IP — must be a bare IP address, no injection import ipaddress as _ip try: ctrld_addr = str(_ip.ip_address(ctrld_ip)) except ValueError: raise ValueError(f"ctrld_ip: invalid IP address {repr(ctrld_ip)}") cmds = [] for vi in vlans_info: vid = san_vid(vi["vlan_id"]) subnet = vi.get("subnet", f"192.168.{vid}.0/24") # Parse subnet into network/wildcard for ERS ACL syntax try: net = _ip.ip_network(subnet, strict=False) net_str = str(net.network_address) wild = str(_ip.ip_address(int(net.hostmask))) except ValueError: net_str = f"192.168.{vid}.0" wild = "0.0.0.255" acl_name = f"DNS-ENFORCE-VLAN{vid}" cmds += [ f"ip access-list extended {acl_name}", # 1 & 2: permit DNS to ctrld only (DHCP-assigned resolver) f" 1 permit udp {net_str} {wild} host {ctrld_addr} eq 53", f" 2 permit tcp {net_str} {wild} host {ctrld_addr} eq 53", # 3 & 4: deny DNS to anywhere else (block 8.8.8.8 and friends) f" 3 deny udp {net_str} {wild} any eq 53", f" 4 deny tcp {net_str} {wild} any eq 53", # 5: deny DNS-over-TLS (port 853) so devices can't use DoT as bypass f" 5 deny tcp {net_str} {wild} any eq 853", # 6: permit everything else — internet still works f" 6 permit ip any any", # Apply inbound on the VLAN interface f"interface vlan {vid}", f" ip access-group {acl_name} in", ] return cmds @app.post("/api/ctrld/dns-enforce-acls") def ctrld_dns_enforce_acls(body: DnsEnforceRequest): """ Generate DNS enforcement ACL commands for the requested VLANs. Returns the raw CLI commands for review — the caller then pushes them via the normal TOTP-gated push endpoint. This endpoint only generates; it does NOT push anything to the switch itself. """ require_session(body.token) # Refuse to touch VLAN 99 (management) — a broken ACL there = lockout safe_vlans = [v for v in body.vlan_ids if v != 99] if not safe_vlans: raise HTTPException(400, "No safe VLANs to enforce — VLAN 99 is excluded automatically") vlans_info = [{"vlan_id": v} for v in safe_vlans] try: cmds = _build_dns_enforce_acls(body.ctrld_ip, vlans_info) except ValueError as e: raise HTTPException(400, str(e)) return { "success": True, "commands": cmds, "count": len(cmds), "note": "Review these commands then push via the Review & Push tab", "vlans": safe_vlans, "ctrld_ip": body.ctrld_ip, } # ── Local hostname resolution (dnsmasq) ────────────────────────────────────── LOCAL_HOSTNAMES_FILE = _Path("/etc/switch-manager/local-hostnames.json") DNSMASQ_CONF_PATH = _Path("/etc/switch-manager/dnsmasq.conf") def _load_local_hostnames() -> list: """Load user-defined hostname→IP mappings for .lan resolution.""" if LOCAL_HOSTNAMES_FILE.exists(): try: return _json.loads(LOCAL_HOSTNAMES_FILE.read_text()) except: pass return [] def _save_local_hostnames(entries: list): """Persist hostname→IP mappings (used to generate dnsmasq.conf).""" LOCAL_HOSTNAMES_FILE.write_text(_json.dumps(entries, indent=2)) def _generate_dnsmasq_conf(entries: list, mgmt_ip: str = "192.168.99.50") -> str: """ Build a dnsmasq.conf for local .lan hostname resolution. dnsmasq runs on port 5353 inside Docker alongside the switch manager. ctrld.toml forwards *.lan and *.local queries to 127.0.0.1:5353. This keeps local names working even when all external DNS goes through ctrld. entries: list of { name: str, ip: str } mgmt_ip: IP of the management computer (switch.mgmt.lan and management.lan point here) """ import ipaddress as _ip lines = [ "# dnsmasq — local .lan hostname resolution", "# Generated by Avaya 5952 Switch Manager", "# Listens on port 5353 (mapped from Docker container port 53)", "# ctrld forwards *.lan and *.local here", "", "port=53", # dnsmasq internal port (Docker maps host:5353 → container:53) "no-resolv", # don't use /etc/resolv.conf — this is a local-only resolver "no-hosts", # don't use /etc/hosts "domain-needed", # never forward bare names upstream "bogus-priv", # don't forward RFC1918 PTR queries upstream "", "# Management computer — always present", f"address=/switch.mgmt.lan/{mgmt_ip}", f"address=/management.lan/{mgmt_ip}", "", ] if entries: lines += ["# User-defined hostnames"] for e in entries: hostname = e.get("name","").strip() ip_addr = e.get("ip","").strip() if not hostname or not ip_addr: continue # Validate the IP — skip malformed entries try: _ip.ip_address(ip_addr) except ValueError: continue # Strip leading/trailing dots, sanitise hostname hostname = hostname.strip(".") if not hostname: continue lines.append(f"address=/{hostname}/{ip_addr}") return "\n".join(lines) + "\n" def _generate_ctrld_split_horizon_block(local_domain: str = "lan", dnsmasq_port: int = 5353) -> str: """ Generate an example ctrld.toml for split-horizon DNS with a local resolver. In the correct ctrld format, split-horizon is done by: 1. Adding [upstream.local] with type='legacy' pointing to dnsmasq/Unbound 2. Adding rules in [listener.0.policy].rules that send *.lan → upstream.local Because the format uses indexed table sections ([network.N], [upstream.N]), you can't simply append a fragment — the full toml must be regenerated via _build_ctrld_toml(vlan_profiles, local_resolver='127.0.0.1:5353'). This function returns a plain-English example for display only. """ port = dnsmasq_port domain = local_domain.strip(".") return "\n".join([ "# Add to your ctrld.toml — regenerate via DNS tab for correct indexing", "", "# In [listener.0.policy], add to the rules array:", f"# {{ '*.{domain}' = ['upstream.local'] }},", "# { '*.local' = ['upstream.local'] },", "", "# Add a new upstream section (increment index as needed):", "[upstream.local]", f" type = 'legacy'", f" endpoint = '127.0.0.1:{port}'", f" timeout = 2000", "", f"# Then restart ctrld: ctrld restart", ]) class LocalHostnameEntry(BaseModel): name: str # e.g. "printer.lan" ip: str # e.g. "192.168.10.50" class LocalHostnamesUpdate(BaseModel): token: str entries: list[LocalHostnameEntry] local_domain: Optional[str] = "lan" @app.get("/api/dns/local-hostnames") def get_local_hostnames(): """Return saved local hostname mappings and the generated dnsmasq.conf.""" entries = _load_local_hostnames() import socket as _sock try: mgmt_ip = _sock.gethostbyname(_sock.gethostname()) except Exception: mgmt_ip = "192.168.99.50" conf = _generate_dnsmasq_conf(entries, mgmt_ip) return { "entries": entries, "dnsmasq_conf": conf, "conf_path": str(DNSMASQ_CONF_PATH), } @app.post("/api/dns/local-hostnames") def save_local_hostnames(body: LocalHostnamesUpdate): """ Save local hostname mappings, write dnsmasq.conf, and return the updated ctrld.toml split-horizon block to append (user applies it via the DNS tab). """ require_session(body.token) entries = [e.dict() for e in body.entries] _save_local_hostnames(entries) import socket as _sock try: mgmt_ip = _sock.gethostbyname(_sock.gethostname()) except Exception: mgmt_ip = "192.168.99.50" conf = _generate_dnsmasq_conf(entries, mgmt_ip) DNSMASQ_CONF_PATH.parent.mkdir(parents=True, exist_ok=True) DNSMASQ_CONF_PATH.write_text(conf) # Regenerate ctrld.toml with split-horizon enabled (if ctrld is configured) ctrld_cfg = _load_ctrld_cfg() split_horizon_toml = None if ctrld_cfg.get("vlan_profiles"): split_horizon_toml = _build_ctrld_toml( ctrld_cfg["vlan_profiles"], local_domain=body.local_domain or "lan", local_resolver=f"127.0.0.1:5353", ) # Write new toml if running locally if ctrld_cfg.get("mode") == "local": cfg_path = _ctrld_config_path() if cfg_path.parent.exists(): cfg_path.write_text(split_horizon_toml) split_horizon = _generate_ctrld_split_horizon_block( local_domain=body.local_domain or "lan" ) return { "success": True, "entries": entries, "dnsmasq_conf": conf, "conf_path": str(DNSMASQ_CONF_PATH), "split_horizon": split_horizon, "full_toml": split_horizon_toml, "docker_compose_snippet": ( " dnsmasq:\n" " image: andyshinn/dnsmasq:latest\n" " ports:\n" " - \"5353:53/udp\"\n" " - \"5353:53/tcp\"\n" " volumes:\n" " - /etc/switch-manager/dnsmasq.conf:/etc/dnsmasq.conf:ro\n" " restart: unless-stopped\n" " cap_add:\n" " - NET_ADMIN\n" ), "message": ( f"Saved {len(entries)} hostname(s). " "Add the docker-compose snippet and split_horizon block to ctrld.toml, " "then run: docker compose up -d dnsmasq" ), } # ══════════════════════════════════════════════════════════════════════ # OPNSENSE WIREGUARD — ROUTER-LEVEL VPN WITH PER-VLAN ACCESS CONTROL # ══════════════════════════════════════════════════════════════════════ # # Moves WireGuard from the management computer onto OPNsense so any # device on any VLAN can VPN home without touching the management PC. # Each peer is granted access only to the VLANs you choose. # # Architecture: # OPNsense wg1 interface (10.99.2.0/24 — separate from local wg0) # Peer Alice → tunnel IP 10.99.2.2 → allowed VLAN 10 + VLAN 20 # Peer Bob → tunnel IP 10.99.2.3 → allowed VLAN 10 only # Private keys are generated here and stored only on the mgmt PC. # OPNsense receives only the public key (standard WireGuard practice). OPN_WG_FILE = _Path("/etc/switch-manager/opnsense_wg.json") def _load_opnsense_wg() -> dict: if OPN_WG_FILE.exists(): try: return _json.loads(OPN_WG_FILE.read_text()) except Exception: pass return {} def _save_opnsense_wg(cfg: dict): OPN_WG_FILE.parent.mkdir(parents=True, exist_ok=True) OPN_WG_FILE.write_text(_json.dumps(cfg, indent=2)) OPN_WG_FILE.chmod(0o600) class OPNWGServerSetup(BaseModel): token: str server_name: str = "switch-mgmt-vpn" listen_port: int = 51820 tunnel_subnet: str = "10.99.2.0/24" public_endpoint: str = "" # public IP or DDNS hostname for client configs class OPNWGAddPeer(BaseModel): token: str name: str allowed_vlans: list # list of VLAN IDs: [10, 20, 30] vlan_subnets: dict # {10: "192.168.10.0/24", 20: "192.168.20.0/24", ...} @app.get("/api/opnsense/wireguard/status") def opnsense_wg_status(): """Check OPNsense WireGuard plugin, server, and peer state.""" opn_cfg = _load_opnsense_cfg() if not opn_cfg: return {"opnsense_configured": False} wg = _load_opnsense_wg() # Probe for the WireGuard plugin — a 404 means the plugin isn't installed try: _opnsense_request(opn_cfg, "wireguard/server/searchServer") plugin_ok = True except ValueError as e: msg = str(e) # 404 → plugin absent; other errors → reachability / auth issue plugin_ok = False return { "opnsense_configured": True, "plugin_installed": False, "error": msg, "server": None, "peers": [], } # If we have a saved server UUID, fetch live info server_info = None if wg.get("server_uuid"): try: s = _opnsense_request(opn_cfg, f"wireguard/server/getServer/{wg['server_uuid']}") srv = s.get("server", {}) server_info = { "uuid": wg["server_uuid"], "name": srv.get("name", wg.get("server_name","")), "pubkey": srv.get("pubkey", wg.get("server_pubkey","")), "tunnel_ip": wg.get("server_tunnel_ip",""), "listen_port": wg.get("listen_port", 51820), "public_endpoint": wg.get("public_endpoint",""), } except Exception: # Server UUID no longer valid (e.g. OPNsense was reset) server_info = None # Merge OPNsense peer list with local metadata (which holds allowed_vlans) local_peers = {p["name"]: p for p in wg.get("peers", [])} opn_peers = [] try: resp = _opnsense_request(opn_cfg, "wireguard/client/searchClient") opn_peers = resp.get("rows", []) except Exception: pass merged = [] for p in opn_peers: name = p.get("name", "") loc = local_peers.get(name, {}) merged.append({ "uuid": p.get("uuid", ""), "name": name, "enabled": p.get("enabled", "0") == "1", "tunnel_ip": p.get("tunneladdress", ""), "allowed_vlans": loc.get("allowed_vlans", []), }) return { "opnsense_configured": True, "plugin_installed": plugin_ok, "server": server_info, "peers": merged, } @app.post("/api/opnsense/wireguard/setup-server") def opnsense_wg_setup_server(body: OPNWGServerSetup): """Create (or replace) a WireGuard server on OPNsense via its API.""" require_session(body.token) opn_cfg = _load_opnsense_cfg() if not opn_cfg: raise HTTPException(400, "OPNsense not configured — connect it in the DHCP tab first") import ipaddress as _ip, time as _time try: net = _ip.ip_network(body.tunnel_subnet, strict=False) except Exception: raise HTTPException(400, "Invalid tunnel_subnet — use CIDR notation e.g. 10.99.2.0/24") server_tunnel_ip = f"{list(net.hosts())[0]}/{net.prefixlen}" wg = _load_opnsense_wg() # Tear down any pre-existing server so we start clean if wg.get("server_uuid"): try: _opnsense_request(opn_cfg, f"wireguard/server/delServer/{wg['server_uuid']}", method="POST") except Exception: pass payload = { "server": { "enabled": "1", "name": body.server_name, "instance": "1", # creates wg1 — leaves wg0 (local) untouched "port": str(body.listen_port), "tunneladdress": server_tunnel_ip, "dns": "", "peers": "", } } try: result = _opnsense_request(opn_cfg, "wireguard/server/addServer", method="POST", body=payload) except Exception as e: raise HTTPException(500, f"OPNsense rejected server creation: {e}") server_uuid = result.get("uuid","") if not server_uuid: raise HTTPException(500, "OPNsense did not return a server UUID") # Apply so OPNsense generates the keypair, then read it back try: _opnsense_request(opn_cfg, "wireguard/service/reconfigure", method="POST") except Exception: pass _time.sleep(1.5) # give the daemon a moment to generate keys server_pubkey = "" try: s = _opnsense_request(opn_cfg, f"wireguard/server/getServer/{server_uuid}") server_pubkey = s.get("server", {}).get("pubkey", "") except Exception: pass wg = { "server_uuid": server_uuid, "server_name": body.server_name, "listen_port": body.listen_port, "tunnel_subnet": body.tunnel_subnet, "server_tunnel_ip": server_tunnel_ip, "server_pubkey": server_pubkey, "public_endpoint": body.public_endpoint, "peers": [], } _save_opnsense_wg(wg) log.info(f"OPNsense WG server created: {body.server_name} uuid={server_uuid}") return { "success": True, "server_uuid": server_uuid, "server_pubkey": server_pubkey, "server_tunnel_ip": server_tunnel_ip, "listen_port": body.listen_port, } @app.delete("/api/opnsense/wireguard/server") def opnsense_wg_delete_server(token: str): """Remove the WireGuard server from OPNsense and clear local state.""" require_session(token) opn_cfg = _load_opnsense_cfg() if not opn_cfg: raise HTTPException(400, "OPNsense not configured") wg = _load_opnsense_wg() if not wg.get("server_uuid"): raise HTTPException(404, "No server is configured") try: _opnsense_request(opn_cfg, f"wireguard/server/delServer/{wg['server_uuid']}", method="POST") _opnsense_request(opn_cfg, "wireguard/service/reconfigure", method="POST") except Exception as e: raise HTTPException(500, f"Failed to delete server: {e}") _save_opnsense_wg({}) return {"success": True} @app.post("/api/opnsense/wireguard/add-peer") def opnsense_wg_add_peer(body: OPNWGAddPeer): """ Generate a WireGuard keypair, register the peer on OPNsense, link it to the server, and return a ready-to-use .conf for the client. The private key is stored only on the management PC (never sent to OPNsense). """ require_session(body.token) opn_cfg = _load_opnsense_cfg() if not opn_cfg: raise HTTPException(400, "OPNsense not configured") wg = _load_opnsense_wg() if not wg.get("server_uuid"): raise HTTPException(400, "Set up the WireGuard server on OPNsense first") import ipaddress as _ip, re as _re # ── Allocate next free IP in the tunnel subnet ──────────────────── net = _ip.ip_network(wg["tunnel_subnet"], strict=False) hosts = list(net.hosts()) used = set() # Reserve the server's own tunnel IP m = _re.match(r'(\S+)/\d+', wg.get("server_tunnel_ip", "")) if m: used.add(m.group(1)) for p in wg.get("peers", []): m2 = _re.match(r'(\S+)/\d+', p.get("tunnel_ip", "")) if m2: used.add(m2.group(1)) peer_ip_obj = next((h for h in hosts if str(h) not in used), None) if not peer_ip_obj: raise HTTPException(400, "Tunnel subnet is full — no IPs available for new peer") peer_ip = f"{peer_ip_obj}/{net.prefixlen}" # ── Build the AllowedIPs list from chosen VLANs ─────────────────── vlan_cidrs = [] for vid in body.allowed_vlans: subnet = (body.vlan_subnets.get(str(vid)) or body.vlan_subnets.get(int(vid)) or f"192.168.{vid}.0/24") vlan_cidrs.append(subnet) # Always include the tunnel subnet so the client can reach the server allowed_ips = ", ".join([str(net)] + vlan_cidrs) if vlan_cidrs else str(net) # ── Generate keypair (private key stays on mgmt PC only) ───────── c_priv, c_pub = _wg_genkey_api() # ── Register peer (client) on OPNsense ─────────────────────────── peer_payload = { "client": { "enabled": "1", "name": body.name, "pubkey": c_pub, "psk": "", "tunneladdress": peer_ip, "serveraddress": "", "serverport": "", "keepalive": "25", } } try: result = _opnsense_request(opn_cfg, "wireguard/client/addClient", method="POST", body=peer_payload) except Exception as e: raise HTTPException(500, f"OPNsense rejected peer creation: {e}") peer_uuid = result.get("uuid", "") if not peer_uuid: raise HTTPException(500, "OPNsense did not return a peer UUID") # ── Link peer to server (append to server's peers list) ────────── try: s = _opnsense_request(opn_cfg, f"wireguard/server/getServer/{wg['server_uuid']}") srv = s.get("server", {}) existing = srv.get("peers", "") new_peers = f"{existing},{peer_uuid}" if existing else peer_uuid _opnsense_request(opn_cfg, f"wireguard/server/setServer/{wg['server_uuid']}", method="POST", body={"server": {**srv, "peers": new_peers}}) except Exception as e: log.warning(f"Could not link peer to server (peer still registered): {e}") # Apply config try: _opnsense_request(opn_cfg, "wireguard/service/reconfigure", method="POST") except Exception: pass # ── Build client .conf ──────────────────────────────────────────── server_pubkey = wg.get("server_pubkey", "") endpoint_host = wg.get("public_endpoint", "") or "" endpoint_port = wg.get("listen_port", 51820) tunnel_gw = wg.get("server_tunnel_ip", "").split("/")[0] client_conf = ( f"[Interface]\n" f"PrivateKey = {c_priv}\n" f"Address = {peer_ip}\n" f"DNS = {tunnel_gw}\n\n" f"[Peer]\n" f"PublicKey = {server_pubkey or ''}\n" f"Endpoint = {endpoint_host}:{endpoint_port}\n" f"AllowedIPs = {allowed_ips}\n" f"PersistentKeepalive = 25\n" ) # ── Persist peer metadata locally ──────────────────────────────── peer_meta = { "uuid": peer_uuid, "name": body.name, "pub_key": c_pub, "priv_key": c_priv, # NEVER sent to OPNsense "tunnel_ip": peer_ip, "allowed_vlans": body.allowed_vlans, "allowed_ips": allowed_ips, "config": client_conf, } peers = [p for p in wg.get("peers", []) if p.get("name") != body.name] peers.append(peer_meta) wg["peers"] = peers _save_opnsense_wg(wg) log.info(f"OPNsense WG peer added: {body.name} → {peer_ip} VLANs={body.allowed_vlans}") return { "success": True, "uuid": peer_uuid, "name": body.name, "tunnel_ip": peer_ip, "allowed_vlans": body.allowed_vlans, "config": client_conf, } @app.delete("/api/opnsense/wireguard/peer/{uuid}") def opnsense_wg_remove_peer(uuid: str, token: str): """Remove a peer from OPNsense and from local metadata.""" require_session(token) opn_cfg = _load_opnsense_cfg() if not opn_cfg: raise HTTPException(400, "OPNsense not configured") wg = _load_opnsense_wg() # Remove from OPNsense try: _opnsense_request(opn_cfg, f"wireguard/client/delClient/{uuid}", method="POST") except Exception as e: raise HTTPException(500, f"Failed to remove peer from OPNsense: {e}") # Unlink from server peers list if wg.get("server_uuid"): try: s = _opnsense_request(opn_cfg, f"wireguard/server/getServer/{wg['server_uuid']}") srv = s.get("server", {}) existing = srv.get("peers", "") updated = ",".join(p for p in existing.split(",") if p and p != uuid) _opnsense_request(opn_cfg, f"wireguard/server/setServer/{wg['server_uuid']}", method="POST", body={"server": {**srv, "peers": updated}}) except Exception: pass # Apply try: _opnsense_request(opn_cfg, "wireguard/service/reconfigure", method="POST") except Exception: pass wg["peers"] = [p for p in wg.get("peers", []) if p.get("uuid") != uuid] _save_opnsense_wg(wg) return {"success": True} @app.get("/api/opnsense/wireguard/peer-config/{name}") def opnsense_wg_peer_config(name: str): """Return the saved .conf text for a named peer (includes private key).""" wg = _load_opnsense_wg() peer = next((p for p in wg.get("peers", []) if p["name"] == name), None) if not peer: raise HTTPException(404, f"Peer '{name}' not found in local store") return {"name": name, "config": peer.get("config", "")}