From 5d5bd77b08e12718e84b60188d3d610e19bbf091 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Sat, 6 Jun 2026 15:46:55 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20get=5Froute=5Fto()=20=E2=80=94=20Protok?= =?UTF-8?q?oll=20aus=20BSD-Flags,=20family-Feld,=20OSPF-Enrichment?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Protokoll-Erkennung aus flags: S=static, kein Gateway=connected, sonst=kernel - Optionale OSPF-Anreicherung via /api/quagga/ospf/routes (FRR) - family-Feld (ipv4/ipv6) aus Netzadresse abgeleitet - link#X und 0.0.0.0 als Next-Hop bereinigt - API-Response kann Liste oder Dict sein (beide Formate unterstützt) Co-Authored-By: Claude Sonnet 4.6 --- napalm_opnsense/opnsense.py | 47 ++++++++++++++++++++++++------------- 1 file changed, 31 insertions(+), 16 deletions(-) diff --git a/napalm_opnsense/opnsense.py b/napalm_opnsense/opnsense.py index 22a0064..24cc66c 100644 --- a/napalm_opnsense/opnsense.py +++ b/napalm_opnsense/opnsense.py @@ -471,17 +471,17 @@ class OPNsenseDriver(FirewallDriver): data = self._get("/api/diagnostics/interface/get_routes") routes: Dict[str, List[Dict[str, Any]]] = {} - proto_map = { - "static": "static", - "ospf": "ospf", - "bgp": "bgp", - "rip": "rip", - "kernel": "connected", - "connected": "connected", - "local": "connected", - } + # Optionally enrich with OSPF routes from FRR/Quagga + ospf_networks: set = set() + try: + ospf_data = self._get("/api/quagga/ospf/routes") + for prefix in (ospf_data if isinstance(ospf_data, list) else ospf_data.get("routes", {}).keys()): + ospf_networks.add(str(prefix)) + except Exception: + pass - for route in data.get("route", []): + route_list = data if isinstance(data, list) else data.get("route", []) + for route in route_list: network = route.get("network") or route.get("destination", "") if not network: continue @@ -490,21 +490,36 @@ class OPNsenseDriver(FirewallDriver): continue flags = route.get("flags", "").upper() - proto_raw = route.get("proto", "").lower() - proto = proto_map.get(proto_raw, proto_raw) + gateway = route.get("gateway") or route.get("nexthop", "") + iface = route.get("netif") or route.get("interface", "") + + # Clean up BSD link-layer gateway references + clean_gateway = "" if (not gateway or gateway.startswith("link#") or gateway == "0.0.0.0") else gateway + + # Determine address family from network address + family = "ipv6" if (":" in network or (gateway and ":" in gateway)) else "ipv4" + + # Determine routing protocol from BSD flags: + # S = Static, dynamic routes have no S flag + if network in ospf_networks: + proto = "ospf" + elif "S" in flags: + proto = "static" + elif not clean_gateway: + proto = "connected" + else: + proto = "kernel" if protocol and proto != protocol.lower(): continue - gateway = route.get("gateway") or route.get("nexthop", "") - iface = route.get("netif") or route.get("interface", "") - entry: Dict[str, Any] = { "protocol": proto, + "family": family, "current_active": "U" in flags, "last_active": False, "age": -1, - "next_hop": gateway if gateway not in ("link#", "0.0.0.0", "") else "", + "next_hop": clean_gateway, "outgoing_interface": iface, "selected_next_hop": True, "preference": int(route.get("priority", 0)),