commit 0ba884c808be00f09a88bbd1ddc2856e5b71489e Author: Christian Manivong Date: Sat Jun 13 03:18:28 2026 +0200 init diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..2207b4c --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,44 @@ +name: CI + +on: + push: + branches: ["**"] + pull_request: + branches: ["**"] + +jobs: + test: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python-version: ["3.9", "3.10", "3.11", "3.12"] + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Setup Python + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + cache: pip + + - name: Install package with dev extras + run: | + python -m pip install --upgrade pip + python -m pip install -e ".[dev]" + + - name: Run unit tests + run: | + python -m pytest -q --tb=short + + - name: Build wheel and sdist + run: | + python -m pip install build + python -m build + + - name: Upload dist artifacts + uses: actions/upload-artifact@v4 + with: + name: dist-${{ matrix.python-version }} + path: dist/* \ No newline at end of file diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..80bbc57 --- /dev/null +++ b/.gitignore @@ -0,0 +1,42 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# Virtual environments +.venv/ +venv/ +ENV/ +env/ + +# Distribution / packaging +.Python +build/ +dist/ +*.egg-info/ +*.egg +*.whl +pip-wheel-metadata/ + +# Testing / coverage +.pytest_cache/ +.coverage +.coverage.* +coverage.xml +htmlcov/ + +# Type checking / lint +.mypy_cache/ +.ruff_cache/ +.tox/ + +# IDE / editor +.vscode/ +.idea/ + +# OS / misc +.DS_Store +Thumbs.db + +# Logs +*.log diff --git a/README.md b/README.md new file mode 100644 index 0000000..427cfd6 --- /dev/null +++ b/README.md @@ -0,0 +1,96 @@ +# napalm-fritzbox + +NAPALM community driver for **AVM FritzBox** routers (read-only, via TR-064). + +## Requirements + +| Dependency | Minimum version | +|---|---| +| Python | 3.9 | +| NAPALM | 4.0 | +| fritzconnection | 1.13 | + +## Installation + +```bash +pip install napalm-fritzbox +``` + +Or from source: + +```bash +git clone https://github.com/napalm-automation-community/napalm-fritzbox +cd napalm-fritzbox +pip install -e . +``` + +## Quick start + +```python +from napalm import get_network_driver + +driver = get_network_driver("fritzbox") +with driver( + "192.168.178.1", + "user", + "pass", + optional_args={"use_tls": True}, +) as d: + print(d.get_facts()) + print(d.get_wan_status()) + print(d.get_hosts()) +``` + +### Example + +```python +driver = get_network_driver("fritzbox") +d = driver("fritz.box", "user", "pass") +d.open() + +print(d.get_wan_status()) +print(d.get_port_forwards()) +print(d.get_ssids()) +print(d.get_wireless_clients()) +d.close() +``` + +## Optional arguments + +| Argument | Default | Description | +|---|---|---| +| `use_tls` | `True` | Use HTTPS (port 49443) instead of HTTP (port 49000) | +| `port` | `49443` / `49000` | Override the TR-064 port | + +FritzBox serves TR-064 over HTTPS with a self-signed certificate; `fritzconnection` +does not verify it, so no separate `verify` option is needed. + +## Limitations + +- Read-only: configuration management methods (`load_merge_candidate`, + `commit_config`, ...) are not implemented. +- `get_nat_translations()` is derived from the enabled port-forwarding rules — + TR-064 does not expose a live connection-tracking table. +- `get_vpn_tunnels()` returns an empty dict — FritzOS does not expose VPN + tunnel status via TR-064. +- `channel_width`, `tx_power` and `frequency` in `get_radio_status()` are not + available via TR-064 and are reported as `0`. + +## Development + +```bash +pip install -e ".[dev]" +pytest tests/ +``` + +## CI + +This project includes a GitHub Actions workflow that: +- Runs unit tests across Python 3.9–3.12 +- Builds sdist and wheel + +See [.github/workflows/ci.yml](.github/workflows/ci.yml). + +## License + +Apache 2.0 — see [LICENSE](LICENSE). diff --git a/napalm_fritzbox/__init__.py b/napalm_fritzbox/__init__.py new file mode 100644 index 0000000..7f941eb --- /dev/null +++ b/napalm_fritzbox/__init__.py @@ -0,0 +1,3 @@ +from napalm_fritzbox.fritzbox import FritzBoxDriver + +__all__ = ["FritzBoxDriver"] diff --git a/napalm_fritzbox/fritzbox.py b/napalm_fritzbox/fritzbox.py new file mode 100644 index 0000000..1b87ae9 --- /dev/null +++ b/napalm_fritzbox/fritzbox.py @@ -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 diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..5e18b53 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,55 @@ +[build-system] +requires = ["setuptools>=68", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "napalm-fritzbox" +version = "0.1.0" +description = "NAPALM driver for AVM FritzBox (TR-064 via fritzconnection)." +readme = "README.md" +license = { text = "Apache-2.0" } +requires-python = ">=3.9" +authors = [ + { name = "Christian Manivong" }, +] +classifiers = [ + "Topic :: Utilities", + "License :: OSI Approved :: Apache Software License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Operating System :: POSIX :: Linux", + "Operating System :: MacOS", +] +dependencies = [ + "napalm>=4.0.0", + "napalm_device_types>=0.4.0", + "fritzconnection>=1.13.0", +] + +[project.optional-dependencies] +dev = [ + "pytest", + "pytest-cov", + "black", + "ruff", +] + +[project.entry-points."napalm.drivers"] +fritzbox = "napalm_fritzbox.fritzbox:FritzBoxDriver" + +[project.urls] +Repository = "https://github.com/napalm-automation-community/napalm-fritzbox" + +[tool.setuptools.packages.find] +where = ["."] +include = ["napalm_fritzbox*"] + +[tool.ruff] +line-length = 100 +target-version = "py39" + +[tool.pytest.ini_options] +testpaths = ["tests"] diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/__init__.py b/tests/unit/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/test_driver.py b/tests/unit/test_driver.py new file mode 100644 index 0000000..61706a9 --- /dev/null +++ b/tests/unit/test_driver.py @@ -0,0 +1,464 @@ +"""Unit tests for FritzBoxDriver — no real device required.""" + +from unittest.mock import MagicMock, patch + +import pytest +from fritzconnection.core.exceptions import ( + FritzActionError, + FritzArrayIndexError, + FritzServiceError, +) +from napalm.base.exceptions import ConnectionClosedException, ConnectionException + +from napalm_fritzbox.fritzbox import FritzBoxDriver + +# --------------------------------------------------------------------------- +# Sample TR-064 responses +# --------------------------------------------------------------------------- + +DEVICE_INFO = { + "NewModelName": "FRITZ!Box 7590", + "NewSoftwareVersion": "7.57", + "NewSerialNumber": "1234567890ABC", + "NewUpTime": 123456, +} + +LAN_INFO = { + "NewEnable": True, + "NewStatus": "Up", + "NewMACAddress": "AA:BB:CC:DD:EE:01", + "NewMaxBitRate": 1000, +} + +WAN_LINK_PROPERTIES = { + "NewWANAccessType": "DSL", + "NewPhysicalLinkStatus": "Up", + "NewLayer1UpstreamMaxBitRate": 40000000, + "NewLayer1DownstreamMaxBitRate": 250000000, +} + +WAN_ADDON_INFOS = { + "NewTotalBytesSent": 123456789, + "NewTotalBytesReceived": 987654321, +} + +WAN_STATUS_INFO = { + "NewConnectionStatus": "Connected", + "NewUptime": 345600, +} + +WAN_EXTERNAL_IP = {"NewExternalIPAddress": "203.0.113.7"} + +HOSTS_COUNT = {"NewHostNumberOfEntries": 2} + +HOST_ENTRIES = [ + { + "NewMACAddress": "11:22:33:44:55:66", + "NewIPAddress": "192.168.1.50", + "NewHostName": "laptop", + "NewInterfaceType": "WLAN", + "NewActive": True, + "NewLeaseTimeRemaining": 3600, + }, + { + "NewMACAddress": "AA:11:22:33:44:55", + "NewIPAddress": "192.168.1.51", + "NewHostName": "desktop", + "NewInterfaceType": "LAN", + "NewActive": False, + }, +] + +PORT_FORWARDS = [ + { + "NewPortMappingDescription": "Web", + "NewProtocol": "TCP", + "NewExternalPort": 443, + "NewInternalClient": "192.168.1.10", + "NewInternalPort": 443, + "NewPortMappingEnabled": True, + "NewRemoteHost": "", + } +] + +WLAN1_INFO = { + "NewEnable": True, + "NewStatus": "Up", + "NewSSID": "MyWifi", + "NewBSSID": "AA:BB:CC:DD:EE:02", + "NewMaxBitRate": 866, + "NewChannel": 36, + "NewBeaconType": "11i", +} + +WLAN1_TOTAL_ASSOCIATIONS = {"NewTotalAssociations": 1} + +WLAN1_ASSOCIATED_DEVICE = { + "NewAssociatedDeviceMACAddress": "11:22:33:44:55:66", + "NewAssociatedDeviceIPAddress": "192.168.1.50", + "NewX_AVM-DE_SignalStrength": 80, + "NewX_AVM-DE_Speed": 866, +} + +WLAN2_INFO = { + "NewEnable": False, + "NewStatus": "Down", + "NewSSID": "MyWifi-Guest", + "NewBSSID": "AA:BB:CC:DD:EE:03", + "NewMaxBitRate": 0, + "NewChannel": 1, + "NewBeaconType": "11i", +} + +WLAN2_TOTAL_ASSOCIATIONS = {"NewTotalAssociations": 0} + + +# --------------------------------------------------------------------------- +# call_action dispatch +# --------------------------------------------------------------------------- + +_SIMPLE_RESPONSES = { + ("DeviceInfo1", "GetInfo"): DEVICE_INFO, + ("LANEthernetInterfaceConfig1", "GetInfo"): LAN_INFO, + ("WANCommonInterfaceConfig1", "GetCommonLinkProperties"): WAN_LINK_PROPERTIES, + ("WANCommonInterfaceConfig1", "GetAddonInfos"): WAN_ADDON_INFOS, + ("WANIPConnection1", "GetStatusInfo"): WAN_STATUS_INFO, + ("WANIPConnection1", "GetExternalIPAddress"): WAN_EXTERNAL_IP, + ("Hosts1", "GetHostNumberOfEntries"): HOSTS_COUNT, + ("WLANConfiguration1", "GetInfo"): WLAN1_INFO, + ("WLANConfiguration1", "GetTotalAssociations"): WLAN1_TOTAL_ASSOCIATIONS, + ("WLANConfiguration2", "GetInfo"): WLAN2_INFO, + ("WLANConfiguration2", "GetTotalAssociations"): WLAN2_TOTAL_ASSOCIATIONS, +} + + +def _call_action(service, action, **kwargs): + if (service, action) == ("Hosts1", "GetGenericHostEntry"): + index = kwargs["NewIndex"] + if index >= len(HOST_ENTRIES): + raise FritzArrayIndexError("index out of range") + return HOST_ENTRIES[index] + + if (service, action) == ("WANIPConnection1", "GetGenericPortMappingEntry"): + index = kwargs["NewPortMappingIndex"] + if index >= len(PORT_FORWARDS): + raise FritzArrayIndexError("index out of range") + return PORT_FORWARDS[index] + + if (service, action) == ("WLANConfiguration1", "GetGenericAssociatedDeviceInfo"): + index = kwargs["NewAssociatedDeviceIndex"] + if index >= 1: + raise FritzArrayIndexError("index out of range") + return WLAN1_ASSOCIATED_DEVICE + + if (service, action) == ("WLANConfiguration2", "GetGenericAssociatedDeviceInfo"): + raise FritzArrayIndexError("index out of range") + + if action == "X_AVM-DE_GetExternalIPv6Address": + raise FritzActionError("action not supported") + + return _SIMPLE_RESPONSES[(service, action)] + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def driver(): + """Return a driver instance with a mocked FritzConnection.""" + mock_fc = MagicMock() + mock_fc.services = { + "DeviceInfo1": MagicMock(), + "LANEthernetInterfaceConfig1": MagicMock(), + "WANCommonInterfaceConfig1": MagicMock(), + "WANIPConnection1": MagicMock(), + "Hosts1": MagicMock(), + "WLANConfiguration1": MagicMock(), + "WLANConfiguration2": MagicMock(), + } + mock_fc.call_action.side_effect = _call_action + + drv = FritzBoxDriver(hostname="fritz.box", username="user", password="pass") + drv.fc = mock_fc + yield drv + + +# --------------------------------------------------------------------------- +# open() / close() / is_alive() +# --------------------------------------------------------------------------- + + +class TestOpenClose: + def test_open_raises_connection_exception_on_error(self): + drv = FritzBoxDriver("unreachable.invalid", "u", "p") + with patch("napalm_fritzbox.fritzbox.FritzConnection", side_effect=OSError("boom")): + with pytest.raises(ConnectionException): + drv.open() + + def test_open_sets_fc_on_success(self): + drv = FritzBoxDriver("fritz.box", "u", "p") + mock_fc = MagicMock() + mock_fc.call_action.return_value = DEVICE_INFO + with patch("napalm_fritzbox.fritzbox.FritzConnection", return_value=mock_fc): + drv.open() + assert drv.fc is mock_fc + + def test_close_clears_fc(self, driver): + driver.close() + assert driver.fc is None + + def test_close_is_idempotent(self, driver): + driver.close() + driver.close() + + +class TestIsAlive: + def test_returns_false_when_no_fc(self): + drv = FritzBoxDriver("fritz.box", "u", "p") + assert drv.is_alive() == {"is_alive": False} + + def test_returns_true_on_successful_connection(self, driver): + with patch("napalm_fritzbox.fritzbox.socket.create_connection") as mock_conn: + mock_conn.return_value.__enter__ = MagicMock(return_value=None) + mock_conn.return_value.__exit__ = MagicMock(return_value=False) + result = driver.is_alive() + assert result == {"is_alive": True} + + def test_returns_false_on_socket_error(self, driver): + with patch( + "napalm_fritzbox.fritzbox.socket.create_connection", + side_effect=OSError("refused"), + ): + result = driver.is_alive() + assert result == {"is_alive": False} + + +# --------------------------------------------------------------------------- +# _call() +# --------------------------------------------------------------------------- + + +class TestCall: + def test_raises_when_not_connected(self): + drv = FritzBoxDriver("fritz.box", "u", "p") + with pytest.raises(ConnectionClosedException): + drv._call("DeviceInfo1", "GetInfo") + + +# --------------------------------------------------------------------------- +# get_facts() +# --------------------------------------------------------------------------- + + +class TestGetFacts: + def test_vendor_is_avm(self, driver): + assert driver.get_facts()["vendor"] == "AVM" + + def test_model_parsed(self, driver): + assert driver.get_facts()["model"] == "FRITZ!Box 7590" + + def test_os_version_parsed(self, driver): + assert driver.get_facts()["os_version"] == "7.57" + + def test_serial_number_parsed(self, driver): + assert driver.get_facts()["serial_number"] == "1234567890ABC" + + def test_uptime_parsed(self, driver): + assert driver.get_facts()["uptime"] == 123456 + + def test_interface_list_populated(self, driver): + facts = driver.get_facts() + assert "lan" in facts["interface_list"] + assert "wan" in facts["interface_list"] + + +# --------------------------------------------------------------------------- +# get_interfaces() / get_interfaces_ip() +# --------------------------------------------------------------------------- + + +class TestGetInterfaces: + def test_lan_interface_present(self, driver): + interfaces = driver.get_interfaces() + assert interfaces["lan"]["is_up"] is True + assert interfaces["lan"]["mac_address"] == "aa:bb:cc:dd:ee:01" + + def test_wan_interface_present(self, driver): + interfaces = driver.get_interfaces() + assert interfaces["wan"]["is_up"] is True + assert interfaces["wan"]["description"] == "DSL" + + def test_wlan_interfaces_present(self, driver): + interfaces = driver.get_interfaces() + assert interfaces["wlanconfiguration1"]["description"] == "MyWifi" + assert interfaces["wlanconfiguration2"]["is_enabled"] is False + + +class TestGetInterfacesIp: + def test_wan_external_ip(self, driver): + result = driver.get_interfaces_ip() + assert result["wan"]["ipv4"] == {"203.0.113.7": {"prefix_length": 32}} + + def test_lan_address_falls_back_to_hostname(self, driver): + result = driver.get_interfaces_ip() + assert "fritz.box" in result["lan"]["ipv4"] + + +# --------------------------------------------------------------------------- +# get_hosts() / get_arp_table() +# --------------------------------------------------------------------------- + + +class TestGetHosts: + def test_returns_all_hosts(self, driver): + assert len(driver.get_hosts()) == 2 + + def test_active_host_fields(self, driver): + host = driver.get_hosts()[0] + assert host["mac"] == "11:22:33:44:55:66" + assert host["ip"] == "192.168.1.50" + assert host["hostname"] == "laptop" + assert host["interface_type"] == "WLAN" + assert host["is_active"] is True + assert host["lease_time_remaining"] == 3600 + + def test_inactive_host_has_no_lease(self, driver): + host = driver.get_hosts()[1] + assert host["is_active"] is False + assert "lease_time_remaining" not in host + + +class TestGetArpTable: + def test_only_active_hosts_with_mac_and_ip(self, driver): + arp_table = driver.get_arp_table() + assert len(arp_table) == 1 + assert arp_table[0]["mac"] == "11:22:33:44:55:66" + assert arp_table[0]["ip"] == "192.168.1.50" + + +# --------------------------------------------------------------------------- +# get_wan_status() +# --------------------------------------------------------------------------- + + +class TestGetWanStatus: + def test_connection_type(self, driver): + assert driver.get_wan_status()["connection_type"] == "DSL" + + def test_is_connected(self, driver): + assert driver.get_wan_status()["is_connected"] is True + + def test_external_ip(self, driver): + assert driver.get_wan_status()["external_ip"] == "203.0.113.7" + + def test_uptime(self, driver): + assert driver.get_wan_status()["uptime"] == 345600 + + def test_traffic_counters(self, driver): + status = driver.get_wan_status() + assert status["bytes_sent"] == 123456789 + assert status["bytes_received"] == 987654321 + + def test_max_bitrates_in_kbit(self, driver): + status = driver.get_wan_status() + assert status["max_bitrate_up"] == 40000 + assert status["max_bitrate_down"] == 250000 + + def test_no_ipv6_when_unsupported(self, driver): + assert "external_ipv6" not in driver.get_wan_status() + + +# --------------------------------------------------------------------------- +# get_port_forwards() / get_nat_translations() +# --------------------------------------------------------------------------- + + +class TestGetPortForwards: + def test_returns_all_rules(self, driver): + assert len(driver.get_port_forwards()) == 1 + + def test_rule_fields(self, driver): + forward = driver.get_port_forwards()[0] + assert forward["name"] == "Web" + assert forward["protocol"] == "TCP" + assert forward["external_port"] == 443 + assert forward["internal_ip"] == "192.168.1.10" + assert forward["internal_port"] == 443 + assert forward["enabled"] is True + + def test_empty_remote_host_not_included(self, driver): + assert "remote_host" not in driver.get_port_forwards()[0] + + +class TestGetNatTranslations: + def test_derived_from_enabled_port_forwards(self, driver): + translations = driver.get_nat_translations() + assert len(translations) == 1 + assert translations[0]["protocol"] == "tcp" + assert translations[0]["inside_local"] == "192.168.1.10:443" + assert translations[0]["inside_global"] == "203.0.113.7:443" + + +# --------------------------------------------------------------------------- +# get_vpn_tunnels() +# --------------------------------------------------------------------------- + + +class TestGetVpnTunnels: + def test_returns_empty_dict(self, driver): + assert driver.get_vpn_tunnels() == {} + + +# --------------------------------------------------------------------------- +# get_wireless_clients() / get_ssids() / get_radio_status() +# --------------------------------------------------------------------------- + + +class TestGetWirelessClients: + def test_returns_associated_clients(self, driver): + clients = driver.get_wireless_clients() + assert len(clients) == 1 + client = clients[0] + assert client["mac"] == "11:22:33:44:55:66" + assert client["ssid"] == "MyWifi" + assert client["radio"] == "WLANConfiguration1" + assert client["signal"] == 80 + assert client["ip"] == "192.168.1.50" + + +class TestGetSsids: + def test_returns_ssid_for_each_radio(self, driver): + ssids = driver.get_ssids() + assert "MyWifi" in ssids + assert "MyWifi-Guest" in ssids + + def test_ssid_fields(self, driver): + ssid = driver.get_ssids()["MyWifi"] + assert ssid["enabled"] is True + assert ssid["radio"] == "WLANConfiguration1" + assert ssid["bssid"] == "aa:bb:cc:dd:ee:02" + assert ssid["clients"] == 1 + + def test_disabled_ssid_has_zero_clients(self, driver): + assert driver.get_ssids()["MyWifi-Guest"]["clients"] == 0 + + +class TestGetRadioStatus: + def test_returns_entry_per_radio(self, driver): + radios = driver.get_radio_status() + assert "WLANConfiguration1" in radios + assert "WLANConfiguration2" in radios + + def test_5ghz_radio_enabled(self, driver): + radio = driver.get_radio_status()["WLANConfiguration1"] + assert radio["enabled"] is True + assert radio["band"] == "5GHz" + assert radio["channel"] == 36 + + def test_2_4ghz_radio_disabled(self, driver): + radio = driver.get_radio_status()["WLANConfiguration2"] + assert radio["enabled"] is False + assert radio["band"] == "2.4GHz" + assert radio["channel"] == 1