NAPALM/napalm-zyxel
1
0 Fork 0
Code Issues 4 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

style: ruff format and the lint rules of the driver standard

Browse Source 
Mechanical: ruff format, and ruff check --fix with the rule set every NAPALM
driver uses (E, F, I, UP, B at line length 100, Python 3.10). What ruff
cannot fix comes in the next commit.
This commit is contained in:
Christian Manivong committed 2026-10-23 08:38:39 +02:00
1 parent f94c0a3171
commit f9fc4e3d06
3 files changed
+89 -81
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

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