diff --git a/switch_backend.py b/switch_backend.py index 89f9b80..3f90ba0 100644 --- a/switch_backend.py +++ b/switch_backend.py @@ -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 + 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"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. """