Christian ManivongandClaude Sonnet 4.6
e121496df5
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
551 lines
20 KiB
Python
551 lines
20 KiB
Python
# -*- 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)
|
||
|
||
|
||
class FritzBoxDriver(ResidentialGatewayDriver):
|
||
"""NAPALM driver for AVM FritzBox (read-only, TR-064)."""
|
||
|
||
VENDOR = "AVM"
|
||
DRIVER_NAME = "fritzbox"
|
||
OUI_PREFIXES = ["00:26:C6", "3C:A6:2F", "9C:C7:A6", "C4:86:E9", "00:04:0E", "E0:28:6D"]
|
||
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]] = {}
|
||
|
||
try:
|
||
lan = self._call("LANEthernetInterfaceConfig1", "GetInfo")
|
||
interfaces["lan"] = {
|
||
"is_up": lan.get("NewStatus", "") == "Up",
|
||
"is_enabled": bool(lan.get("NewEnable", False)),
|
||
"description": "LAN",
|
||
"last_flapped": -1.0,
|
||
"mac_address": (lan.get("NewMACAddress") or "").lower(),
|
||
"speed": float(lan.get("NewMaxBitRate", 0) or 0),
|
||
"mtu": 1500,
|
||
}
|
||
except _OPTIONAL_SERVICE_ERRORS:
|
||
pass
|
||
|
||
try:
|
||
link = self._call("WANCommonInterfaceConfig1", "GetCommonLinkProperties")
|
||
interfaces["wan"] = {
|
||
"is_up": link.get("NewPhysicalLinkStatus", "") == "Up",
|
||
"is_enabled": True,
|
||
"description": link.get("NewWANAccessType", "WAN"),
|
||
"last_flapped": -1.0,
|
||
"mac_address": "",
|
||
"speed": float(link.get("NewLayer1DownstreamMaxBitRate", 0) or 0) / 1000,
|
||
"mtu": 1500,
|
||
}
|
||
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", "") == "Up",
|
||
"is_enabled": bool(wlan.get("NewEnable", False)),
|
||
"description": wlan.get("NewSSID", ""),
|
||
"last_flapped": -1.0,
|
||
"mac_address": (wlan.get("NewBSSID") or "").lower(),
|
||
"speed": float(wlan.get("NewMaxBitRate", 0) or 0),
|
||
"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. 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:
|
||
result["wan"] = {"ipv4": {ext_ip: {"prefix_length": 32}}}
|
||
ipv6 = self._wan_external_ipv6(service)
|
||
if ipv6:
|
||
result["wan"]["ipv6"] = {ipv6: {"prefix_length": 64}}
|
||
except _OPTIONAL_SERVICE_ERRORS:
|
||
pass
|
||
|
||
result["lan"] = {"ipv4": {self.hostname: {"prefix_length": 24}}}
|
||
return result
|
||
|
||
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 line/traffic stats, plus ``GetStatusInfo`` and
|
||
``GetExternalIPAddress`` on the active WAN connection service.
|
||
"""
|
||
link = self._call("WANCommonInterfaceConfig1", "GetCommonLinkProperties")
|
||
|
||
try:
|
||
addon = self._call("WANCommonInterfaceConfig1", "GetAddonInfos")
|
||
except _OPTIONAL_SERVICE_ERRORS:
|
||
addon = {}
|
||
|
||
service = self._wan_ip_service()
|
||
status = self._call(service, "GetStatusInfo")
|
||
ext_ip = self._call(service, "GetExternalIPAddress").get("NewExternalIPAddress", "")
|
||
|
||
result: WANStatusDict = {
|
||
"connection_type": link.get("NewWANAccessType", ""),
|
||
"is_connected": status.get("NewConnectionStatus", "") == "Connected",
|
||
"external_ip": ext_ip,
|
||
"uptime": int(status.get("NewUptime", 0) or 0),
|
||
"bytes_sent": int(addon.get("NewTotalBytesSent", 0) or 0),
|
||
"bytes_received": int(addon.get("NewTotalBytesReceived", 0) or 0),
|
||
"max_bitrate_up": int(link.get("NewLayer1UpstreamMaxBitRate", 0) or 0) // 1000,
|
||
"max_bitrate_down": int(link.get("NewLayer1DownstreamMaxBitRate", 0) or 0) // 1000,
|
||
"link_status": link.get("NewPhysicalLinkStatus", ""),
|
||
}
|
||
|
||
ipv6 = self._wan_external_ipv6(service)
|
||
if ipv6:
|
||
result["external_ipv6"] = ipv6
|
||
|
||
return result
|
||
|
||
def get_port_forwards(self) -> list[PortForwardDict]:
|
||
"""Return configured port forwarding rules.
|
||
|
||
Iterates ``GetGenericPortMappingEntry`` on the active WAN connection
|
||
service until the device reports an out-of-range index.
|
||
"""
|
||
service = self._wan_ip_service()
|
||
forwards: list[PortForwardDict] = []
|
||
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
|
||
|
||
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(dev.get("NewX_AVM-DE_SignalStrength", 0) or 0),
|
||
"noise": 0,
|
||
"tx_rate": float(dev.get("NewX_AVM-DE_Speed", 0) or 0),
|
||
"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(info.get("NewChannel", 0) or 0)
|
||
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
|