diff --git a/switch_backend.py b/switch_backend.py index 58b6ec4..ee8201e 100644 --- a/switch_backend.py +++ b/switch_backend.py @@ -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 - ip dhcp-relay enable fwd-path - """ - 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 # ══════════════════════════════════════════════════════════════════════