Christian Manivong
13e0073659
Closes netork#116. `get_port_forwards` reports IPv4 mappings and IPv6 pinholes together, which is right — both are inbound rules someone configured. `get_nat_translations` then iterated all of them, and that is not: a pinhole is a firewall hole, and IPv6 does not do NAT at all. The entries it produced were nonsense in two ways. They paired an IPv6 host with the IPv4 WAN address, and `inside_local` came out as "2001:db8::10:22", where the last colon is a port separator and everything before it is an address that also contains colons. The two failing tests were pulling in opposite directions. `TestGetNatTranslations` was right and the implementation had drifted past it when pinhole support landed. `TestGetPortForwards::test_returns_all_rules` was the opposite: it still expected one rule from before pinholes existed, which put it in direct contradiction with TestIPv6Pinholes further down the same file.
758 lines
30 KiB
Python
758 lines
30 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)
|
||
|
||
|
||
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.
|
||
|
||
IPv6 pinholes are skipped. ``get_port_forwards`` reports them alongside
|
||
the IPv4 mappings because both are inbound rules a user configured, but
|
||
a pinhole is a firewall hole, not a translation — IPv6 does not do NAT.
|
||
Including them produced entries pairing an IPv6 host with the IPv4 WAN
|
||
address, and an ``inside_local`` of ``2001:db8::10:22`` where the last
|
||
colon is a port separator and the rest is an address.
|
||
"""
|
||
ext_ip = self._wan_external_ip()
|
||
translations: list[NATTranslationDict] = []
|
||
for forward in self.get_port_forwards():
|
||
if not forward["enabled"]:
|
||
continue
|
||
if ":" in forward["internal_ip"]:
|
||
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
|