Add GET /api/switch/capabilities endpoint with license detection
Probes the switch using read-only show commands to detect whether the Advanced Software License is installed. Base Software rejects ACL and L3 VLAN interface commands with 'Invalid input detected'. - GET /api/switch/capabilities: non-destructive probe (show ip access-list, show interface vlan 1), returns acl/l3_vlan/dhcp_relay_config/ management_pinholes/dns_enforce_acls flags and license_tier. Cached 5 min. - _require_advanced_license(): guard helper that raises HTTP 402 with a clear message before attempting any ACL push to the switch. - Applied guard to: POST /api/switch/acl, /api/devices/push-pinhole, /api/dhcp/relay/configure, /api/ctrld/dns-enforce-acls. https://claude.ai/code/session_01JR2EMK7rwrZJowpstcaxQ6
This commit is contained in:
+96
-3
@@ -781,6 +781,95 @@ def running_config():
|
|||||||
out = read_cmd("show config")
|
out = read_cmd("show config")
|
||||||
return {"config": out, "lines": len(out.splitlines())}
|
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) ─────────
|
# ── Danger pre-flight (no auth — check before prompting TOTP) ─────────
|
||||||
|
|
||||||
@app.post("/api/check/danger")
|
@app.post("/api/check/danger")
|
||||||
@@ -851,8 +940,9 @@ def configure_port(body: PortConfig):
|
|||||||
|
|
||||||
@app.post("/api/switch/acl")
|
@app.post("/api/switch/acl")
|
||||||
def create_acl(body: AclCreate):
|
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_session(body.token)
|
||||||
|
_require_advanced_license()
|
||||||
return push_one_by_one(build_acl(body))
|
return push_one_by_one(build_acl(body))
|
||||||
|
|
||||||
# ── Serve React app ────────────────────────────────────────────────────
|
# ── Serve React app ────────────────────────────────────────────────────
|
||||||
@@ -1034,8 +1124,9 @@ def push_reservation(body: DeviceUpdate):
|
|||||||
|
|
||||||
@app.post("/api/devices/push-pinhole")
|
@app.post("/api/devices/push-pinhole")
|
||||||
def push_pinhole(body: PinholeRequest):
|
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_session(body.token)
|
||||||
|
_require_advanced_license()
|
||||||
devices = _load_devices()
|
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:
|
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
|
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
|
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_session(body.token)
|
||||||
|
_require_advanced_license()
|
||||||
safe_vlans = [int(v) for v in body.vlans if int(v) != 99]
|
safe_vlans = [int(v) for v in body.vlans if int(v) != 99]
|
||||||
if not safe_vlans:
|
if not safe_vlans:
|
||||||
raise HTTPException(400, "No VLANs to configure (VLAN 99 is excluded)")
|
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.
|
it does NOT push anything to the switch itself.
|
||||||
"""
|
"""
|
||||||
require_session(body.token)
|
require_session(body.token)
|
||||||
|
_require_advanced_license()
|
||||||
|
|
||||||
# Refuse to touch VLAN 99 (management) — a broken ACL there = lockout
|
# Refuse to touch VLAN 99 (management) — a broken ACL there = lockout
|
||||||
safe_vlans = [v for v in body.vlan_ids if v != 99]
|
safe_vlans = [v for v in body.vlan_ids if v != 99]
|
||||||
|
|||||||
Reference in New Issue
Block a user