Collect gateway IP, MAC, vendor, interface, and subnet CIDR. Stores the result on shared_data.gateway_info so the topology API can place the router at the center of the network map.
(self)
| 975 | self.logger.error(f"Error in get_network: {e}") |
| 976 | |
| 977 | def get_gateway_info(self): |
| 978 | """Collect gateway IP, MAC, vendor, interface, and subnet CIDR. |
| 979 | |
| 980 | Stores the result on shared_data.gateway_info so the topology API |
| 981 | can place the router at the center of the network map. |
| 982 | """ |
| 983 | info = {"gateway_ip": None, "gateway_mac": None, "gateway_vendor": None, |
| 984 | "interface": None, "subnet": None, "ragnar_ip": None} |
| 985 | try: |
| 986 | if netifaces is None: |
| 987 | return info |
| 988 | gws = netifaces.gateways() |
| 989 | gw_tuple = gws.get("default", {}).get(netifaces.AF_INET) |
| 990 | if not gw_tuple: |
| 991 | return info |
| 992 | info["gateway_ip"] = gw_tuple[0] |
| 993 | info["interface"] = gw_tuple[1] |
| 994 | |
| 995 | iface_addrs = netifaces.ifaddresses(gw_tuple[1]).get(netifaces.AF_INET) |
| 996 | if iface_addrs: |
| 997 | my_ip = iface_addrs[0]["addr"] |
| 998 | netmask = iface_addrs[0]["netmask"] |
| 999 | cidr = sum(bin(int(x)).count("1") for x in netmask.split(".")) |
| 1000 | info["ragnar_ip"] = my_ip |
| 1001 | info["subnet"] = str(ipaddress.IPv4Network(f"{my_ip}/{cidr}", strict=False)) |
| 1002 | |
| 1003 | # Resolve gateway MAC from kernel ARP cache |
| 1004 | try: |
| 1005 | result = subprocess.run( |
| 1006 | ["ip", "neigh", "show", info["gateway_ip"]], |
| 1007 | capture_output=True, text=True, timeout=5 |
| 1008 | ) |
| 1009 | for line in result.stdout.strip().splitlines(): |
| 1010 | parts = line.split() |
| 1011 | # Format: 192.168.1.1 dev wlan0 lladdr aa:bb:cc:dd:ee:ff ... |
| 1012 | if "lladdr" in parts: |
| 1013 | idx = parts.index("lladdr") |
| 1014 | if idx + 1 < len(parts): |
| 1015 | info["gateway_mac"] = parts[idx + 1] |
| 1016 | break |
| 1017 | except Exception: |
| 1018 | pass |
| 1019 | |
| 1020 | # Look up vendor from DB if we have the MAC |
| 1021 | if info["gateway_mac"]: |
| 1022 | try: |
| 1023 | db = get_db() |
| 1024 | host = db.get_host(info["gateway_mac"]) |
| 1025 | if host: |
| 1026 | info["gateway_vendor"] = host.get("vendor", "") |
| 1027 | except Exception: |
| 1028 | pass |
| 1029 | |
| 1030 | self.shared_data.gateway_info = info |
| 1031 | self.logger.info(f"Gateway info: {info['gateway_ip']} (MAC={info['gateway_mac']}) via {info['interface']}") |
| 1032 | except Exception as e: |
| 1033 | self.logger.debug(f"Could not collect gateway info: {e}") |
| 1034 | return info |