Fix DHCP server commands to use correct 'show ip dhcp-server' syntax
The original commands were missing the 'ip' prefix. Correct ACLI syntax:
show dhcp-server -> show ip dhcp-server
show dhcp-server leases -> show ip dhcp-server leases
show dhcp-server static-binding -> show ip dhcp-server static-binding
The ERS 59100GTS-PWR+ has a DHCP server but it may need to be enabled
first ('ip dhcp-server enable' in config mode) or may require an
Advanced License. All DHCP server calls now have try/except so device
discovery falls back to ARP if the feature is not yet active.
https://claude.ai/code/session_01JR2EMK7rwrZJowpstcaxQ6
This commit is contained in:
+105
-17
@@ -952,8 +952,13 @@ class PinholeRequest(BaseModel):
|
||||
allow: bool = True
|
||||
|
||||
def _build_dhcp_reservation_cmds(device: DeviceEntry) -> list:
|
||||
"""ERS 59100GTS-PWR+ has no built-in DHCP server; reservations go to OPNsense."""
|
||||
raise NotImplementedError("Switch has no DHCP server — use OPNsense for reservations")
|
||||
"""Generate ERS 59100GTS-PWR+ CLI for DHCP 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}\"",
|
||||
]
|
||||
|
||||
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."""
|
||||
@@ -972,12 +977,19 @@ 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 ARP table from switch."""
|
||||
"""Return saved device list plus live DHCP leases and ARP from switch."""
|
||||
saved = _load_devices()
|
||||
live_leases = []
|
||||
try:
|
||||
arp_raw = read_cmd("show arp")
|
||||
live_leases = _parse_arp_table(arp_raw)
|
||||
try:
|
||||
dhcp_raw = read_cmd("show ip dhcp-server leases")
|
||||
dhcp_leases = _parse_dhcp_leases(dhcp_raw)
|
||||
lease_ips = {l["ip"] for l in dhcp_leases}
|
||||
live_leases = dhcp_leases + [e for e in live_leases if e["ip"] not in lease_ips]
|
||||
except Exception:
|
||||
pass # DHCP server may not be enabled
|
||||
except Exception as e:
|
||||
log.warning(f"Could not pull ARP from switch: {e}")
|
||||
return {
|
||||
@@ -1011,8 +1023,14 @@ def delete_device(body: DeviceDelete):
|
||||
|
||||
@app.post("/api/devices/push-reservation")
|
||||
def push_reservation(body: DeviceUpdate):
|
||||
"""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")
|
||||
"""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)
|
||||
|
||||
@app.post("/api/devices/push-pinhole")
|
||||
def push_pinhole(body: PinholeRequest):
|
||||
@@ -1320,12 +1338,47 @@ def _get_opnsense_leases(cfg: dict) -> list:
|
||||
return []
|
||||
|
||||
def _get_switch_reservations() -> list:
|
||||
"""ERS 59100GTS-PWR+ has no DHCP server — always returns empty."""
|
||||
return []
|
||||
"""Fetch DHCP static bindings from ERS 59100GTS-PWR+.
|
||||
|
||||
Requires ip dhcp-server to be enabled on the switch.
|
||||
Returns empty list if DHCP server is not enabled/licensed.
|
||||
"""
|
||||
import re as _re
|
||||
try:
|
||||
raw = read_cmd("show ip 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 []
|
||||
|
||||
def _get_switch_dhcp_status() -> dict:
|
||||
"""ERS 59100GTS-PWR+ has no DHCP server — always returns not running."""
|
||||
return {"running": False, "vlans": []}
|
||||
"""Check if switch DHCP server is running and which VLANs it serves.
|
||||
|
||||
Requires ip dhcp-server to be enabled on the switch.
|
||||
"""
|
||||
import re as _re
|
||||
try:
|
||||
raw = read_cmd("show ip 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": []}
|
||||
|
||||
def _get_relay_status() -> dict:
|
||||
"""Read current DHCP relay config from each VLAN interface."""
|
||||
@@ -1450,10 +1503,18 @@ def dhcp_overview():
|
||||
switch_leases = []
|
||||
switch_status = _get_switch_dhcp_status()
|
||||
|
||||
# Discover devices via ARP (switch has no DHCP server)
|
||||
# Discover devices via DHCP leases + ARP
|
||||
try:
|
||||
arp_raw = read_cmd("show arp")
|
||||
switch_leases = _parse_arp_table(arp_raw)
|
||||
try:
|
||||
dhcp_raw = read_cmd("show ip dhcp-server leases")
|
||||
dhcp_leases = _parse_dhcp_leases(dhcp_raw)
|
||||
lease_ips = {l["ip"] for l in dhcp_leases}
|
||||
# Merge ARP entries not already in leases
|
||||
switch_leases = dhcp_leases + [e for e in switch_leases if e["ip"] not in lease_ips]
|
||||
except Exception:
|
||||
pass # DHCP server may not be enabled
|
||||
except Exception as e:
|
||||
log.warning(f"Switch ARP fetch failed: {e}")
|
||||
|
||||
@@ -1557,21 +1618,48 @@ def push_reservation_to_opnsense(body: OPNsenseReservationPush):
|
||||
@app.post("/api/dhcp/sync")
|
||||
def sync_reservation(body: SyncRequest):
|
||||
"""
|
||||
Sync a reservation in OPNsense (the switch has no DHCP server).
|
||||
Sync a reservation between switch and OPNsense.
|
||||
Directions:
|
||||
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
|
||||
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
|
||||
"""
|
||||
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 in ("to_switch", "to_opnsense", "remove_switch"):
|
||||
raise HTTPException(501, "Switch has no DHCP server; manage reservations directly in OPNsense")
|
||||
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']}"])
|
||||
|
||||
elif body.direction == "remove_opnsense":
|
||||
if not ops_res or not ops_res.get("uuid"):
|
||||
|
||||
Reference in New Issue
Block a user