diff --git a/napalm_zyxel/zyxel_driver.py b/napalm_zyxel/zyxel_driver.py index 3b7c099..5ae0460 100644 --- a/napalm_zyxel/zyxel_driver.py +++ b/napalm_zyxel/zyxel_driver.py @@ -15,13 +15,12 @@ from typing import Any from urllib.parse import urljoin import requests -from cryptography.hazmat.primitives import serialization -from cryptography.hazmat.primitives.asymmetric import padding as asym_padding, rsa -from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes from cryptography.hazmat.backends import default_backend - +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import padding as asym_padding +from cryptography.hazmat.primitives.asymmetric import rsa +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes from napalm.base.exceptions import ConnectionClosedException, ConnectionException - from napalm_device_types import FingerprintRule, ResidentialGatewayDriver from napalm_device_types.models import ( HostDict, @@ -70,12 +69,14 @@ class ZyxelCrypto: decryptor = cipher.decryptor() padded = decryptor.update(content) + decryptor.finalize() pad_len = padded[-1] - plain = padded[: -pad_len] if 0 < pad_len <= self._AES_KEY_SIZE else padded + plain = padded[:-pad_len] if 0 < pad_len <= self._AES_KEY_SIZE else padded return json.loads(plain.decode("ascii")) def encrypt_aes_key_with_rsa(self, pubkey_pem: str) -> str: """Encrypt the AES key with the device's RSA public key for key exchange.""" - pubkey = serialization.load_pem_public_key(pubkey_pem.encode("ascii"), backend=default_backend()) + pubkey = serialization.load_pem_public_key( + pubkey_pem.encode("ascii"), backend=default_backend() + ) if not isinstance(pubkey, rsa.RSAPublicKey): raise ValueError("Not an RSA public key") encrypted = pubkey.encrypt( @@ -127,9 +128,7 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): self._session: requests.Session | None = None self._sessionkey: str | None = None self._crypto: ZyxelCrypto | None = None - self._base_url: str = ( - f"{'https' if self.use_tls else 'http'}://{self.hostname}:{self.port}" - ) + self._base_url: str = f"{'https' if self.use_tls else 'http'}://{self.hostname}:{self.port}" def _url(self, path: str) -> str: return urljoin(f"{self._base_url}/", path) @@ -143,24 +142,18 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): info = self._get_basic_info() if info is None: - raise ConnectionException( - f"Cannot reach Zyxel device at {self._base_url}" - ) + raise ConnectionException(f"Cannot reach Zyxel device at {self._base_url}") self._device_info = info crypto = ZyxelCrypto() pubkey = self._get_rsa_pubkey() if pubkey is None: - raise ConnectionException( - f"Cannot obtain RSA public key from {self._base_url}" - ) + raise ConnectionException(f"Cannot obtain RSA public key from {self._base_url}") enc_key = crypto.encrypt_aes_key_with_rsa(pubkey) login_data = { "Input_Account": self.username, - "Input_Passwd": base64.b64encode( - self.password.encode("ascii") - ).decode("ascii"), + "Input_Passwd": base64.b64encode(self.password.encode("ascii")).decode("ascii"), "RememberPassword": 0, "SHA512_password": False, } @@ -332,12 +325,14 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): for host in self.get_hosts(): if not host["is_active"] or not host["mac"] or not host["ip"]: continue - arp.append({ - "interface": host["interface_type"], - "mac": host["mac"], - "ip": host["ip"], - "age": 0.0, - }) + arp.append( + { + "interface": host["interface_type"], + "mac": host["mac"], + "ip": host["ip"], + "age": 0.0, + } + ) return arp def get_lldp_neighbors(self) -> dict[str, Any]: @@ -390,14 +385,17 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): source_interface: str = "", ) -> dict[str, Any]: """Ping a destination via the device's DAL diagnostics.""" - result = self._dal_query("PINGTEST", { - "ProtocolVersion": "IPv4", - "Host": destination, - "DiagnosticsState": "Requested", - "DNSServer": "", - "NumberOfRepetitions": count, - "type": 0, - }) + result = self._dal_query( + "PINGTEST", + { + "ProtocolVersion": "IPv4", + "Host": destination, + "DiagnosticsState": "Requested", + "DNSServer": "", + "NumberOfRepetitions": count, + "type": 0, + }, + ) if result is None: return {"success": {}, "error": "Ping not supported via DAL"} return { @@ -487,9 +485,7 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): """Not supported on Zyxel VMG.""" raise NotImplementedError("Zyxel VMG does not support config replacement via NAPALM") - def load_merge_candidate( - self, filename: str | None = None, config: str | None = None - ) -> None: + def load_merge_candidate(self, filename: str | None = None, config: str | None = None) -> None: """Not supported on Zyxel VMG.""" raise NotImplementedError("Zyxel VMG does not support config merge via NAPALM") @@ -562,14 +558,16 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): def _parse_port_forwards_html(self, html: str) -> list[PortForwardDict]: forwards: list[PortForwardDict] = [] for match in self._PORT_FORWARD_RE.finditer(html): - forwards.append({ - "name": match.group("name").strip(), - "protocol": match.group("proto").upper(), - "external_port": int(match.group("ext_port")), - "internal_ip": match.group("int_ip"), - "internal_port": int(match.group("int_port")), - "enabled": True, - }) + forwards.append( + { + "name": match.group("name").strip(), + "protocol": match.group("proto").upper(), + "external_port": int(match.group("ext_port")), + "internal_ip": match.group("int_ip"), + "internal_port": int(match.group("int_port")), + "enabled": True, + } + ) return forwards def get_hosts(self) -> list[HostDict]: @@ -620,13 +618,15 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): def _parse_hosts_html(self, html: str) -> list[HostDict]: hosts: list[HostDict] = [] for match in self._HOST_RE.finditer(html): - hosts.append({ - "mac": match.group("mac").lower(), - "ip": match.group("ip"), - "hostname": match.group("hostname") or "", - "interface_type": "lan", - "is_active": True, - }) + hosts.append( + { + "mac": match.group("mac").lower(), + "ip": match.group("ip"), + "hostname": match.group("hostname") or "", + "interface_type": "lan", + "is_active": True, + } + ) return hosts def get_nat_translations(self) -> list[NATTranslationDict]: @@ -636,14 +636,16 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): for forward in self.get_port_forwards(): if not forward["enabled"]: continue - translations.append({ - "protocol": forward["protocol"].lower(), - "inside_local": f"{forward['internal_ip']}:{forward['internal_port']}", - "inside_global": f"{wan_ip}:{forward['external_port']}", - "outside_local": "0.0.0.0:0", - "outside_global": "0.0.0.0:0", - "age": 0.0, - }) + translations.append( + { + "protocol": forward["protocol"].lower(), + "inside_local": f"{forward['internal_ip']}:{forward['internal_port']}", + "inside_global": f"{wan_ip}:{forward['external_port']}", + "outside_local": "0.0.0.0:0", + "outside_global": "0.0.0.0:0", + "age": 0.0, + } + ) return translations def get_vpn_tunnels(self) -> dict[str, VPNTunnelDict]: @@ -701,17 +703,19 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): def _parse_wireless_clients_html(self, html: str) -> list[WirelessClientDict]: clients: list[WirelessClientDict] = [] for match in self._WIFI_CLIENT_RE.finditer(html): - clients.append({ - "mac": match.group("mac").lower(), - "ssid": "", - "radio": "wlan0", - "signal": int(match.group("rssi")), - "noise": 0, - "tx_rate": 0.0, - "rx_rate": 0.0, - "uptime": 0, - "ip": match.group("ip"), - }) + clients.append( + { + "mac": match.group("mac").lower(), + "ssid": "", + "radio": "wlan0", + "signal": int(match.group("rssi")), + "noise": 0, + "tx_rate": 0.0, + "rx_rate": 0.0, + "uptime": 0, + "ip": match.group("ip"), + } + ) return clients def get_ssids(self) -> dict[str, SSIDDict]: @@ -784,7 +788,9 @@ class ZyxelVMGDriver(ResidentialGatewayDriver): "enabled": bool(entry.get("Enabled", entry.get("enabled", True))), "band": "2.4GHz" if channel <= 14 else "5GHz", "channel": channel, - "channel_width": int(entry.get("ChannelWidth", entry.get("channel_width", 0)) or 0), + "channel_width": int( + entry.get("ChannelWidth", entry.get("channel_width", 0)) or 0 + ), "tx_power": int(entry.get("TxPower", entry.get("tx_power", 0)) or 0), "frequency": float(entry.get("Frequency", entry.get("frequency", 0)) or 0), } diff --git a/pyproject.toml b/pyproject.toml index b40b5a2..4af67ca 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -49,7 +49,14 @@ include = ["napalm_zyxel*"] [tool.ruff] line-length = 100 -target-version = "py39" +target-version = "py310" + +[tool.ruff.lint] +select = ["E", "F", "I", "UP", "B"] + +[tool.ruff.lint.per-file-ignores] +# Recorded device output keeps its lines. +"tests/**" = ["E501"] [tool.pytest.ini_options] testpaths = ["tests"] diff --git a/tests/test_zyxel.py b/tests/test_zyxel.py index 904020c..eab9db1 100644 --- a/tests/test_zyxel.py +++ b/tests/test_zyxel.py @@ -3,12 +3,10 @@ from unittest.mock import MagicMock, patch import pytest import requests - from napalm.base.exceptions import ConnectionException from napalm_zyxel import ZyxelVMGDriver - MOCK_BASIC_INFO = { "ModelName": "VMG4005-B50A", "SoftwareVersion": "V5.13(ABKS.1)C0", @@ -117,7 +115,8 @@ class TestZyxelVMGDriver: mock_crypto = MagicMock() mock_crypto.encrypt_json.return_value = {"content": "ENC", "iv": "IV", "key": ""} mock_crypto.decrypt_response.return_value = { - "result": "ZCFG_SUCCESS", "sessionkey": "12345" + "result": "ZCFG_SUCCESS", + "sessionkey": "12345", } mock_crypto.encrypt_aes_key_with_rsa.return_value = "ENC_KEY" mock_crypto_cls.return_value = mock_crypto @@ -127,12 +126,8 @@ class TestZyxelVMGDriver: session.post.return_value = MockResponse(MOCK_LOGIN_RESPONSE) with patch.object(self.driver, "_session", session, create=False): - with patch.object( - self.driver, "_get_rsa_pubkey", return_value=MOCK_RSA_KEY - ): - with patch.object( - self.driver, "_get_basic_info", return_value=MOCK_BASIC_INFO - ): + with patch.object(self.driver, "_get_rsa_pubkey", return_value=MOCK_RSA_KEY): + with patch.object(self.driver, "_get_basic_info", return_value=MOCK_BASIC_INFO): self.driver._session = None self.driver._crypto = None self.driver._sessionkey = None