# -*- coding: utf-8 -*- # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """NAPALM driver for AVM FritzBox routers (TR-064 via fritzconnection). FritzBox exposes its configuration and status through TR-064 (a UPnP/SOAP based management protocol). Authentication uses the FritzBox device username/password (System -> FRITZ!Box Users). Pass connection tuning via *optional_args*: optional_args={ "use_tls": True, # use HTTPS (port 49443); False for HTTP (port 49000) "port": 49443, # override the TR-064 port } FritzBox always serves TR-064 over HTTPS with a self-signed certificate; ``fritzconnection`` does not verify it, so no separate ``verify`` option is needed. This driver is read-only: it does not implement NAPALM's configuration management methods (``load_merge_candidate``, ``commit_config``, ...). """ from __future__ import annotations import socket from typing import Any from fritzconnection import FritzConnection from fritzconnection.core.exceptions import ( FritzActionError, FritzArrayIndexError, FritzConnectionException, FritzServiceError, ) from napalm.base.exceptions import ConnectionClosedException, ConnectionException from napalm_device_types import FingerprintRule, ResidentialGatewayDriver from napalm_device_types.models import ( HostDict, NATTranslationDict, PortForwardDict, RadioStatusDict, SSIDDict, VPNTunnelDict, WANStatusDict, WirelessClientDict, ) # Exceptions that indicate an optional/unavailable TR-064 service or action # (e.g. a service not present on a given FritzOS version/model). _OPTIONAL_SERVICE_ERRORS = (FritzServiceError, FritzActionError, FritzConnectionException) def _to_float(value: Any, default: float = 0.0) -> float: """Convert a TR-064 field to float, returning *default* for non-numeric values like 'Auto'.""" try: return float(value or 0) except (TypeError, ValueError): return default class FritzBoxDriver(ResidentialGatewayDriver): """NAPALM driver for AVM FritzBox (read-only, TR-064).""" VENDOR = "AVM" DRIVER_NAME = "fritzbox" OUI_PREFIXES = [ "00:04:0E", # AVM GmbH — IEEE "3C:A6:2F", # AVM Audiovisuelles Marketing — IEEE "9C:C7:A6", # AVM GmbH — IEEE "B0:F2:08", # AVM Audiovisuelles Marketing — IEEE "E0:28:6D", # AVM Audiovisuelles Marketing — IEEE ] HTTP_FINGERPRINT = [ FingerprintRule("fritz!box", weight=10.0, mandatory=True), FingerprintRule("fritzbox", weight=4.0), FingerprintRule("avm fritz", weight=4.0), ] _WAN_IP_SERVICES = ("WANIPConnection1", "WANPPPConnection1") _WLAN_SERVICES = ("WLANConfiguration1", "WLANConfiguration2", "WLANConfiguration3") def __init__( self, hostname: str, username: str, password: str, timeout: int = 60, optional_args: dict[str, Any] | None = None, ) -> None: self.hostname = hostname self.username = username self.password = password self.timeout = timeout self.optional_args = optional_args or {} # NAPALM standard attributes self.force_no_enable = True self.use_canonical_interface = False self.use_tls = bool(self.optional_args.get("use_tls", True)) self.port = int(self.optional_args.get("port") or (49443 if self.use_tls else 49000)) self.fc: FritzConnection | None = None # ------------------------------------------------------------------ # Connection management # ------------------------------------------------------------------ def open(self) -> None: """Open a TR-064 connection to the FritzBox and validate credentials.""" try: fc = FritzConnection( address=self.hostname, port=self.port, user=self.username, password=self.password, use_tls=self.use_tls, timeout=self.timeout, ) self.fc = fc # Lightweight connectivity and auth check self._call("DeviceInfo1", "GetInfo") except Exception as exc: self.fc = None raise ConnectionException( f"Cannot connect to FritzBox TR-064 at {self.hostname}:{self.port}: {exc}" ) from exc def close(self) -> None: """Drop the TR-064 connection.""" self.fc = None def is_alive(self) -> dict[str, bool]: """Return whether the connection is usable. Performs a lightweight socket-level check without sending a full TR-064 request. """ if self.fc is None: return {"is_alive": False} try: with socket.create_connection((self.hostname, self.port), timeout=5): pass return {"is_alive": True} except OSError: return {"is_alive": False} # ------------------------------------------------------------------ # Internal helpers # ------------------------------------------------------------------ def _call(self, service: str, action: str, **kwargs: Any) -> dict[str, Any]: """Invoke a TR-064 SOAP action and return its output arguments. :raises ConnectionClosedException: if called before :meth:`open`. """ if self.fc is None: raise ConnectionClosedException("Not connected – call open() first.") return self.fc.call_action(service, action, **kwargs) def _wlan_services(self) -> list[str]: """Return the WLAN service names actually present on this device.""" if self.fc is None: return [] return [s for s in self._WLAN_SERVICES if s in self.fc.services] def _wan_ip_service(self) -> str: """Return whichever WAN IP connection service this device exposes.""" if self.fc is not None: for service in self._WAN_IP_SERVICES: if service in self.fc.services: return service raise ConnectionException("No WAN IP connection service found on this FritzBox") def _wan_external_ip(self) -> str: try: service = self._wan_ip_service() return self._call(service, "GetExternalIPAddress").get("NewExternalIPAddress", "") except _OPTIONAL_SERVICE_ERRORS: return "" def _wan_external_ipv6(self, service: str) -> str: try: return self._call(service, "X_AVM-DE_GetExternalIPv6Address").get( "NewExternalIPv6Address", "" ) except _OPTIONAL_SERVICE_ERRORS: return "" # ------------------------------------------------------------------ # NAPALM standard getters # ------------------------------------------------------------------ def get_facts(self) -> dict[str, Any]: """Return a dictionary of general device facts. Calls ``DeviceInfo1.GetInfo``. """ info = self._call("DeviceInfo1", "GetInfo") try: interface_list = list(self.get_interfaces().keys()) except Exception: interface_list = [] return { "vendor": self.VENDOR, "model": info.get("NewModelName", "") or self.VENDOR, "hostname": self.hostname, "fqdn": self.hostname, "os_version": info.get("NewSoftwareVersion", ""), "serial_number": info.get("NewSerialNumber", ""), "uptime": int(info.get("NewUpTime", 0)), "interface_list": interface_list, } def get_interfaces(self) -> dict[str, dict[str, Any]]: """Return interface details keyed by interface name. Combines ``LANEthernetInterfaceConfig1.GetInfo``, ``WANCommonInterfaceConfig1.GetCommonLinkProperties`` and ``WLANConfiguration{1,2,3}.GetInfo``. """ interfaces: dict[str, dict[str, Any]] = {} available = self.fc.services if self.fc else {} if "LANEthernetInterfaceConfig1" in available: try: lan = self._call("LANEthernetInterfaceConfig1", "GetInfo") interfaces["lan"] = { "is_up": lan.get("NewStatus", "").lower() in ("up", "connected"), "is_enabled": bool(lan.get("NewEnable", False)), "description": "LAN", "last_flapped": -1.0, "mac_address": (lan.get("NewMACAddress") or "").lower(), "speed": _to_float(lan.get("NewMaxBitRate")), "mtu": 1500, } except _OPTIONAL_SERVICE_ERRORS: pass if "WANCommonInterfaceConfig1" in available: try: link = self._call("WANCommonInterfaceConfig1", "GetCommonLinkProperties") interfaces["wan"] = { "is_up": link.get("NewPhysicalLinkStatus", "").lower() in ("up", "connected"), "is_enabled": True, "description": link.get("NewWANAccessType", "WAN"), "last_flapped": -1.0, "mac_address": "", "speed": _to_float(link.get("NewLayer1DownstreamMaxBitRate")) / 1000, "mtu": 1500, } except _OPTIONAL_SERVICE_ERRORS: pass if "WANPPPConnection1" in available: try: status = self._call("WANPPPConnection1", "GetStatusInfo") interfaces["pppoe0"] = { "is_up": status.get("NewConnectionStatus", "").lower() == "connected", "is_enabled": True, "description": "PPPoE", "last_flapped": -1.0, "mac_address": "", "speed": 0.0, "mtu": 1492, } except _OPTIONAL_SERVICE_ERRORS: pass for service in self._wlan_services(): try: wlan = self._call(service, "GetInfo") except _OPTIONAL_SERVICE_ERRORS: continue interfaces[service.lower()] = { "is_up": wlan.get("NewStatus", "").lower() in ("up", "connected"), "is_enabled": bool(wlan.get("NewEnable", False)), "description": wlan.get("NewSSID", ""), "last_flapped": -1.0, "mac_address": (wlan.get("NewBSSID") or "").lower(), "speed": _to_float(wlan.get("NewMaxBitRate")), "mtu": 1500, } return interfaces def get_interfaces_ip(self) -> dict[str, dict[str, Any]]: """Return IP addresses keyed by interface name. WAN address comes from ``GetExternalIPAddress`` (and, if available, ``X_AVM-DE_GetExternalIPv6Address``) on the active WAN connection service. For PPPoE connections the IP is reported on ``pppoe0`` to match the interface name produced by :meth:`get_interfaces`. The LAN address is the address used to reach the device (TR-064 does not expose the router's own LAN IP generically). """ result: dict[str, dict[str, Any]] = {} try: service = self._wan_ip_service() ext_ip = self._call(service, "GetExternalIPAddress").get("NewExternalIPAddress", "") if ext_ip: wan_key = "pppoe0" if service == "WANPPPConnection1" else "wan" result[wan_key] = {"ipv4": {ext_ip: {"prefix_length": 32}}} ipv6 = self._wan_external_ipv6(service) if ipv6: result[wan_key]["ipv6"] = {ipv6: {"prefix_length": 64}} except _OPTIONAL_SERVICE_ERRORS: pass result["lan"] = {"ipv4": {self.hostname: {"prefix_length": 24}}} return result def get_system_config(self) -> dict[str, Any]: """Return a system configuration snapshot for storage as ``system_snapshot``. Collects DSL line statistics, firmware update status, upstream DNS servers and NTP settings. All fields are best-effort: missing TR-064 services are silently skipped. """ available = self.fc.services if self.fc else {} result: dict[str, Any] = {} if "WANDSLInterfaceConfig1" in available: try: info = self._call("WANDSLInterfaceConfig1", "GetInfo") result["dsl_enabled"] = bool(info.get("NewEnable", False)) result["dsl_status"] = info.get("NewStatus", "") result["dsl_downstream_rate"] = int(_to_float(info.get("NewDownstreamCurrRate"))) result["dsl_upstream_rate"] = int(_to_float(info.get("NewUpstreamCurrRate"))) result["dsl_downstream_max_rate"] = int(_to_float(info.get("NewDownstreamMaxRate"))) result["dsl_upstream_max_rate"] = int(_to_float(info.get("NewUpstreamMaxRate"))) # SNR margins and attenuation are stored as dB×10 by TR-064 result["dsl_downstream_noise_margin"] = int(_to_float(info.get("NewDownstreamNoiseMargin"))) result["dsl_upstream_noise_margin"] = int(_to_float(info.get("NewUpstreamNoiseMargin"))) result["dsl_downstream_attenuation"] = int(_to_float(info.get("NewDownstreamAttenuation"))) result["dsl_upstream_attenuation"] = int(_to_float(info.get("NewUpstreamAttenuation"))) except _OPTIONAL_SERVICE_ERRORS: pass try: stats = self._call("WANDSLInterfaceConfig1", "GetStatisticsTotal") result["dsl_crc_errors"] = int(_to_float(stats.get("NewCRCErrors"))) result["dsl_fec_errors"] = int(_to_float(stats.get("NewFECErrors"))) result["dsl_hec_errors"] = int(_to_float(stats.get("NewHECErrors"))) except _OPTIONAL_SERVICE_ERRORS: pass if "UserInterface1" in available: try: ui_info = self._call("UserInterface1", "X_AVM-DE_GetInfo") result["firmware_update_available"] = bool( ui_info.get("NewX_AVM-DE_UpdateState", "") == "Available" ) result["firmware_version"] = ui_info.get("NewX_AVM-DE_Version", "") except _OPTIONAL_SERVICE_ERRORS: try: ui_info = self._call("UserInterface1", "GetInfo") result["firmware_update_available"] = bool( ui_info.get("NewUpgradeAvailable", False) ) except _OPTIONAL_SERVICE_ERRORS: pass if "LANHostConfigManagement1" in available: try: dns_resp = self._call("LANHostConfigManagement1", "GetDNSServers") raw = dns_resp.get("NewDNSServers", "") or "" result["dns_servers"] = [s.strip() for s in raw.split(",") if s.strip()] except _OPTIONAL_SERVICE_ERRORS: pass if "Time1" in available: try: time_info = self._call("Time1", "GetInfo") result["ntp_server"] = time_info.get("NewNTPServer1", "") result["ntp_server2"] = time_info.get("NewNTPServer2", "") result["timezone"] = time_info.get("NewCurrentLocalTime", "") except _OPTIONAL_SERVICE_ERRORS: pass return result def get_device_warnings(self) -> list[dict[str, Any]]: """Return device warnings derived from DSL line quality and firmware state. Called during the poll cycle via ``_collect_system_snapshots``. Reads ``UserInterface1`` for firmware updates and ``WANDSLInterfaceConfig1`` for line quality. """ warnings: list[dict[str, Any]] = [] available = self.fc.services if self.fc else {} if "UserInterface1" in available: try: ui_info = self._call("UserInterface1", "X_AVM-DE_GetInfo") if ui_info.get("NewX_AVM-DE_UpdateState", "") == "Available": warnings.append({ "code": "firmware_update_available", "meta": {"latest_version": ui_info.get("NewX_AVM-DE_Version", "")}, }) except _OPTIONAL_SERVICE_ERRORS: pass if "WANDSLInterfaceConfig1" in available: try: info = self._call("WANDSLInterfaceConfig1", "GetInfo") # SNR margin is in dB×10; < 60 means < 6 dB — borderline line quality snr_down = int(_to_float(info.get("NewDownstreamNoiseMargin"))) if 0 < snr_down < 60: warnings.append({"code": "dsl_snr_low", "meta": {"snr_db_x10": snr_down}}) except _OPTIONAL_SERVICE_ERRORS: pass return warnings def get_arp_table(self, vrf: str = "") -> list[dict[str, Any]]: """Return the ARP table, derived from :meth:`get_hosts`. Each entry contains: ``interface``, ``mac``, ``ip``, ``age``. """ arp_table: list[dict[str, Any]] = [] for host in self.get_hosts(): if not host["is_active"] or not host["mac"] or not host["ip"]: continue arp_table.append( { "interface": host["interface_type"], "mac": host["mac"], "ip": host["ip"], "age": 0.0, } ) return arp_table # ------------------------------------------------------------------ # Residential gateway extensions # ------------------------------------------------------------------ def get_wan_status(self) -> WANStatusDict: """Return WAN/internet connection status. Calls ``WANCommonInterfaceConfig1.GetCommonLinkProperties`` and ``GetAddonInfos`` for physical link info, then tries each WAN connection service (WANIPConnection1, WANPPPConnection1) to get the logical connection status and external IP. """ available = self.fc.services if self.fc else {} link: dict = {} if "WANCommonInterfaceConfig1" in available: try: link = self._call("WANCommonInterfaceConfig1", "GetCommonLinkProperties") except _OPTIONAL_SERVICE_ERRORS: pass addon: dict = {} if "WANCommonInterfaceConfig1" in available: try: addon = self._call("WANCommonInterfaceConfig1", "GetAddonInfos") except _OPTIONAL_SERVICE_ERRORS: pass # Try each WAN connection service; PPPoE devices may only work with WANPPPConnection1 status: dict = {} ext_ip = "" ext_ipv6 = "" wan_service_used = "" for svc in self._WAN_IP_SERVICES: if svc not in available: continue try: status = self._call(svc, "GetStatusInfo") ext_ip = self._call(svc, "GetExternalIPAddress").get("NewExternalIPAddress", "") wan_service_used = svc break except (FritzArrayIndexError, FritzServiceError, FritzActionError, FritzConnectionException): continue except Exception: continue if wan_service_used: ext_ipv6 = self._wan_external_ipv6(wan_service_used) result: WANStatusDict = { "connection_type": link.get("NewWANAccessType", ""), "is_connected": status.get("NewConnectionStatus", "") == "Connected", "external_ip": ext_ip, "uptime": int(_to_float(status.get("NewUptime"))), "bytes_sent": int(_to_float(addon.get("NewTotalBytesSent"))), "bytes_received": int(_to_float(addon.get("NewTotalBytesReceived"))), "max_bitrate_up": int(_to_float(link.get("NewLayer1UpstreamMaxBitRate"))) // 1000, "max_bitrate_down": int(_to_float(link.get("NewLayer1DownstreamMaxBitRate"))) // 1000, "link_status": link.get("NewPhysicalLinkStatus", ""), } if ext_ipv6: result["external_ipv6"] = ext_ipv6 return result def get_port_forwards(self) -> list[PortForwardDict]: """Return configured port forwarding rules (IPv4 + IPv6 pinholes). Iterates ``GetGenericPortMappingEntry`` on the active WAN connection service until the device reports an out-of-range index. IPv6 pinholes from ``WANIPv6Firewall1`` are appended with protocol ``TCP6``/``UDP6``. """ forwards: list[PortForwardDict] = [] try: service = self._wan_ip_service() index = 0 while True: try: entry = self._call(service, "GetGenericPortMappingEntry", NewPortMappingIndex=index) except (FritzArrayIndexError, FritzConnectionException): break forward: PortForwardDict = { "name": entry.get("NewPortMappingDescription", ""), "protocol": entry.get("NewProtocol", ""), "external_port": int(entry.get("NewExternalPort", 0) or 0), "internal_ip": entry.get("NewInternalClient", ""), "internal_port": int(entry.get("NewInternalPort", 0) or 0), "enabled": bool(entry.get("NewPortMappingEnabled", False)), } remote_host = entry.get("NewRemoteHost", "") if remote_host: forward["remote_host"] = remote_host forwards.append(forward) index += 1 except _OPTIONAL_SERVICE_ERRORS: pass available = self.fc.services if self.fc else {} if "WANIPv6Firewall1" in available: try: fw = self._call("WANIPv6Firewall1", "GetFirewallStatus") if fw.get("NewFirewallEnabled"): index = 0 while True: try: ph = self._call( "WANIPv6Firewall1", "GetGenericPinholeEntry", NewPinholeIndex=index, ) except (FritzArrayIndexError, FritzConnectionException, FritzActionError): break proto_num = int(ph.get("NewProtocol", 0) or 0) proto = "TCP6" if proto_num == 6 else ("UDP6" if proto_num == 17 else str(proto_num)) forwards.append({ "name": ph.get("NewPinholeDescription", ""), "protocol": proto, "external_port": int(ph.get("NewRemotePort", 0) or 0), "internal_ip": ph.get("NewInternalIPAddress", ""), "internal_port": int(ph.get("NewInternalPort", 0) or 0), "enabled": True, }) index += 1 except _OPTIONAL_SERVICE_ERRORS: pass return forwards def get_hosts(self) -> list[HostDict]: """Return hosts known to the FritzBox. Iterates ``Hosts1.GetGenericHostEntry`` for ``Hosts1.GetHostNumberOfEntries`` entries. """ try: count = int(self._call("Hosts1", "GetHostNumberOfEntries").get( "NewHostNumberOfEntries", 0 ) or 0) except _OPTIONAL_SERVICE_ERRORS: return [] hosts: list[HostDict] = [] for index in range(count): try: entry = self._call("Hosts1", "GetGenericHostEntry", NewIndex=index) except (FritzArrayIndexError, FritzConnectionException): continue host: HostDict = { "mac": (entry.get("NewMACAddress") or "").lower(), "ip": entry.get("NewIPAddress", ""), "hostname": entry.get("NewHostName", ""), "interface_type": entry.get("NewInterfaceType", ""), "is_active": bool(entry.get("NewActive", False)), } lease = entry.get("NewLeaseTimeRemaining") if lease is not None: host["lease_time_remaining"] = int(lease) hosts.append(host) return hosts def get_nat_translations(self) -> list[NATTranslationDict]: """Return NAT translation entries derived from active port forwards. TR-064 does not expose a live connection-tracking table; this derives static NAT-PT entries from the enabled port forwarding rules, which is the closest equivalent FritzBox provides. """ ext_ip = self._wan_external_ip() translations: list[NATTranslationDict] = [] for forward in self.get_port_forwards(): if not forward["enabled"]: continue remote = forward.get("remote_host") or "0.0.0.0" translations.append( { "protocol": forward["protocol"].lower(), "inside_local": f"{forward['internal_ip']}:{forward['internal_port']}", "inside_global": f"{ext_ip}:{forward['external_port']}", "outside_local": f"{remote}:0", "outside_global": f"{remote}:0", "age": 0.0, } ) return translations def get_vpn_tunnels(self) -> dict[str, VPNTunnelDict]: """Return VPN tunnel status. FritzOS does not expose VPN (IPsec/WireGuard) tunnel state through TR-064; returns an empty dict. """ return {} def get_wireless_clients(self) -> list[WirelessClientDict]: """Return wireless clients associated with the built-in access point(s). Iterates ``GetGenericAssociatedDeviceInfo`` for each ``WLANConfiguration{1,2,3}`` service present on the device. """ clients: list[WirelessClientDict] = [] for service in self._wlan_services(): try: ssid = self._call(service, "GetInfo").get("NewSSID", "") total = int(self._call(service, "GetTotalAssociations").get( "NewTotalAssociations", 0 ) or 0) except _OPTIONAL_SERVICE_ERRORS: continue for index in range(total): try: dev = self._call( service, "GetGenericAssociatedDeviceInfo", NewAssociatedDeviceIndex=index ) except (FritzArrayIndexError, FritzConnectionException): continue client: WirelessClientDict = { "mac": (dev.get("NewAssociatedDeviceMACAddress") or "").lower(), "ssid": ssid, "radio": service, "signal": int(_to_float(dev.get("NewX_AVM-DE_SignalStrength"))), "noise": 0, "tx_rate": _to_float(dev.get("NewX_AVM-DE_Speed")), "rx_rate": 0.0, "uptime": 0, } ip = dev.get("NewAssociatedDeviceIPAddress", "") if ip: client["ip"] = ip clients.append(client) return clients def get_ssids(self) -> dict[str, SSIDDict]: """Return configured wireless networks (SSIDs). Calls ``GetInfo`` and ``GetTotalAssociations`` on each ``WLANConfiguration{1,2,3}`` service present on the device. """ ssids: dict[str, SSIDDict] = {} for service in self._wlan_services(): try: info = self._call(service, "GetInfo") except _OPTIONAL_SERVICE_ERRORS: continue ssid = info.get("NewSSID", "") if not ssid: continue try: clients = int(self._call(service, "GetTotalAssociations").get( "NewTotalAssociations", 0 ) or 0) except _OPTIONAL_SERVICE_ERRORS: clients = 0 ssids[ssid] = { "enabled": bool(info.get("NewEnable", False)), "radio": service, "bssid": (info.get("NewBSSID") or "").lower(), "encryption": info.get("NewBeaconType", ""), "hidden": bool(info.get("NewX_AVM-DE_HiddenSSID", False)), "clients": clients, } return ssids def get_radio_status(self) -> dict[str, RadioStatusDict]: """Return wireless radio status. Calls ``GetInfo`` on each ``WLANConfiguration{1,2,3}`` service present on the device. ``channel_width``, ``tx_power`` and ``frequency`` are not exposed via TR-064 and are reported as ``0``. """ radios: dict[str, RadioStatusDict] = {} for service in self._wlan_services(): try: info = self._call(service, "GetInfo") except _OPTIONAL_SERVICE_ERRORS: continue channel = int(_to_float(info.get("NewChannel"))) radios[service] = { "enabled": bool(info.get("NewEnable", False)), "band": "2.4GHz" if channel <= 14 else "5GHz", "channel": channel, "channel_width": 0, "tx_power": 0, "frequency": 0.0, } return radios