root/avaya_switch_management
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Bridge Base Software gaps via OPNsense API

Browse Source 
- Add POST /api/vlan/provision: end-to-end VLAN wizard that creates the
  switch VLAN, OPNsense VLAN tag, DHCP scope, and allow-outbound firewall
  rule in one call; returns pending_steps for anything needing manual
  OPNsense UI finish (interface assignment when opnsense_if not provided)

- Rewire POST /api/devices/push-reservation to target OPNsense DHCP when
  configured (no Advanced License required); falls back to switch CLI only
  if OPNsense is not set up; uses stored VLAN→interface map for iface lookup

- Rewrite POST /api/devices/push-pinhole to use OPNsense firewall/filter
  API instead of switch ACLs; stores rule UUIDs in pinholes.json for clean
  removal; no longer requires Advanced License

- Remove dead relay endpoints (GET/POST /api/dhcp/relay/*), RelayConfig
  model, and helpers (_get_relay_status, _get_vlan_ips, _build_relay_cmds);
  relay config is irrelevant when OPNsense is the DHCP server

- Add VLAN_IF_MAP_FILE and PINHOLE_FILE with load/save helpers to persist
  the VLAN→OPNsense interface mapping and pinhole rule UUIDs across restarts

https://claude.ai/code/session_01JR2EMK7rwrZJowpstcaxQ6
This commit is contained in:
Claude
2026-03-24 14:27:21 +00:00
parent 3f05be1f1d
commit c5d3383abe
1 changed files with 270 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files
switch_backend.py
+270 -108
View File
@@ -561,6 +561,27 @@ class VlanDelete(BaseModel):
if vid == 1: raise ValueError("Cannot delete VLAN 1")
return vid
class VlanProvision(BaseModel):
token: str
vlan_id: int
name: str
subnet: str # e.g. "192.168.20.0/24"
gateway: str # OPNsense IP on this VLAN, e.g. "192.168.20.1"
dhcp_start: str # e.g. "192.168.20.100"
dhcp_end: str # e.g. "192.168.20.200"
parent_if: str # OPNsense physical parent, e.g. "em0" or "igb0"
opnsense_if: Optional[str] = "" # assigned interface name, e.g. "opt2"
allow_internet: bool = True # add default allow-out firewall rule
@field_validator("vlan_id")
@classmethod
def cv(cls, v):
vid = san_vid(v)
if vid in (1, 99): raise ValueError("VLAN 1 and 99 are reserved")
return vid
@field_validator("name")
@classmethod
def cn(cls, v): return _san(v, _RE_VNAME, "name")
class AclRule(BaseModel):
action: str
proto: str
@@ -841,13 +862,11 @@ def switch_capabilities():
current software license. Results are cached for 5 minutes.
Base Software supports L2 only (VLANs, ports, PoE, show commands).
Advanced License adds ACLs, L3 VLAN interfaces, and DHCP relay config.
Advanced License adds ACLs and L3 VLAN interfaces.
Affected endpoints when license_tier == 'base':
- POST /api/switch/acl (acl)
- POST /api/devices/push-pinhole (management_pinholes)
- POST /api/ctrld/dns-enforce-acls (dns_enforce_acls)
- POST /api/dhcp/relay/configure (dhcp_relay_config)
"""
global _caps_cache, _caps_ts
with _caps_lock:
@@ -945,6 +964,137 @@ def create_acl(body: AclCreate):
_require_advanced_license()
return push_one_by_one(build_acl(body))
@app.post("/api/vlan/provision")
def provision_vlan(body: VlanProvision):
"""
End-to-end VLAN provisioning: switch VLAN + OPNsense interface tag +
DHCP scope + optional internet-allow firewall rule.
Steps performed:
1. Create VLAN on switch
2. Create VLAN tag on OPNsense (interfaces/vlan_settings)
3. Apply OPNsense VLAN config
4. If opnsense_if provided: create DHCP subnet + apply
5. If opnsense_if + allow_internet: add allow-outbound firewall rule + apply
Returns steps_done, pending_steps (anything needing manual finish in OPNsense UI).
"""
import ipaddress as _ipaddr
require_session(body.token)
steps_done: list[str] = []
pending_steps: list[str] = []
# Validate subnet/gateway/range are sane
try:
net = _ipaddr.ip_network(body.subnet, strict=False)
_ipaddr.ip_address(body.gateway)
_ipaddr.ip_address(body.dhcp_start)
_ipaddr.ip_address(body.dhcp_end)
except ValueError as e:
raise HTTPException(400, f"Invalid address: {e}")
# ── Step 1: Switch VLAN ────────────────────────────────────────────
result = push_one_by_one([f'vlan create {body.vlan_id} name "{body.name}" type port'])
if not result.get("success"):
raise HTTPException(502, {"message": "Switch VLAN create failed", "detail": result})
steps_done.append(f"switch: vlan {body.vlan_id} '{body.name}' created")
# ── Steps 2–5: OPNsense ───────────────────────────────────────────
cfg = _load_opnsense_cfg()
if not cfg.get("key"):
pending_steps += [
f"OPNsense: create VLAN tag {body.vlan_id} on {body.parent_if}",
f"OPNsense: assign VLAN interface, set IP {body.gateway}/{net.prefixlen}",
f"OPNsense: create DHCP scope {body.dhcp_start}–{body.dhcp_end}",
]
if body.allow_internet:
pending_steps.append("OPNsense: add allow-outbound firewall rule for VLAN")
return {"success": True, "steps_done": steps_done, "pending_steps": pending_steps,
"note": "OPNsense not configured — connect it under DHCP settings to automate these steps"}
errors: list[str] = []
# Step 2: create VLAN tag
try:
vlan_r = _opnsense_request(cfg, "interfaces/vlan_settings/addItem", "POST", {
"vlan": {"if": body.parent_if, "tag": str(body.vlan_id), "pcp": "0", "descr": body.name}
})
vlan_uuid = vlan_r.get("uuid", "")
steps_done.append(f"OPNsense: VLAN tag {body.vlan_id} created on {body.parent_if} (uuid={vlan_uuid})")
except ValueError as e:
errors.append(f"OPNsense VLAN tag: {e}")
vlan_uuid = ""
# Step 3: apply VLAN config
if vlan_uuid:
try:
_opnsense_request(cfg, "interfaces/vlan_settings/reconfigure", "POST")
steps_done.append("OPNsense: VLAN config applied")
except ValueError as e:
errors.append(f"OPNsense VLAN apply: {e}")
# Interface assignment must be done in OPNsense UI unless opnsense_if is provided
if not body.opnsense_if:
pending_steps += [
f"OPNsense UI: assign {body.parent_if}.{body.vlan_id} as a new interface, "
f"set static IP {body.gateway}/{net.prefixlen}, note the interface name (e.g. opt2)",
f"OPNsense: create DHCP scope {body.dhcp_start}–{body.dhcp_end} once interface is assigned",
]
if body.allow_internet:
pending_steps.append("OPNsense: add allow-outbound firewall rule for new interface")
else:
# Step 4: DHCP scope
try:
_opnsense_request(cfg, "dhcpv4/settings/addSubnet", "POST", {
"subnet": {
"interface": body.opnsense_if,
"subnet": str(net),
"gateway": body.gateway,
"dns_servers": body.gateway,
"range": {"from": body.dhcp_start, "to": body.dhcp_end},
}
})
_opnsense_request(cfg, "dhcpv4/service/reconfigure", "POST")
steps_done.append(
f"OPNsense: DHCP scope {body.dhcp_start}–{body.dhcp_end} on {body.opnsense_if} created")
except ValueError as e:
errors.append(f"OPNsense DHCP scope: {e}")
# Step 5: firewall allow-outbound
if body.allow_internet:
try:
fw_r = _opnsense_request(cfg, "firewall/filter/addRule", "POST", {
"rule": {
"enabled": "1",
"action": "pass",
"interface": body.opnsense_if,
"direction": "in",
"ipprotocol": "inet",
"protocol": "any",
"source": {"network": f"{body.opnsense_if}net"},
"destination":{"any": "1"},
"descr": f"Allow VLAN {body.vlan_id} {body.name} outbound",
}
})
_opnsense_request(cfg, "firewall/filter/apply", "POST")
steps_done.append(
f"OPNsense: allow-outbound rule for {body.opnsense_if} added (uuid={fw_r.get('uuid','')})")
except ValueError as e:
errors.append(f"OPNsense firewall rule: {e}")
# Persist VLAN→interface mapping for push-reservation lookups
vmap = _load_vlan_if_map()
vmap[str(body.vlan_id)] = body.opnsense_if
_save_vlan_if_map(vmap)
return {
"success": len(errors) == 0,
"steps_done": steps_done,
"pending_steps": pending_steps,
"errors": errors,
}
# ── Serve React app ────────────────────────────────────────────────────
if os.path.isdir(STATIC_DIR):
app.mount("/", StaticFiles(directory=STATIC_DIR, html=True), name="frontend")
@@ -1113,32 +1263,111 @@ def delete_device(body: DeviceDelete):
@app.post("/api/devices/push-reservation")
def push_reservation(body: DeviceUpdate):
"""Push a DHCP static binding for this device to the switch."""
"""
Push a DHCP static reservation for this device.
Routes to OPNsense if configured (preferred — no license required).
Falls back to switch DHCP CLI only if OPNsense is not configured, which
requires Advanced License on 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)
device = body.device
cfg = _load_opnsense_cfg()
if cfg.get("key"):
# Derive OPNsense interface from stored VLAN→interface map
vmap = _load_vlan_if_map()
iface = vmap.get(str(device.vlan), "")
try:
result = _opnsense_request(cfg, "dhcpv4/reservations/addReservation", "POST", {
"reservation": {
"interface": iface,
"mac": device.mac,
"ipaddr": device.ip,
"hostname": device.name,
"descr": f"Added by switch-manager",
}
})
_opnsense_request(cfg, "dhcpv4/service/reconfigure", "POST")
log.info(f"OPNsense DHCP reservation pushed: {device.name} ({device.mac}) → {device.ip}")
return {"success": True, "target": "opnsense", "result": result}
except ValueError as e:
raise HTTPException(400, str(e))
else:
# Switch DHCP — requires Advanced License
cmds = _build_dhcp_reservation_cmds(device)
danger = check_danger(cmds)
if danger["has_hard_block"]:
raise HTTPException(400, {"message": "Blocked", "blocked": danger["hard_blocked"]})
log.info(f"Switch DHCP reservation pushed: {device.name}")
result = push_one_by_one(cmds)
result["target"] = "switch"
return result
@app.post("/api/devices/push-pinhole")
def push_pinhole(body: PinholeRequest):
"""Add or remove an ACL pinhole for a device to reach management. Requires Advanced License."""
"""
Add or remove a management-access firewall pinhole for a device.
Uses OPNsense firewall API if configured. Rule UUIDs are stored locally
so the same device can be cleanly de-pinholed later.
"""
require_session(body.token)
_require_advanced_license()
cfg = _load_opnsense_cfg()
if not cfg.get("key"):
raise HTTPException(503, "OPNsense not configured — connect it under DHCP settings")
devices = _load_devices()
device = next((DeviceEntry(**d) for d in devices if d["mac"] == body.mac), None)
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)
mgmt_ip = SWITCH_HOST.rsplit('.', 1)[0] + '.50'
pinholes = _load_pinholes()
if body.allow:
# Look up OPNsense interface for device VLAN
vmap = _load_vlan_if_map()
iface = vmap.get(str(device.vlan), "")
try:
r = _opnsense_request(cfg, "firewall/filter/addRule", "POST", {
"rule": {
"enabled": "1",
"action": "pass",
"interface": iface,
"direction": "in",
"ipprotocol": "inet",
"protocol": "tcp",
"source": {"address": device.ip},
"destination": {"address": mgmt_ip, "port": "8765"},
"descr": f"switch-manager pinhole {device.name}",
}
})
_opnsense_request(cfg, "firewall/filter/apply", "POST")
pinholes[device.mac] = r.get("uuid", "")
_save_pinholes(pinholes)
log.info(f"Pinhole allow: {device.name} ({device.ip}) → {mgmt_ip}:8765")
return {"success": True, "action": "allow", "uuid": r.get("uuid", "")}
except ValueError as e:
raise HTTPException(400, str(e))
else:
uuid = pinholes.get(device.mac, "")
if not uuid:
raise HTTPException(404, "No pinhole rule found for this device")
try:
_opnsense_request(cfg, f"firewall/filter/delRule/{uuid}", "POST")
_opnsense_request(cfg, "firewall/filter/apply", "POST")
pinholes.pop(device.mac, None)
_save_pinholes(pinholes)
log.info(f"Pinhole removed: {device.name} ({device.ip})")
return {"success": True, "action": "deny", "uuid": uuid}
except ValueError as e:
raise HTTPException(400, str(e))
# ══════════════════════════════════════════════════════════════════════
# WIREGUARD PEER MANAGEMENT
@@ -1334,7 +1563,30 @@ import urllib.error as _urlerr
import ssl as _ssl
import base64 as _b64
OPNSENSE_FILE = _Path("/etc/switch-manager/opnsense.json")
OPNSENSE_FILE = _Path("/etc/switch-manager/opnsense.json")
VLAN_IF_MAP_FILE = _Path("/etc/switch-manager/vlan-if-map.json")
PINHOLE_FILE = _Path("/etc/switch-manager/pinholes.json")
def _load_vlan_if_map() -> dict:
"""Return {vlan_id_str: opnsense_if_name} e.g. {"20": "opt2"}."""
if VLAN_IF_MAP_FILE.exists():
try: return _json.loads(VLAN_IF_MAP_FILE.read_text())
except: pass
return {}
def _save_vlan_if_map(m: dict):
VLAN_IF_MAP_FILE.write_text(_json.dumps(m, indent=2))
def _load_pinholes() -> dict:
"""Return {mac: rule_uuid} for OPNsense firewall pinholes."""
if PINHOLE_FILE.exists():
try: return _json.loads(PINHOLE_FILE.read_text())
except: pass
return {}
def _save_pinholes(m: dict):
PINHOLE_FILE.write_text(_json.dumps(m, indent=2))
PINHOLE_FILE.chmod(0o600)
def _load_opnsense_cfg() -> dict:
"""Load saved OPNsense API credentials from opnsense.json, returning {} if absent."""
@@ -1471,65 +1723,6 @@ def _get_switch_dhcp_status() -> dict:
except Exception:
return {"running": False, "vlans": []}
def _get_relay_status() -> dict:
"""Read current DHCP relay config from each VLAN interface."""
import re as _re
try:
raw = read_cmd("show ip dhcp-relay fwd-path")
configured = {}
for line in raw.splitlines():
# Output columns: VLAN INTERFACE SERVER ENABLE MODE
# e.g. " 10 192.168.10.1 192.168.99.1 enable bootp_dhcp"
m = _re.match(r'\s*(\d+)\s+\S+\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 _get_vlan_ips() -> dict:
"""Return {vlan_id: ip_address} for all VLAN interfaces on the switch.
Parses 'show interfaces vlan' output. Expected format (ERS 59100GTS-PWR+):
Vlan Interface-IP Mask Name
---- ----------------- ----------------- -------
10 192.168.10.1 255.255.255.0 Users
# TODO: confirm exact output by running 'show interfaces vlan' on switch
"""
import re as _re
try:
raw = read_cmd("show interfaces vlan")
result = {}
for line in raw.splitlines():
m = _re.match(r'\s*(\d+)\s+(\d+\.\d+\.\d+\.\d+)', line)
if m:
result[int(m.group(1))] = m.group(2)
return result
except Exception as e:
log.warning(f"VLAN IP fetch failed: {e}")
return {}
def _build_relay_cmds(opnsense_ip: str, vlan_ids: list) -> list:
"""Generate ERS 59100GTS-PWR+ CLI to set DHCP relay on the specified VLANs.
ERS syntax: ip dhcp-relay fwd-path <vlan-interface-ip> <server-ip>
ip dhcp-relay enable fwd-path <vlan-interface-ip> <server-ip>
"""
vlan_ips = _get_vlan_ips()
cmds = []
for vid in vlan_ids:
vip = vlan_ips.get(vid)
if not vip:
log.warning(f"No interface IP found for VLAN {vid}, skipping relay config")
continue
cmds += [
f"ip dhcp-relay fwd-path {vip} {opnsense_ip}",
f"ip dhcp-relay enable fwd-path {vip} {opnsense_ip}",
]
return cmds
def _find_conflicts(switch_res: list, opnsense_res: list) -> list:
"""
Find same MAC in both switch and OPNsense.
@@ -1571,13 +1764,6 @@ class SyncRequest(BaseModel):
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 ─────────────────────────────────────────────────────────
@@ -1613,10 +1799,7 @@ def dhcp_overview():
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 {
@@ -1632,7 +1815,6 @@ def dhcp_overview():
"leases": opnsense_leases,
"interfaces": opnsense_ifaces,
},
"relay": relay_status,
"conflicts": conflicts,
"has_conflicts": len(conflicts) > 0,
}
@@ -1763,26 +1945,6 @@ def sync_reservation(body: SyncRequest):
raise HTTPException(400, f"Unknown direction: {body.direction}")
@app.get("/api/dhcp/relay/status")
def relay_status_endpoint():
"""Return current DHCP relay (ip dhcp-relay fwd-path) config from the switch per VLAN."""
return _get_relay_status()
@app.post("/api/dhcp/relay/configure")
def configure_relay(body: RelayConfig):
"""
Push ip dhcp-relay fwd-path 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. Requires Advanced License.
"""
require_session(body.token)
_require_advanced_license()
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
# ══════════════════════════════════════════════════════════════════════