init
This commit is contained in:
Christian Manivong
@@ -0,0 +1,543 @@
|
||||
# -*- 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, Dict, List, Optional
|
||||
|
||||
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 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"
|
||||
|
||||
_WAN_IP_SERVICES = ("WANIPConnection1", "WANPPPConnection1")
|
||||
_WLAN_SERVICES = ("WLANConfiguration1", "WLANConfiguration2", "WLANConfiguration3")
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
hostname: str,
|
||||
username: str,
|
||||
password: str,
|
||||
timeout: int = 60,
|
||||
optional_args: Optional[Dict[str, Any]] = 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: Optional[FritzConnection] = 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 Exception:
|
||||
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
|
||||
Reference in New Issue
Block a user