feat: get_route_to() — Protokoll aus BSD-Flags, family-Feld, OSPF-Enrichment

- 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 <noreply@anthropic.com>
This commit is contained in:
Christian Manivong
2026-06-06 15:46:55 +02:00
co-authored by Claude Sonnet 4.6
parent 908024345e
commit 5d5bd77b08
+31 -16
View File
@@ -471,17 +471,17 @@ class OPNsenseDriver(FirewallDriver):
data = self._get("/api/diagnostics/interface/get_routes") data = self._get("/api/diagnostics/interface/get_routes")
routes: Dict[str, List[Dict[str, Any]]] = {} routes: Dict[str, List[Dict[str, Any]]] = {}
proto_map = { # Optionally enrich with OSPF routes from FRR/Quagga
"static": "static", ospf_networks: set = set()
"ospf": "ospf", try:
"bgp": "bgp", ospf_data = self._get("/api/quagga/ospf/routes")
"rip": "rip", for prefix in (ospf_data if isinstance(ospf_data, list) else ospf_data.get("routes", {}).keys()):
"kernel": "connected", ospf_networks.add(str(prefix))
"connected": "connected", except Exception:
"local": "connected", 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", "") network = route.get("network") or route.get("destination", "")
if not network: if not network:
continue continue
@@ -490,21 +490,36 @@ class OPNsenseDriver(FirewallDriver):
continue continue
flags = route.get("flags", "").upper() flags = route.get("flags", "").upper()
proto_raw = route.get("proto", "").lower() gateway = route.get("gateway") or route.get("nexthop", "")
proto = proto_map.get(proto_raw, proto_raw) 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(): if protocol and proto != protocol.lower():
continue continue
gateway = route.get("gateway") or route.get("nexthop", "")
iface = route.get("netif") or route.get("interface", "")
entry: Dict[str, Any] = { entry: Dict[str, Any] = {
"protocol": proto, "protocol": proto,
"family": family,
"current_active": "U" in flags, "current_active": "U" in flags,
"last_active": False, "last_active": False,
"age": -1, "age": -1,
"next_hop": gateway if gateway not in ("link#", "0.0.0.0", "") else "", "next_hop": clean_gateway,
"outgoing_interface": iface, "outgoing_interface": iface,
"selected_next_hop": True, "selected_next_hop": True,
"preference": int(route.get("priority", 0)), "preference": int(route.get("priority", 0)),