Fix CLI commands to match ERS 59100GTS-PWR+ actual ACLI syntax
- show ip route default → show ip route (parse 0.0.0.0 row for gateway) - show ip helper-address → show ip dhcp-relay fwd-path + update parser - ip helper-address → ip dhcp-relay fwd-path <vlan-ip> <server-ip> (add _get_vlan_ips() to resolve VLAN interface IPs before building cmds) - Remove all show dhcp-server / show dhcp-server leases / show dhcp-server static-binding calls — switch has no DHCP server (show ip dhcp ? only shows 'client'). Device discovery now uses show arp only. - push-reservation, sync to_switch/remove_switch → 501 Not Implemented - _get_switch_reservations() / _get_switch_dhcp_status() return empty/false https://claude.ai/code/session_01JR2EMK7rwrZJowpstcaxQ6
This commit is contained in:
Claude
+67
-115
@@ -952,13 +952,8 @@ class PinholeRequest(BaseModel):
|
||||
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}\"",
|
||||
]
|
||||
"""ERS 59100GTS-PWR+ has no built-in DHCP server; reservations go to OPNsense."""
|
||||
raise NotImplementedError("Switch has no DHCP server — use OPNsense for reservations")
|
||||
|
||||
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."""
|
||||
@@ -977,21 +972,14 @@ def _build_pinhole_acl_cmds(device: DeviceEntry, mgmt_ip: str, allow: bool) -> l
|
||||
|
||||
@app.get("/api/devices")
|
||||
def get_devices():
|
||||
"""Return saved device list plus live DHCP leases from switch."""
|
||||
"""Return saved device list plus ARP table 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)
|
||||
arp_raw = read_cmd("show arp")
|
||||
live_leases = _parse_arp_table(arp_raw)
|
||||
except Exception as e:
|
||||
log.warning(f"Could not pull DHCP/ARP from switch: {e}")
|
||||
log.warning(f"Could not pull ARP from switch: {e}")
|
||||
return {
|
||||
"saved": saved,
|
||||
"live": live_leases,
|
||||
@@ -1023,14 +1011,8 @@ 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."""
|
||||
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)
|
||||
"""The switch has no DHCP server — push reservations to OPNsense instead."""
|
||||
raise HTTPException(501, "Switch has no DHCP server; use /api/dhcp/sync with direction=to_opnsense")
|
||||
|
||||
@app.post("/api/devices/push-pinhole")
|
||||
def push_pinhole(body: PinholeRequest):
|
||||
@@ -1338,50 +1320,23 @@ def _get_opnsense_leases(cfg: dict) -> list:
|
||||
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 []
|
||||
"""ERS 59100GTS-PWR+ has no DHCP server — always returns empty."""
|
||||
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": []}
|
||||
"""ERS 59100GTS-PWR+ has no DHCP server — always returns not running."""
|
||||
return {"running": False, "vlans": []}
|
||||
|
||||
def _get_relay_status() -> dict:
|
||||
"""Read current DHCP relay (ip helper-address) config from each VLAN interface."""
|
||||
"""Read current DHCP relay config from each VLAN interface."""
|
||||
import re as _re
|
||||
try:
|
||||
raw = read_cmd("show ip helper-address")
|
||||
raw = read_cmd("show ip dhcp-relay fwd-path")
|
||||
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)
|
||||
# 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}
|
||||
@@ -1389,13 +1344,45 @@ def _get_relay_status() -> dict:
|
||||
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 ip helper-address on the specified VLANs."""
|
||||
"""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"interface vlan {vid}",
|
||||
f" ip helper-address {opnsense_ip}",
|
||||
f"ip dhcp-relay fwd-path {vip} {opnsense_ip}",
|
||||
f"ip dhcp-relay enable fwd-path {vip} {opnsense_ip}",
|
||||
]
|
||||
return cmds
|
||||
|
||||
@@ -1463,21 +1450,12 @@ def dhcp_overview():
|
||||
switch_leases = []
|
||||
switch_status = _get_switch_dhcp_status()
|
||||
|
||||
# Also pull ARP for discovery
|
||||
# Discover devices via ARP (switch has no DHCP server)
|
||||
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
|
||||
switch_leases = _parse_arp_table(arp_raw)
|
||||
except Exception as e:
|
||||
log.warning(f"Switch lease fetch failed: {e}")
|
||||
log.warning(f"Switch ARP fetch failed: {e}")
|
||||
|
||||
cfg = _load_opnsense_cfg()
|
||||
opnsense_res = _get_opnsense_reservations(cfg) if cfg.get("key") else []
|
||||
@@ -1512,8 +1490,9 @@ 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)
|
||||
route = read_cmd("show ip route")
|
||||
# Look for default route: DST=0.0.0.0, MASK=0.0.0.0 — NEXT column is gateway
|
||||
m = _re.search(r'^0\.0\.0\.0\s+0\.0\.0\.0\s+(\d+\.\d+\.\d+\.\d+)', route, _re.MULTILINE)
|
||||
gateway = m.group(1) if m else None
|
||||
except Exception:
|
||||
gateway = None
|
||||
@@ -1578,48 +1557,21 @@ def push_reservation_to_opnsense(body: OPNsenseReservationPush):
|
||||
@app.post("/api/dhcp/sync")
|
||||
def sync_reservation(body: SyncRequest):
|
||||
"""
|
||||
Sync a reservation between switch and OPNsense.
|
||||
Sync a reservation in OPNsense (the switch has no DHCP server).
|
||||
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
|
||||
to_switch — not supported (switch has no DHCP server)
|
||||
to_opnsense — not supported (switch has no DHCP reservations)
|
||||
remove_switch — not supported (switch has no DHCP server)
|
||||
remove_opnsense — remove from OPNsense
|
||||
"""
|
||||
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']}"])
|
||||
if body.direction in ("to_switch", "to_opnsense", "remove_switch"):
|
||||
raise HTTPException(501, "Switch has no DHCP server; manage reservations directly in OPNsense")
|
||||
|
||||
elif body.direction == "remove_opnsense":
|
||||
if not ops_res or not ops_res.get("uuid"):
|
||||
@@ -1634,13 +1586,13 @@ def sync_reservation(body: SyncRequest):
|
||||
|
||||
@app.get("/api/dhcp/relay/status")
|
||||
def relay_status_endpoint():
|
||||
"""Return current ip helper-address config from the switch per VLAN."""
|
||||
"""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 helper-address 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
|
||||
as the management / recovery path.
|
||||
"""
|
||||
|
||||
Reference in New Issue
Block a user