diff --git a/switch_backend.py b/switch_backend.py index 4db8e0d..58b6ec4 100644 --- a/switch_backend.py +++ b/switch_backend.py @@ -781,6 +781,95 @@ def running_config(): out = read_cmd("show config") return {"config": out, "lines": len(out.splitlines())} + +# ── Capability probe ──────────────────────────────────────────────────── + +_caps_cache: dict = {} +_caps_ts: float = 0.0 +_caps_lock = threading.Lock() +_CAPS_TTL = 300 # seconds — re-probe every 5 min; license won't change mid-session + + +def _probe_capabilities() -> dict: + """ + Non-destructive read-only probes to detect which features the switch + supports under its current software license. + + Base Software: ACL and L3 VLAN commands return '% Invalid input detected'. + Advanced License: commands succeed (may show empty output, but no error). + """ + acl_out = read_cmd("show ip access-list") + vlan_out = read_cmd("show interface vlan 1") + acl_ok = not _SWITCH_ERR.search(acl_out) + l3_ok = not _SWITCH_ERR.search(vlan_out) + return { + "acl": acl_ok, + "l3_vlan": l3_ok, + "dhcp_relay_config": l3_ok, # relay config uses 'interface vlan' + "management_pinholes": acl_ok, + "dns_enforce_acls": acl_ok, + "license_tier": "advanced" if acl_ok else "base", + } + + +def _require_advanced_license(): + """Raise 402 if the switch reports Base Software (no ACL/L3 support).""" + global _caps_ts + with _caps_lock: + cached = _caps_cache.copy() if _caps_cache else {} + # If we have a cached result use it; otherwise probe now + if not cached: + try: + cached = _probe_capabilities() + with _caps_lock: + _caps_cache.update(cached) + _caps_ts = time.time() + except HTTPException: + return # can't reach switch — let the push fail with its own error + if cached.get("license_tier") == "base": + raise HTTPException( + 402, + "This feature requires the Advanced Software License. " + "The switch reported Base Software — ACLs and L3 VLAN interfaces are not available." + ) + + +@app.get("/api/switch/capabilities") +def switch_capabilities(): + """ + Probe the switch to determine which features are available under its + 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. + + 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: + if time.time() - _caps_ts < _CAPS_TTL and _caps_cache: + return {**_caps_cache, "cached": True} + try: + caps = _probe_capabilities() + except HTTPException as e: + return { + "error": e.detail, + "acl": False, "l3_vlan": False, + "dhcp_relay_config": False, + "management_pinholes": False, + "dns_enforce_acls": False, + "license_tier": "unknown", + "cached": False, + } + with _caps_lock: + _caps_cache = caps + _caps_ts = time.time() + return {**caps, "cached": False} + # ── Danger pre-flight (no auth — check before prompting TOTP) ───────── @app.post("/api/check/danger") @@ -851,8 +940,9 @@ def configure_port(body: PortConfig): @app.post("/api/switch/acl") def create_acl(body: AclCreate): - """Create an extended IP ACL and apply it to a VLAN interface.""" + """Create an extended IP ACL and apply it to a VLAN interface. Requires Advanced License.""" require_session(body.token) + _require_advanced_license() return push_one_by_one(build_acl(body)) # ── Serve React app ──────────────────────────────────────────────────── @@ -1034,8 +1124,9 @@ def push_reservation(body: DeviceUpdate): @app.post("/api/devices/push-pinhole") def push_pinhole(body: PinholeRequest): - """Add or remove an ACL pinhole for a device to reach management.""" + """Add or remove an ACL pinhole for a device to reach management. Requires Advanced License.""" require_session(body.token) + _require_advanced_license() devices = _load_devices() device = next((DeviceEntry(**d) for d in devices if d["mac"] == body.mac), None) if not device: @@ -1682,9 +1773,10 @@ 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. + 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)") @@ -2217,6 +2309,7 @@ def ctrld_dns_enforce_acls(body: DnsEnforceRequest): it does NOT push anything to the switch itself. """ require_session(body.token) + _require_advanced_license() # Refuse to touch VLAN 99 (management) — a broken ACL there = lockout safe_vlans = [v for v in body.vlan_ids if v != 99]