1847 lines
70 KiB
Python
1847 lines
70 KiB
Python
"""
|
||
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:
|
||
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}$')
|
||
|
||
def _san(v: str, pat: re.Pattern, field: str) -> str:
|
||
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:
|
||
_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:
|
||
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 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():
|
||
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():
|
||
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():
|
||
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] = ""
|
||
@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]:
|
||
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]:
|
||
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}"
|
||
port_str = f" eq {r.port}" if r.port else ""
|
||
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():
|
||
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():
|
||
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):
|
||
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):
|
||
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):
|
||
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):
|
||
require_session(body.token)
|
||
return push_one_by_one(build_port(body))
|
||
|
||
@app.post("/api/switch/acl")
|
||
def create_acl(body: AclCreate):
|
||
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:
|
||
if DEVICES_FILE.exists():
|
||
try: return _json.loads(DEVICES_FILE.read_text())
|
||
except: pass
|
||
return []
|
||
|
||
def _save_devices(devices: list):
|
||
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):
|
||
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):
|
||
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():
|
||
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:
|
||
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:
|
||
if OPNSENSE_FILE.exists():
|
||
try: return _json.loads(OPNSENSE_FILE.read_text())
|
||
except: pass
|
||
return {}
|
||
|
||
def _save_opnsense_cfg(cfg: dict):
|
||
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 _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"
|
||
|
||
# ── 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)
|
||
|
||
# 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,
|
||
},
|
||
"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}")
|
||
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
# CONTROL D / ctrld DNS MANAGEMENT
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
|
||
CTRLD_FILE = _Path("/etc/switch-manager/ctrld.json")
|
||
CTRLD_BIN = _Path("/usr/local/bin/ctrld")
|
||
|
||
def _load_ctrld_cfg() -> dict:
|
||
if CTRLD_FILE.exists():
|
||
try: return _json.loads(CTRLD_FILE.read_text())
|
||
except: pass
|
||
return {}
|
||
|
||
def _save_ctrld_cfg(cfg: dict):
|
||
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) -> str:
|
||
"""
|
||
Build a ctrld.toml config for per-VLAN DNS filtering.
|
||
Each VLAN gets its own upstream pointing to its Control D Resolver ID.
|
||
Source IP routing directs traffic to the correct profile automatically.
|
||
|
||
vlan_profiles: list of { vlan_id, name, subnet, resolver_id }
|
||
"""
|
||
lines = [
|
||
"# ctrld configuration — generated by Avaya 5952 Switch Manager",
|
||
"# https://github.com/Control-D-Inc/ctrld",
|
||
"",
|
||
"[service]",
|
||
'name = "ctrld"',
|
||
"",
|
||
"# Single listener on port 53 — all VLANs send DNS here",
|
||
"[[listener]]",
|
||
'ip = "0.0.0.0"',
|
||
"port = 53",
|
||
'tag = "all-vlans"',
|
||
"",
|
||
]
|
||
|
||
# One upstream per VLAN
|
||
for vp in vlan_profiles:
|
||
rid = vp.get("resolver_id","").strip()
|
||
if not rid:
|
||
continue
|
||
tag = f"vlan{vp['vlan_id']}"
|
||
lines += [
|
||
f"# VLAN {vp['vlan_id']} — {vp['name']}",
|
||
f"[[upstream]]",
|
||
f'id = "{tag}"',
|
||
f'type = "doh3"',
|
||
f'endpoint = "https://dns.controld.com/{rid}"',
|
||
f'tag = "{tag}"',
|
||
"",
|
||
]
|
||
|
||
# Routing rules — match source subnet to upstream
|
||
lines += ["# Route each VLAN subnet to its profile"]
|
||
for vp in vlan_profiles:
|
||
rid = vp.get("resolver_id","").strip()
|
||
if not rid:
|
||
continue
|
||
subnet = vp.get("subnet", f"192.168.{vp['vlan_id']}.0/24")
|
||
tag = f"vlan{vp['vlan_id']}"
|
||
lines += [
|
||
f"[[rule]]",
|
||
f'listener = "all-vlans"',
|
||
f'source_ip = "{subnet}"',
|
||
f'upstream = "{tag}"',
|
||
"",
|
||
]
|
||
|
||
# Fallback upstream (first valid profile or safe default)
|
||
first_valid = next((vp for vp in vlan_profiles if vp.get("resolver_id")), None)
|
||
if first_valid:
|
||
lines += [
|
||
"# Fallback for unmatched source IPs",
|
||
"[[upstream]]",
|
||
f'id = "fallback"',
|
||
f'type = "doh3"',
|
||
f'endpoint = "https://dns.controld.com/{first_valid["resolver_id"]}"',
|
||
f'tag = "fallback"',
|
||
"",
|
||
]
|
||
|
||
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 _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"}
|
||
|
||
# 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,
|
||
}
|
||
|
||
# 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",
|
||
}
|
||
|
||
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):
|
||
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}
|