Claude
12646d4184
Replace ERS 5952 (48+4 port) config with ERS 59100GTS-PWR+:
- Port validation extended to 1–100
- All interfaces now use GigabitEthernet 1/{p} slot notation
- PoE boundary moved from port 48 to port 96
- VLAN commands updated to use 1/{p} port notation
- Key path, TOTP name, and app title updated
https://claude.ai/code/session_01JR2EMK7rwrZJowpstcaxQ6
2788 lines
107 KiB
Python
2788 lines
107 KiB
Python
"""
|
||
ERS 59100GTS-PWR+ 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/ers59100_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="ERS59100", issuer_name="SwitchManager")
|
||
print("\n══════════════════════════════════════════════════")
|
||
print(" ERS 59100GTS-PWR+ 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–100 for the ERS 59100GTS-PWR+)."""
|
||
p = int(v)
|
||
if not 1 <= p <= 100:
|
||
raise ValueError("port: must be 1–100")
|
||
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+(GigabitEthernet\s+1/\d+|vlan\s+\d+)$'),
|
||
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 59100GTS-PWR+ CLI commands for a port configuration change.
|
||
|
||
Ports 1–96 are GigabitEthernet copper (PoE capable); ports 97–100 are SFP+ uplinks (no PoE).
|
||
All interfaces use slot/port notation: GigabitEthernet 1/{p}.
|
||
Returns a list of CLI command strings ready for push_one_by_one().
|
||
"""
|
||
p = cfg.port
|
||
iface = f"GigabitEthernet 1/{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} 1/{p}", f"vlan pvid 1/{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} 1/{p}", f"vlan tagging {ts} 1/{p}"]
|
||
cmds.append(f"vlan pvid 1/{p} {native}")
|
||
if p <= 96:
|
||
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 59100GTS-PWR+ 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 59100GTS-PWR+ 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 59100GTS-PWR+ '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 59100GTS-PWR+ 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 59100GTS-PWR+."""
|
||
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 59100GTS-PWR+ 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 59100GTS-PWR+ 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 59100GTS-PWR+ 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 "<YOUR-OPNSENSE-PUBLIC-IP>"
|
||
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 '<SERVER_PUBKEY>'}\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", "")}
|