Merge pull request 'chore: adopt the NAPALM driver standard' (#21) from chore/driver-standard into main

Unfloored, the CI lowest job installs netaddr 0.3.1, which has no
IPAddress, and every test module fails to import. 0.7.19 has it; 194 tests
pass at the lowest declared versions.

.gitea/workflows/ci.yml

@@ -0,0 +1,15 @@
+name: CI
+
+on:
+  push:
+    branches: [main, master]
+    tags: ["v*"]
+  pull_request:
+
+jobs:
+  driver:
+    uses: NAPALM/driver-standard/.gitea/workflows/driver.yml@main
+    with:
+      standard-ref: main
+    secrets:
+      PYPI_TOKEN: ${{ secrets.PYPI_TOKEN }}

CHANGELOG.md

@@ -0,0 +1,25 @@
+# Changelog
+
+All notable changes to napalm-openwrt. The format follows
+[Keep a Changelog](https://keepachangelog.com/), versions follow
+[Semantic Versioning](https://semver.org/). Versions up to 0.6.0 were not
+recorded here; their history is the git log.
+
+## [Unreleased]
+
+## [0.7.0] - 2026-10-23
+
+The first release on PyPI.
+
+### Changed
+- The driver follows the NAPALM driver standard
+  (https://git.netork.io/NAPALM/driver-standard): CI on Gitea runs ruff, mypy, the
+  tests on Python 3.10–3.13 and at the lowest declared versions, bandit, pip-audit and
+  a build check, and a release tag publishes to PyPI. The package now ships `py.typed`
+  and a LICENSE file. **Python 3.8 and 3.9 are no longer supported**;
+  napalm-device-types never ran on them.
+- napalm-device-types 4.21.1 or newer is required.
+- Exceptions the driver deliberately ignores (the snmpd community in
+  `get_snmp_config()`, the network VLANs in `get_ssids()`) are now logged at debug
+  level.
+- The project links point at git.netork.io instead of GitHub.

LICENSE

@@ -0,0 +1,191 @@
+                                 Apache License
+                           Version 2.0, January 2004
+                        http://www.apache.org/licenses/
+
+   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+   1. Definitions.
+
+      "License" shall mean the terms and conditions for use, reproduction,
+      and distribution as defined by Sections 1 through 9 of this document.
+
+      "Licensor" shall mean the copyright owner or entity authorized by
+      the copyright owner that is granting the License.
+
+      "Legal Entity" shall mean the union of the acting entity and all
+      other entities that control, are controlled by, or are under common
+      control with that entity. For the purposes of this definition,
+      "control" means (i) the power, direct or indirect, to cause the
+      direction or management of such entity, whether by contract or
+      otherwise, or (ii) ownership of fifty percent (50%) or more of the
+      outstanding shares, or (iii) beneficial ownership of such entity.
+
+      "You" (or "Your") shall mean an individual or Legal Entity
+      exercising permissions granted by this License.
+
+      "Source" form shall mean the preferred form for making modifications,
+      including but not limited to software source code, documentation
+      source, and configuration files.
+
+      "Object" form shall mean any form resulting from mechanical
+      transformation or translation of a Source form, including but
+      not limited to compiled object code, generated documentation,
+      and conversions to other media types.
+
+      "Work" shall mean the work of authorship made available under
+      the License, as indicated by a copyright notice that is included in
+      or attached to the work (an example is provided in the Appendix below).
+
+      "Derivative Works" shall mean any work, whether in Source or Object
+      form, that is based on (or derived from) the Work and for which the
+      editorial revisions, annotations, elaborations, or other transformations
+      represent, as a whole, an original work of authorship. For the purposes
+      of this License, Derivative Works shall not include works that remain
+      separable from, or merely link (or bind by name) to the interfaces of,
+      the Work and Derivative Works thereof.
+
+      "Contribution" shall mean, as submitted to the Licensor for inclusion
+      in the Work by the copyright owner or by an individual or Legal Entity
+      authorized to submit on behalf of the copyright owner. For the purposes
+      of this definition, "submitted" means any form of electronic, verbal,
+      or written communication sent to the Licensor or its representatives,
+      including but not limited to communication on electronic mailing lists,
+      source code control systems, and issue tracking systems that are managed
+      by, or on behalf of, the Licensor for the purpose of discussing and
+      improving the Work, but excluding communication that is conspicuously
+      marked or designated in writing by the copyright owner as "Not a
+      Contribution."
+
+      "Contributor" shall mean Licensor and any Legal Entity on behalf of
+      whom a Contribution has been received by the Licensor and incorporated
+      within the Work.
+
+   2. Grant of Copyright License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      copyright license to reproduce, prepare Derivative Works of,
+      publicly display, publicly perform, sublicense, and distribute the
+      Work and such Derivative Works in Source or Object form.
+
+   3. Grant of Patent License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      (except as stated in this section) patent license to make, have made,
+      use, offer to sell, sell, import, and otherwise transfer the Work,
+      where such license applies only to those patent claims licensable
+      by such Contributor that are necessarily infringed by their
+      Contribution(s) alone or by the combined Contribution(s) and the Work
+      to which such Contribution(s) was submitted. If You institute patent
+      litigation against any entity (including a cross-claim or counterclaim
+      in a lawsuit) alleging that the Work or any Contribution embodied
+      within the Work constitutes direct or contributory patent infringement,
+      then any patent licenses granted to You under this License for that
+      Work shall terminate as of the date such litigation is filed.
+
+   4. Redistribution. You may reproduce and distribute copies of the
+      Work or Derivative Works thereof in any medium, with or without
+      modifications, and in Source or Object form, provided that You
+      meet the following conditions:
+
+      (a) You must give any other recipients of the Work or Derivative
+          Works a copy of this License; and
+
+      (b) You must cause any modified files to carry prominent notices
+          stating that You changed the files; and
+
+      (c) You must retain, in the Source form of any Derivative Works
+          that You distribute, all copyright, patent, trademark, and
+          attribution notices from the Source form of the Work,
+          excluding those notices that do not pertain to any part of
+          the Derivative Works; and
+
+      (d) If the Work includes a "NOTICE" text file as part of its
+          distribution, You must include a readable copy of the
+          attribution notices contained within such NOTICE file, in
+          at least one of the following places: within a NOTICE text file
+          distributed as part of the Derivative Works; within the Source
+          form or documentation, if provided along with the Derivative
+          Works; or, within a display generated by the Derivative Works,
+          if and wherever such third-party notices normally appear. The
+          contents of the NOTICE file are for informational purposes only
+          and do not modify the License. You may add Your own attribution
+          notices within Derivative Works that You distribute, alongside
+          or as an addendum to the NOTICE text from the Work, provided
+          that such additional attribution notices cannot be construed
+          as modifying the License.
+
+      You may add Your own license statement for Your modifications and
+      may provide additional grant of rights to use, copy, modify, merge,
+      publish, distribute, sublicense, and/or sell copies of the Work,
+      and to permit persons to whom the Work is furnished to do so,
+      subject to the following conditions:
+
+   5. Submission of Contributions. Unless You explicitly state otherwise,
+      any Contribution intentionally submitted for inclusion in the Work
+      by You to the Licensor shall be under the terms and conditions of
+      this License, without any additional terms or conditions.
+      Notwithstanding the above, nothing herein shall supersede or modify
+      the terms of any separate license agreement you may have executed
+      with Licensor regarding such Contributions.
+
+   6. Trademarks. This License does not grant permission to use the trade
+      names, trademarks, service marks, or product names of the Licensor,
+      except as required for reasonable and customary use in describing the
+      origin of the Work and reproducing the content of the NOTICE file.
+
+   7. Disclaimer of Warranty. Unless required by applicable law or
+      agreed to in writing, Licensor provides the Work (and each
+      Contributor provides its Contributions) on an "AS IS" BASIS,
+      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+      implied, including, without limitation, any conditions of TITLE,
+      MERCHANTIBILITY, or FITNESS FOR A PARTICULAR PURPOSE.
+      See the License for the specific language governing permissions and
+      limitations under the License.
+
+   8. Limitation of Liability. In no event and under no legal theory,
+      whether in tort (including negligence), contract, or otherwise,
+      unless required by applicable law (such as deliberate and grossly
+      negligent acts) or agreed to in writing, shall any Contributor be
+      liable to You for damages, including any direct, indirect, special,
+      incidental, or exemplary damages of any character arising as a
+      result of this License or out of the use or inability to use the
+      Work (even if such Contributor has been advised of the possibility
+      of such damages).
+
+   9. Accepting Warranty or Additional Liability. While redistributing
+      the Work or Derivative Works thereof, You may choose to offer,
+      and charge a fee for, acceptance of support, warranty, indemnity,
+      or other liability obligations and/or rights consistent with this
+      License. However, in accepting such obligations, You may offer only
+      conditions consistent with this License, subject to the following
+      additional conditions:
+
+      (a) You may offer additional warranty, indemnity, or other liability
+          terms and conditions; and
+
+      (b) You may offer support for the Work in return for a fee.
+
+   END OF TERMS AND CONDITIONS
+
+   APPENDIX: How to apply the Apache License to your work.
+
+      To apply the Apache License to your work, attach the following
+      boilerplate notice, with the fields enclosed by brackets "[]"
+      replaced with your own identifying information. (Don't include
+      the brackets!)  The text should be enclosed in the appropriate
+      comment syntax for the file format in question. It may also be
+      possible to make it available with a creative commons license.
+
+   Copyright 2026 Christian Manivong
+
+   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.

README.md

@@ -2,7 +2,7 @@
 
 NAPALM community driver for **OpenWrt** routers and access-points.
 
-Communicates over SSH using [Netmiko](https://github.com/ktbyers/netmiko) (`linux` device type).  
+Communicates over SSH using [Netmiko](https://pypi.org/project/netmiko/) (`linux` device type).  
 Requires OpenWrt **19.07** or newer.
 
 ## Tested devices
@@ -17,7 +17,7 @@ Requires OpenWrt **19.07** or newer.
 
 | Dependency | Minimum version |
 |---|---|
-| Python | 3.8 |
+| Python | 3.10 |
 | NAPALM | 4.0 |
 | Netmiko | 4.0 |
 
@@ -26,7 +26,7 @@ Requires OpenWrt **19.07** or newer.
 From source:
 
 ```bash
-git clone https://github.com/napalm-automation-community/napalm-openwrt
+git clone https://git.netork.io/NAPALM/napalm-openwrt
 cd napalm-openwrt
 pip install -e .
 ```
@@ -64,6 +64,7 @@ device.close()
 | `get_environment` | ✅ | CPU from `/proc/stat`, memory from `/proc/meminfo` |
 | `get_lldp_neighbors` | ✅ | Requires `lldpd` package installed on device |
 | `get_lldp_neighbors_detail` | ✅ | Requires `lldpd` package installed on device |
+| `get_listening_sockets` | ✅ | busybox `netstat -lntup`, procd service from the process's cgroup (napalm-device-types `ListeningSocketsMixin`); over an SSH exec channel |
 
 ## Configuration management
 
@@ -73,10 +74,12 @@ Configuration is managed via [UCI](https://openwrt.org/docs/guide-user/base-syst
 ### Merge candidate
 
 ```python
-device.load_merge_candidate(config="""
+device.load_merge_candidate(
+    config="""
 uci set system.@system[0].hostname='my-router'
 uci set network.lan.ipaddr='10.0.0.1'
-""")
+"""
+)
 
 print(device.compare_config())
 device.commit_config()

napalm_openwrt/config_mixin.py

@@ -1,4 +1,3 @@
-# -*- 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
@@ -50,7 +49,9 @@ class OpenWrtConfigMixin(ConfigLifecycleMixin):
         if self._candidate_config is None:
             raise MergeConfigException("No candidate configuration is staged.")
 
-        ex_cls = ReplaceConfigException if self._candidate_mode == "replace" else MergeConfigException
+        ex_cls = (
+            ReplaceConfigException if self._candidate_mode == "replace" else MergeConfigException
+        )
 
         self._backup_config = self._send_command("uci export")
 
@@ -90,7 +91,8 @@ class OpenWrtConfigMixin(ConfigLifecycleMixin):
         """
         if self._backup_config is None:
             raise CommandErrorException(
-                "No backup configuration available – commit_config has not been called in this session."
+                "No backup configuration available – "
+                "commit_config has not been called in this session."
             )
 
         escaped = self._backup_config.replace("'", "'\\''")

napalm_openwrt/interfaces_mixin.py

@@ -1,4 +1,3 @@
-# -*- 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
@@ -33,9 +32,8 @@ class OpenWrtInterfaceMixin:
             m = re.match(r"^\d+:\s+(\S+?)[@:]", line)
             if m:
                 name = m.group(1)
-                if (
+                if name not in self._EXCLUDED_INTERFACES and not name.startswith(
-                    name not in self._EXCLUDED_INTERFACES
+                    self._EXCLUDED_INTERFACE_PREFIXES
-                    and not name.startswith(self._EXCLUDED_INTERFACE_PREFIXES)
                 ):
                     interfaces.append(name)
         return sorted(set(interfaces))
@@ -368,12 +366,14 @@ class OpenWrtInterfaceMixin:
             except Exception:
                 mac_addr = mac_raw
 
-            table.append({
+            table.append(
-                "interface": interface,
+                {
-                "mac": mac_addr,
+                    "interface": interface,
-                "ip": ip_addr,
+                    "mac": mac_addr,
-                "age": -1.0,
+                    "ip": ip_addr,
-                "state": state,
+                    "age": -1.0,
-            })
+                    "state": state,
+                }
+            )
 
         return table

napalm_openwrt/lldp_mixin.py

@@ -1,4 +1,3 @@
-# -*- 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

napalm_openwrt/openwrt.py

@@ -1,4 +1,3 @@
-# -*- 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
@@ -19,28 +18,53 @@ Netmiko device_type: ``linux``
 
 from __future__ import annotations
 
+import ipaddress
 import re
-import socket
 from typing import Any
 
-from netmiko import ConnectHandler
-from netmiko.exceptions import NetmikoTimeoutException, NetmikoAuthenticationException
-
-from napalm_device_types import AccessPointDriver, FingerprintRule
 from napalm.base.exceptions import (
-    ConnectionException,
     ConnectionClosedException,
+    ConnectionException,
 )
 from napalm.base.netmiko_helpers import netmiko_args
+from napalm_device_types import (
+    AccessPointDriver,
+    ByteStream,
+    CommandResult,
+    EnsurePackageMixin,
+    FingerprintRule,
+    SnmpAgentMixin,
+)
+from napalm_device_types.channel import open_stream_on_transport, run_on_transport
+from napalm_device_types.listening import ListeningSocketsMixin
+from napalm_device_types.ssh_host_keys import HostKeyCheck, SshHostKeyMismatch
+from netmiko import ConnectHandler
+from netmiko.exceptions import NetmikoAuthenticationException, NetmikoTimeoutException
 
-from napalm_openwrt.interfaces_mixin import OpenWrtInterfaceMixin
-from napalm_openwrt.wireless_mixin import OpenWrtWirelessMixin
-from napalm_openwrt.lldp_mixin import OpenWrtLLDPMixin
-from napalm_openwrt.vlan_mixin import OpenWrtVLANMixin
 from napalm_openwrt.config_mixin import OpenWrtConfigMixin
-from napalm_openwrt.system_mixin import OpenWrtSystemMixin
+from napalm_openwrt.interfaces_mixin import OpenWrtInterfaceMixin
+from napalm_openwrt.lldp_mixin import OpenWrtLLDPMixin
 from napalm_openwrt.packages_mixin import OpenWrtPackageMixin
 from napalm_openwrt.routing_mixin import OpenWrtRoutingMixin
+from napalm_openwrt.system_mixin import _RC_MARKER, OpenWrtSystemMixin, _split_status
+from napalm_openwrt.vlan_mixin import OpenWrtVLANMixin
+from napalm_openwrt.wireless_mixin import OpenWrtWirelessMixin
+
+#: What may stand in a UCI value written in single quotes: no whitespace, quote or #.
+_SNMP_WORD = re.compile(r"^[^\s'\"#]+$")
+
+
+def _sources(settings: dict) -> list[str]:
+    """The allowed sources as networks, or ValueError."""
+    return [
+        str(ipaddress.ip_network(x.strip(), strict=False))
+        for x in settings.get("allowed_sources", [])
+    ]
+
+
+#: ``reboot`` detached from the session and two seconds late: the answer comes
+#: back first, and a hang-up from the closing session does not stop it.
+_REBOOT_DETACHED = "sh -c '(trap \"\" HUP; sleep 2; reboot) </dev/null >/dev/null 2>&1 &'"
 
 
 class OpenWrtDriver(
@@ -52,6 +76,9 @@ class OpenWrtDriver(
     OpenWrtSystemMixin,
     OpenWrtPackageMixin,
     OpenWrtRoutingMixin,
+    ListeningSocketsMixin,
+    SnmpAgentMixin,
+    EnsurePackageMixin,
     AccessPointDriver,
 ):
     """NAPALM driver for OpenWrt routers and access-points."""
@@ -87,6 +114,8 @@ class OpenWrtDriver(
 
         self.port = optional_args.pop("port", 22)
         self.netmiko_optional_args = netmiko_args(optional_args)
+        # The host key to expect, and where the one seen is reported (netork#1167).
+        self._host_key = HostKeyCheck(optional_args, hostname, self.port)
 
         # Config management state
         self._candidate_config: str | None = None
@@ -100,23 +129,23 @@ class OpenWrtDriver(
     def open(self) -> None:
         """Open an SSH connection to the device."""
         try:
-            self.device = ConnectHandler(
+            with self._host_key.netmiko(self.netmiko_optional_args) as netmiko_kwargs:
-                device_type=self.NETMIKO_DEVICE_TYPE,
+                self.device = ConnectHandler(
-                host=self.hostname,
+                    device_type=self.NETMIKO_DEVICE_TYPE,
-                username=self.username,
+                    host=self.hostname,
-                password=self.password,
+                    username=self.username,
-                timeout=self.timeout,
+                    password=self.password,
-                port=self.port,
+                    timeout=self.timeout,
-                **self.netmiko_optional_args,
+                    port=self.port,
-            )
+                    **netmiko_kwargs,
+                )
+        except SshHostKeyMismatch:
+            raise  # refused before the login; nothing was sent
         except NetmikoTimeoutException as exc:
-            raise ConnectionException(
+            raise ConnectionException(f"Cannot connect to {self.hostname}: {exc}") from exc
-                f"Cannot connect to {self.hostname}: {exc}"
-            ) from exc
         except NetmikoAuthenticationException as exc:
-            raise ConnectionException(
+            raise ConnectionException(f"Authentication failed for {self.hostname}: {exc}") from exc
-                f"Authentication failed for {self.hostname}: {exc}"
+        self._host_key.observe(self.device.remote_conn_pre.get_transport())
-            ) from exc
 
     def close(self) -> None:
         """Close the SSH connection."""
@@ -130,7 +159,7 @@ class OpenWrtDriver(
             return {"is_alive": False}
         try:
             return {"is_alive": self.device.remote_conn.transport.is_active()}
-        except (socket.error, EOFError, AttributeError):
+        except (OSError, EOFError, AttributeError):
             return {"is_alive": False}
 
     # ------------------------------------------------------------------
@@ -143,6 +172,7 @@ class OpenWrtDriver(
         When a list is supplied, commands are tried in order and the first
         one that does not return an error indicator is returned.
         """
+
         def _do_send(cmd: str) -> str:
             return self.device.send_command(
                 cmd,
@@ -158,9 +188,78 @@ class OpenWrtDriver(
                         return output
                 return output
             return _do_send(command)
-        except (socket.error, EOFError) as exc:
+        except (OSError, EOFError) as exc:
             raise ConnectionClosedException(str(exc)) from exc
 
+    # ------------------------------------------------------------------
+    # Command channel (napalm-device-types CommandChannelMixin, via AccessPointDriver)
+    # ------------------------------------------------------------------
+
+    def _transport(self) -> Any:
+        """The SSH transport netmiko already holds, for exec channels next to its shell."""
+        if self.device is None:
+            raise ConnectionClosedException("Not connected")
+        return self.device.remote_conn_pre.get_transport()
+
+    def _check_privilege(self, privileged: bool) -> None:
+        if privileged and self.username != "root":
+            raise PermissionError(
+                f"{self.hostname}: OpenWrt has no sudo; {self.username!r} cannot run as root"
+            )
+
+    def run_command(
+        self,
+        command: str,
+        *,
+        privileged: bool = False,
+        timeout: float = 60,
+        stdin: bytes | None = None,
+    ) -> CommandResult:
+        """Run *command* on an SSH exec channel: no PTY, stderr apart, a real exit code.
+
+        The login is root on OpenWrt as a rule, and there is no sudo: a
+        privileged command from any other login is refused before it runs.
+        """
+        self._check_privilege(privileged)
+        return run_on_transport(self._transport(), command, stdin=stdin, timeout=timeout)
+
+    def open_stream(self, command: str, *, privileged: bool = False) -> ByteStream:
+        """Start *command* on an SSH exec channel and return a stream to it."""
+        self._check_privilege(privileged)
+        return open_stream_on_transport(self._transport(), command)
+
+    def reboot_host(self) -> None:
+        """Restart the AP (``HostRebootMixin``); returns once the restart is under way.
+
+        Detached and two seconds late, so the answer comes back before the
+        session goes down; a refusal is raised, not mistaken for that.
+
+        :raises RuntimeError: for a login other than root, or a refused reboot.
+        """
+        try:
+            result = self.run_command(_REBOOT_DETACHED, privileged=True, timeout=30)
+        except PermissionError as exc:
+            raise RuntimeError(str(exc)) from exc
+        if result.exit_code:
+            reason = result.stderr.strip() or result.stdout.strip()
+            raise RuntimeError(reason or f"reboot exited with status {result.exit_code}")
+
+    def _run_listening_sockets_command(self, command: str, *, privileged: bool) -> str:
+        """Run the listening-socket command for ``ListeningSocketsMixin``.
+
+        Over an SSH exec channel of its own, not the interactive shell: busybox
+        ash cuts a typed line at 512 characters, and the command is longer. As
+        root it names every socket's process; a login other than root has no way
+        to become root here, so a privileged reading comes back empty and the
+        mixin reads again without attributing.
+        """
+        if privileged and self.username != "root":
+            return ""
+        _stdin, stdout, _stderr = self.device.remote_conn_pre.exec_command(
+            command, timeout=self.timeout
+        )
+        return stdout.read().decode("utf-8", "replace")
+
     @staticmethod
     def _parse_openwrt_release(output: str) -> dict[str, str]:
         """Parse ``/etc/openwrt_release`` key=value pairs."""
@@ -211,18 +310,14 @@ class OpenWrtDriver(
         if model_raw and not model_raw.startswith("cat: "):
             facts["model"] = model_raw.strip().replace("\x00", "")
         else:
-            facts["model"] = (
+            facts["model"] = release.get("DISTRIB_TARGET", "").replace("/", " ")
-                release.get("DISTRIB_TARGET", "").replace("/", " ")
-            )
 
         facts["os_version"] = release.get(
             "DISTRIB_DESCRIPTION",
             release.get("DISTRIB_RELEASE", "unknown"),
         )
         facts["serial_number"] = ""
-        facts["uptime"] = self._parse_uptime_seconds(
+        facts["uptime"] = self._parse_uptime_seconds(self._send_command("cat /proc/uptime"))
-            self._send_command("cat /proc/uptime")
-        )
 
         # hostname and FQDN
         uname_out = self._send_command("uname -a")
@@ -264,27 +359,15 @@ class OpenWrtDriver(
 
         return configs
 
-    def cli(
+    def cli(self, commands: list[str], encoding: str = "text") -> dict[str, str]:
-        self, commands: list[str], encoding: str = "text"
-    ) -> dict[str, str]:
         """Execute a list of CLI commands and return their output."""
-        return {
+        return {cmd: self._send_command(cmd) for cmd in commands}
-            cmd: self._send_command(cmd)
-            for cmd in commands
-        }
 
     # ------------------------------------------------------------------
     # Service management (forwarded to mixins because AccessPointDriver
     # defines NotImplementedError stubs for these)
     # ------------------------------------------------------------------
 
-
-
-
-
-
-
-
     # ------------------------------------------------------------------
     # UCI parsing helpers
     # ------------------------------------------------------------------
@@ -340,18 +423,6 @@ class OpenWrtDriver(
     # Device actions
     # ------------------------------------------------------------------
 
-    def run_device_action(self, action: str) -> dict[str, Any]:
-        """Execute a named action on the device."""
-        if action == "install_lldpd":
-            return self._action_install_lldpd()
-        if action == "install_auc":
-            return self._action_install_auc()
-        if action == "install_coreutils_base64":
-            return self._action_install_coreutils_base64()
-        if action == "fix_snmp":
-            return self._action_fix_snmp()
-        raise NotImplementedError(f"Unknown action: {action!r}")
-
     def _mgmt_address(self) -> str:
         """Return the device-side address of our own SSH session.
 
@@ -429,9 +500,7 @@ class OpenWrtDriver(
         return ""
 
     @classmethod
-    def _stale_snmp_rules(
+    def _stale_snmp_rules(cls, fw_sections: dict[str, dict[str, str]], keep: str) -> list[str]:
-        cls, fw_sections: dict[str, dict[str, str]], keep: str
-    ) -> list[str]:
         """Return every SNMP rule section except ``keep``, in delete-safe order."""
         hits = [
             key
@@ -469,43 +538,83 @@ class OpenWrtDriver(
                 return int(line)
         return 0
 
-    def _action_fix_snmp(self) -> dict[str, Any]:
+    # ------------------------------------------------------------------
-        """Ensure snmpd is running and reachable on UDP/161.
+    # SNMP agent (napalm-device-types SnmpAgentMixin): the caller's values
+    # ------------------------------------------------------------------
 
-        On OpenWrt the usual reason SNMP stays unreachable is the firewall:
+    def get_management_peer(self) -> str | None:
-        snmpd listens, but the zone covering the management interface
+        """The address the device sees the SSH connection come from (``SSH_CONNECTION``)."""
-        default-drops everything except SSH/HTTP/HTTPS/ICMP.  A rule only
+        parts = self._send_command("echo $SSH_CONNECTION").split()
-        lands in the right nftables chain when its ``src`` names the zone
+        try:
-        that actually owns the management network.
+            return str(ipaddress.ip_address(parts[0])) if parts else None
+        except ValueError:
+            return None
 
-        This action:
+    def _rc_step(self, command: str, done: str) -> dict[str, Any]:
+        output, status = _split_status(self._send_command(f"{command} 2>&1; echo {_RC_MARKER}$?"))
+        return {"ok": status == 0, "output": done if status == 0 else output.strip()[-300:]}
 
-        1. Resolves the management address from ``$SSH_CONNECTION``, maps it
+    def _install_snmp_agent(self) -> dict[str, Any]:
-           to its ``network`` section and from there to the owning firewall
+        if self._send_command("[ -x /usr/sbin/snmpd ] && echo present").strip() == "present":
-           zone — no guessing from rule names.
+            return {"ok": True, "output": "snmpd already installed"}
-        2. Aborts with a diagnosis when that zone has no ``option name``:
+        if self._pm_type() == "apk":
-           fw4 skips such a section, so any rule written against it is dead
+            return self._rc_step("apk add snmpd", "snmpd installed")
-           on arrival.  Naming the zone is left to the operator because an
+        return self._rc_step("opkg update >/dev/null 2>&1; opkg install snmpd", "snmpd installed")
-           inert zone becoming active changes what the AP filters.
+
-        3. Removes stale SNMP rules from earlier versions (highest anonymous
+    def _write_snmp_agent_config(self, settings: dict) -> dict[str, Any]:
-           index first) and commits the deletion.
+        """Community, sources, contact and location in UCI's snmpd config.
-        4. Writes the ``allow_snmp_from_<zone>`` rule in full every run, so a
+
-           half-written rule from a previous attempt gets repaired.
+        One ``com2sec`` (``com2sec6``) per allowed source, all mapped to the
-        5. Reloads the firewall and fails on any reload complaint.
+        read-only security name the stock groups use. Sections of an earlier run
-        6. Verifies on the device that snmpd listens and that a udp/161
+        (``snmp_src*``) go first. Without sources every address may ask; with only
-           accept rule is live in the packet filter.
+        IPv4 sources, IPv6 answers localhost only.
+        """
+        community = settings["community"]
+        if not _SNMP_WORD.match(community):
+            raise ValueError("the community must be one word without quotes or #")
+        sources = _sources(settings)
+        v4 = [x for x in sources if ":" not in x] or (["127.0.0.1/32"] if sources else ["default"])
+        v6 = [x for x in sources if ":" in x] or (["::1/128"] if sources else ["default"])
+        old = self._parse_uci_sections(self._send_command("uci show snmpd 2>/dev/null"))
+        stale = sorted((key for key in old if key.startswith("snmp_src")), reverse=True)
+        cmds = [f"uci -q delete snmpd.{key}" for key in stale]
+        for kind, first, nets in (("com2sec", "public", v4), ("com2sec6", "public6", v6)):
+            for i, net in enumerate(nets):
+                key = first if i == 0 else f"snmp_src{'6' if kind.endswith('6') else ''}_{i}"
+                cmds += [
+                    f"uci set snmpd.{key}={kind}",
+                    f"uci set snmpd.{key}.secname='ro'",
+                    f"uci set snmpd.{key}.source='{net}'",
+                    f"uci set snmpd.{key}.community='{community}'",
+                ]
+        for key, option in (("location", "sysLocation"), ("contact", "sysContact")):
+            value = settings.get(key, "")
+            if value:
+                if "'" in value or "\n" in value:
+                    raise ValueError(f"{key} must be one line without '")
+                cmds.append(f"uci set snmpd.@system[0].{option}='{value}'")
+        cmds.append("uci commit snmpd")
+        done = f"community set for {', '.join(sources) or 'every address'}"
+        return self._rc_step("; ".join(cmds), done)
+
+    def _allow_snmp_sources(self, settings: dict) -> dict[str, Any]:
+        """A udp/161 accept rule in the zone that owns the management address.
+
+        The zone is resolved from ``$SSH_CONNECTION`` → its ``network`` section →
+        the zone listing it, not guessed from rule names. A zone without ``option
+        name`` is skipped by fw4, so the step fails with how to fix it instead of
+        writing a dead rule. Stale SNMP rules go (highest anonymous index first);
+        the rule is written in full every run and limited to the allowed sources
+        (``src_ip``); the reload and the live packet filter decide success.
         """
         lines: list[str] = []
-
+        sources = _sources(settings)
-        # ── 1. Resolve the zone that owns the management address ─────────
+        fw_sections = self._parse_uci_sections(self._send_command("uci show firewall 2>/dev/null"))
-        fw_raw = self._send_command("uci show firewall 2>/dev/null")
-        fw_sections = self._parse_uci_sections(fw_raw)
         zones, nameless = self._firewall_zones(fw_sections)
-
         address = self._mgmt_address()
         net_sections = self._parse_uci_sections(self._send_command("uci show network 2>/dev/null"))
         mgmt_net = self._mgmt_network(net_sections, address)
-        lines.append(f"[firewall] Management address {address} on network {mgmt_net or '<unknown>'}")
+        lines.append(f"Management address {address} on network {mgmt_net or '<unknown>'}")
 
         broken_zone = (
             self._zone_for_network(fw_sections, {key: key for key in nameless}, mgmt_net)
@@ -514,50 +623,45 @@ class OpenWrtDriver(
         )
         if broken_zone:
             lines.append(
-                f"[firewall] Zone {broken_zone} owns network {mgmt_net!r} but has no "
+                f"Zone {broken_zone} owns network {mgmt_net!r} but has no 'name' option — "
-                f"'name' option — fw4 skips the section and drops every rule that "
+                "fw4 skips the section and drops every rule that references it. "
-                f"references it. Fix on the device, then re-run:"
+                "Fix on the device, then re-run:"
             )
             lines.append(
-                f"[firewall]   uci set firewall.{broken_zone}.name='{mgmt_net}' "
+                f"  uci set firewall.{broken_zone}.name='{mgmt_net}'"
-                f"&& uci commit firewall && fw4 reload"
+                " && uci commit firewall && fw4 reload"
             )
-            return {"success": False, "output": "\n".join(lines)}
+            return {"ok": False, "output": "\n".join(lines)}
 
         mgmt_zone = self._zone_for_network(fw_sections, zones, mgmt_net) if mgmt_net else ""
         if not mgmt_zone and "lan" in zones:
             mgmt_zone = "lan"
         if not mgmt_zone and len(zones) == 1:
             mgmt_zone = next(iter(zones))
-
         if mgmt_zone:
-            lines.append(f"[firewall] Management zone: {mgmt_zone!r}")
+            lines.append(f"Management zone: {mgmt_zone!r}")
         elif zones:
-            lines.append(
+            present = ", ".join(sorted(zones))
-                f"[firewall] No zone covers the management network — zones present: "
+            lines.append(f"No zone covers the management network — zones present: {present}")
-                f"{', '.join(sorted(zones))}"
+            return {"ok": False, "output": "\n".join(lines)}
-            )
-            return {"success": False, "output": "\n".join(lines)}
         else:
-            lines.append("[firewall] No zones configured — writing an unscoped rule")
+            lines.append("No zones configured — writing an unscoped rule")
 
-        # ── 2. Drop stale SNMP rules from earlier versions of this action ─
         named_key = f"allow_snmp_from_{mgmt_zone}" if mgmt_zone else "allow_snmp"
         stale = self._stale_snmp_rules(fw_sections, keep=named_key)
         if stale:
-            self._send_command(
+            deletes = "; ".join(f"uci -q delete firewall.{key}" for key in stale)
-                "; ".join(f"uci -q delete firewall.{key}" for key in stale)
+            self._send_command(f"{deletes}; uci commit firewall")
-                + "; uci commit firewall"
+            lines.append(f"Removed stale SNMP rule(s): {', '.join(stale)}")
-            )
-            lines.append(f"[firewall] Removed stale SNMP rule(s): {', '.join(stale)}")
 
-        # ── 3. Write the rule in full — repairs a half-written one ────────
         rule_cmds = [
             f"uci set firewall.{named_key}=rule",
-            f"uci set firewall.{named_key}.name='Allow-SNMP-netOrk'",
+            f"uci set firewall.{named_key}.name='Allow-SNMP'",
             f"uci set firewall.{named_key}.target='ACCEPT'",
             f"uci set firewall.{named_key}.proto='udp'",
             f"uci set firewall.{named_key}.dest_port='161'",
+            f"uci -q delete firewall.{named_key}.src_ip",
+            *(f"uci add_list firewall.{named_key}.src_ip='{net}'" for net in sources),
         ]
         if mgmt_zone:
             rule_cmds.append(f"uci set firewall.{named_key}.src='{mgmt_zone}'")
@@ -565,86 +669,33 @@ class OpenWrtDriver(
             rule_cmds.append(f"uci -q delete firewall.{named_key}.src")
         rule_cmds.append("uci commit firewall")
         rule_out = self._send_command("; ".join(rule_cmds) + " 2>&1").strip()
-        lines.append(f"[firewall] Wrote rule {named_key!r}: {rule_out or 'ok'}")
+        scope = ", ".join(sources) or "every source in the zone"
+        lines.append(f"Wrote rule {named_key!r} for {scope}: {rule_out or 'ok'}")
 
-        # ── 4. Reload the firewall, and believe what it says ──────────────
         reload_out = self._send_command(
             "fw4 reload 2>&1 || /etc/init.d/firewall reload 2>&1"
         ).strip()
         fw_ok = not self._fw_reload_failed(reload_out)
-        lines.append(f"[firewall] Reload: {reload_out or 'ok'}")
+        lines.append(f"Reload: {reload_out or 'ok'}")
         if not fw_ok:
-            lines.append("[firewall] Reload reported invalid sections — ruleset not applied")
+            lines.append("Reload reported invalid sections — ruleset not applied")
-
-        # ── 5. Ensure snmpd is enabled and running ────────────────────────
-        status = self._send_command("/etc/init.d/snmpd status 2>/dev/null")
-        if "running" not in status.lower() and "active" not in status.lower():
-            self._send_command(
-                "/etc/init.d/snmpd enable 2>/dev/null;"
-                " /etc/init.d/snmpd start 2>/dev/null"
-            )
-            lines.append("[snmpd] Service started and enabled")
-        else:
-            lines.append("[snmpd] Service already running")
-
-        # ── 6. Verify on the device instead of assuming success ───────────
-        listening = self._grep_count(
-            self._send_command("ss -lun 2>/dev/null | grep -c ':161'")
-        )
-        lines.append(f"[probe] snmpd listening on udp/161: {'yes' if listening else 'no'}")
-
         live = self._grep_count(
             self._send_command(
                 "{ nft list ruleset 2>/dev/null || iptables-save 2>/dev/null; }"
                 " | grep -c 'dport 161'"
             )
         )
-        lines.append(f"[probe] udp/161 accept rules live in the packet filter: {live}")
+        lines.append(f"udp/161 accept rules live in the packet filter: {live}")
+        return {"ok": fw_ok and bool(live), "output": "\n".join(lines)}
 
-        success = fw_ok and bool(listening) and bool(live)
+    def _start_snmp_agent(self) -> dict[str, Any]:
-        return {"success": success, "output": "\n".join(lines)}
-
-    def _action_install_coreutils_base64(self) -> dict[str, Any]:
-        """Install coreutils-base64 via the device package manager."""
-        pm = self._pm_type()
-        if pm == "apk":
-            raw = self._send_command("apk add coreutils-base64 2>&1")
-        else:
-            self._send_command("opkg update 2>&1")
-            raw = self._send_command("opkg install coreutils-base64 2>&1")
-        out = self._clean_pkg_output(raw)
-        low = out.lower()
-        success = not any(kw in low for kw in ("error:", "failed", "not found", "unknown package"))
-        return {"success": success, "output": out}
-
-    def _action_install_auc(self) -> dict[str, Any]:
-        """Install the attended sysupgrade client (auc) via opkg."""
-        self._send_command("opkg update 2>&1")
-        raw = self._send_command("opkg install auc 2>&1")
-        out = self._clean_pkg_output(raw)
-        low = out.lower()
-        success = not any(kw in low for kw in ("error:", "failed", "not found", "unknown package"))
-        return {"success": success, "output": out}
-
-    def _action_install_lldpd(self) -> dict[str, Any]:
-        """Install lldpd, add eth0 to its interface list and start the service."""
-        pm = self._pm_type()
-        if pm == "apk":
-            raw = self._send_command("apk add lldpd 2>&1")
-        else:
-            self._send_command("opkg update 2>&1")
-            raw = self._send_command("opkg install lldpd 2>&1")
-        out = self._clean_pkg_output(raw)
-
-        # Configure the correct management interface and enable the service
-        self._lldpd_fix_interface()
-
-        # Enable and start the service
         self._send_command(
-            "/etc/init.d/lldpd enable 2>/dev/null; "
+            "/etc/init.d/snmpd enable 2>/dev/null; /etc/init.d/snmpd restart 2>/dev/null"
-            "/etc/init.d/lldpd start 2>/dev/null"
         )
-
+        listening = self._grep_count(self._send_command("ss -lun 2>/dev/null | grep -c ':161'"))
-        low = out.lower()
+        return {
-        success = not any(kw in low for kw in ("error:", "failed", "not found", "unknown package"))
+            "ok": bool(listening),
-        return {"success": success, "output": out}
+            "output": "snmpd enabled and listening on udp/161"
+            if listening
+            else "snmpd is not listening on udp/161",
+        }

napalm_openwrt/packages_mixin.py

@@ -1,4 +1,3 @@
-# -*- 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
@@ -99,14 +98,16 @@ class OpenWrtPackageMixin:
                 name = token[: vm.start()] if vm else token
                 version = vm.group(1) if vm else ""
                 origin = ""
-            packages.append({
+            packages.append(
-                "name": name,
+                {
-                "version": version,
+                    "name": name,
-                "installed": True,
+                    "version": version,
-                "description": "",
+                    "installed": True,
-                "size": 0,
+                    "description": "",
-                "source": origin,
+                    "size": 0,
-            })
+                    "source": origin,
+                }
+            )
         return sorted(packages, key=lambda p: p["name"].lower())
 
     def _pm_type(self) -> str:
@@ -121,6 +122,7 @@ class OpenWrtPackageMixin:
         index is populated (OpenWrt stores it in RAM and loses it on reboot).
         """
         import shlex
+
         safe_q = shlex.quote(query)
         if self._pm_type() == "apk":
             # Refresh index (no-ops if already current, safe to run every time)
@@ -137,14 +139,16 @@ class OpenWrtPackageMixin:
                     name, version, description = m.group(1), m.group(2), (m.group(3) or "")
                 else:
                     name, version, description = line, "", ""
-                packages.append({
+                packages.append(
-                    "name": name,
+                    {
-                    "version": version,
+                        "name": name,
-                    "installed": name in installed,
+                        "version": version,
-                    "description": description,
+                        "installed": name in installed,
-                    "size": 0,
+                        "description": description,
-                    "source": "",
+                        "size": 0,
-                })
+                        "source": "",
+                    }
+                )
         else:
             # opkg lists live in /var/opkg-lists/ (RAM) — cleared on reboot
             self._send_command("opkg update 2>/dev/null || true")
@@ -159,14 +163,16 @@ class OpenWrtPackageMixin:
                 name = parts[0].strip()
                 version = parts[1].strip() if len(parts) > 1 else ""
                 description = parts[2].strip() if len(parts) > 2 else ""
-                packages.append({
+                packages.append(
-                    "name": name,
+                    {
-                    "version": version,
+                        "name": name,
-                    "installed": name in installed,
+                        "version": version,
-                    "description": description,
+                        "installed": name in installed,
-                    "size": 0,
+                        "description": description,
-                    "source": "",
+                        "size": 0,
-                })
+                        "source": "",
+                    }
+                )
         return packages
 
     @staticmethod
@@ -175,27 +181,29 @@ class OpenWrtPackageMixin:
         # Strip CSI sequences (\x1b[...X), OSC, charset designations, and
         # 2-byte DEC private sequences like ESC 7 (cursor save) / ESC 8 (restore)
         cleaned = re.sub(
-            r'\x1b(?:\[[0-9;?]*[a-zA-Z]|\][^\x07]*\x07|[()][0-9A-Za-z]|[\x30-\x7e])',
+            r"\x1b(?:\[[0-9;?]*[a-zA-Z]|\][^\x07]*\x07|[()][0-9A-Za-z]|[\x30-\x7e])",
-            '', raw,
+            "",
+            raw,
         )
         # After stripping cursor-save/restore sequences, apk progress updates
         # end up concatenated on a single line.  Strip those inline patterns.
-        cleaned = re.sub(r'\s*\d{1,3}%\s*#*', ' ', cleaned)
+        cleaned = re.sub(r"\s*\d{1,3}%\s*#*", " ", cleaned)
         lines = []
-        for segment in cleaned.split('\n'):
+        for segment in cleaned.split("\n"):
             # \r overwrites the line; keep only the portion after the last \r
-            part = segment.split('\r')[-1].strip()
+            part = segment.split("\r")[-1].strip()
             if not part:
                 continue
             # Drop pure progress-bar lines (only #, spaces, digits, %)
-            if re.match(r'^[#\s\d%]*$', part):
+            if re.match(r"^[#\s\d%]*$", part):
                 continue
             lines.append(part)
-        return '\n'.join(lines)
+        return "\n".join(lines)
 
     def install_package(self, name: str) -> dict[str, Any]:
         """Install a package by name. Returns ``{"success": bool, "output": str}``."""
         import shlex
+
         safe_name = shlex.quote(name)
         if self._pm_type() == "apk":
             raw = self._send_command(f"apk add {safe_name} 2>&1")
@@ -209,6 +217,7 @@ class OpenWrtPackageMixin:
     def uninstall_package(self, name: str) -> dict[str, Any]:
         """Remove a package by name. Returns ``{"success": bool, "output": str}``."""
         import shlex
+
         safe_name = shlex.quote(name)
         if self._pm_type() == "apk":
             raw = self._send_command(f"apk del {safe_name} 2>&1")

napalm_openwrt/routing_mixin.py

@@ -1,4 +1,3 @@
-# -*- 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
@@ -145,7 +144,7 @@ class OpenWrtRoutingMixin:
             # "192.168.1.0/24 dev ..." → prefix as-is
             if line_s.startswith("default"):
                 prefix = "0.0.0.0/0"
-                rest = line_s[len("default"):].strip()
+                rest = line_s[len("default") :].strip()
             else:
                 parts = line_s.split()
                 prefix = parts[0]

napalm_openwrt/system_mixin.py

@@ -1,4 +1,3 @@
-# -*- 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
@@ -13,11 +12,11 @@
 
 from __future__ import annotations
 
+import logging
 import re
-import time as _time
 from typing import Any
 
-
+logger = logging.getLogger(__name__)
 
 #: Printed after a command so its exit status survives the shell session.
 _RC_MARKER = "__OWRT_RC="
@@ -32,8 +31,12 @@ def _split_status(raw: str) -> tuple[str, int | None]:
     output = _RC_RE.sub("", raw or "").strip()
     return output, int(statuses[-1]) if statuses else None
 
+
 class OpenWrtSystemMixin:
-    """Mixin providing system-level NAPALM getters (environment, NTP, SNMP, users, services, updates, device actions)."""
+    """Mixin providing system-level NAPALM getters.
+
+    Environment, NTP, SNMP, users, services, updates and device actions.
+    """
 
     def get_environment(self) -> dict[str, Any]:
         """Return device environment data (CPU, memory).
@@ -42,7 +45,8 @@ class OpenWrtSystemMixin:
         Memory from ``/proc/meminfo``.
         """
         cpu_out = self._send_command(
-            "awk '/^cpu /{idle1=$5; total1=$2+$3+$4+$5+$6+$7+$8} END{print (1-(idle1/total1))*100}' /proc/stat"
+            "awk '/^cpu /{idle1=$5; total1=$2+$3+$4+$5+$6+$7+$8} "
+            "END{print (1-(idle1/total1))*100}' /proc/stat"
         )
         mem_out = self._send_command("cat /proc/meminfo")
 
@@ -96,9 +100,7 @@ class OpenWrtSystemMixin:
         """
         sys_out = self._send_command("uci show system 2>/dev/null || true")
         db_out = self._send_command("uci show dropbear 2>/dev/null || true")
-        luci_out = self._send_command(
+        luci_out = self._send_command("/etc/init.d/uhttpd enabled 2>/dev/null && echo 1 || echo 0")
-            "/etc/init.d/uhttpd enabled 2>/dev/null && echo 1 || echo 0"
-        )
         net_out = self._send_command("uci show network 2>/dev/null || true")
 
         sys_cfg: dict[str, str] = {}
@@ -239,7 +241,8 @@ class OpenWrtSystemMixin:
         running = (
             self._send_command(
                 "/etc/init.d/snmpd running 2>/dev/null && echo active || echo inactive"
-            ).strip() == "active"
+            ).strip()
+            == "active"
         )
         if not running:
             return None
@@ -254,160 +257,10 @@ class OpenWrtSystemMixin:
             if uci_comm:
                 community = uci_comm
         except Exception:
-            pass
+            logger.debug("snmpd community: ignored", exc_info=True)
 
         return SNMPConfigDict(running=True, community=community, port=161, version="2c")
 
-    def _action_fix_snmp(self) -> dict[str, Any]:
-        """Install and configure snmpd on OpenWrt.
-
-        Installs snmpd-nossl (the daemon) and luci-app-snmpd (UCI schema +
-        proper procd init script).  Configures community 'public' via UCI.
-        """
-        lines: list = []
-
-        # 1. Install packages — snmpd-nossl (daemon) + luci-app-snmpd (UCI init)
-        pm = self._pm_type()
-        if pm == "apk":
-            raw = self._send_command("apk add snmpd-nossl luci-app-snmpd 2>&1")
-        else:
-            self._send_command("opkg update 2>/dev/null || true")
-            raw = self._send_command("opkg install snmpd-nossl luci-app-snmpd 2>&1")
-        out = self._clean_pkg_output(raw)
-        low = out.lower()
-        installed = not any(kw in low for kw in ("error:", "failed"))
-        lines.append(f"[install] {out[-300:]}")
-
-        if not installed and "already installed" not in low:
-            return {"success": False, "output": "\n".join(lines)}
-
-        # 2. Configure via UCI — modify the existing default sections only.
-        #    Do NOT create new named sections (causes duplicate directives in
-        #    the generated /var/run/snmpd.conf which crashes snmpd).
-        #    Also remove any stale named sections from previous fix attempts.
-        # The init script reads these UCI field names to generate /var/run/snmpd.conf:
-        #   agent:  agentaddress
-        #   com2sec: secname, source, community
-        #   group:  group (name!), version, secname
-        #   view:   viewname (not name!), type, oid
-        #   access: group, version, level, prefix, read, write, notify
-        # Default luci-app-snmpd schema uses different field names for group/view/access,
-        # so we patch all required fields explicitly.
-        uci_cmds = [
-            # Remove any stale named sections from previous runs
-            "uci -q delete snmpd.agent",
-            "uci -q delete snmpd.public",
-            # agent
-            "uci set snmpd.@agent[0].agentaddress='161'",
-            # com2sec
-            "uci set snmpd.@com2sec[0].secname='ro'",
-            "uci set snmpd.@com2sec[0].source='0.0.0.0/0'",
-            "uci set snmpd.@com2sec[0].community='public'",
-            # group — init script reads field 'group' (not 'name')
-            "uci set snmpd.@group[0].group='rogroup'",
-            "uci set snmpd.@group[0].version='v2c'",
-            "uci set snmpd.@group[0].secname='ro'",
-            # view — init script reads field 'viewname' (not 'name')
-            "uci set snmpd.@view[0].viewname='all'",
-            "uci set snmpd.@view[0].type='included'",
-            "uci set snmpd.@view[0].oid='.1'",
-            # access — init script needs write + notify or it returns early
-            "uci set snmpd.@access[0].group='rogroup'",
-            "uci set snmpd.@access[0].context='none'",
-            "uci set snmpd.@access[0].version='v2c'",
-            "uci set snmpd.@access[0].level='noAuthNoPriv'",
-            "uci set snmpd.@access[0].prefix='exact'",
-            "uci set snmpd.@access[0].read='all'",
-            "uci set snmpd.@access[0].write='none'",
-            "uci set snmpd.@access[0].notify='none'",
-            "uci commit snmpd",
-        ]
-        for cmd in uci_cmds:
-            self._send_command(f"{cmd} 2>/dev/null || true")
-        lines.append("[config] Configured snmpd via UCI (all required fields set).")
-
-        # 3. Firewall: allow UDP 161 from the management zone
-        try:
-            # Use `ss` (always available on OpenWrt) to find the IP of the current SSH client
-            raw_conn = self._send_command(
-                "ss -tnp 2>/dev/null | awk '/ESTAB.*:22/{print $5}' | head -1 | sed 's/:[0-9]*$//'"
-            ).strip()
-            # Strip IPv6-mapped prefix if present
-            if raw_conn.startswith("::ffff:"):
-                raw_conn = raw_conn[7:]
-            peer_ip = raw_conn.splitlines()[-1].strip() if raw_conn else ""
-
-            if peer_ip and peer_ip not in ("", "0.0.0.0", "::"):
-                # Determine which firewall zone owns the interface that routes to peer_ip.
-                # Walk uci firewall zones: find the zone whose associated network interface
-                # has an address in the same /24 as peer_ip.
-                peer_prefix = peer_ip.rsplit(".", 1)[0] if "." in peer_ip else ""
-                fw_out = self._send_command("uci show firewall 2>/dev/null || true")
-                src_zone = "*"
-                # Build map zone_name → zone_uci_key
-                zone_map: dict[str, str] = {}
-                for fw_line in fw_out.splitlines():
-                    m = re.match(r"firewall\.(\w+)\.name='([^']+)'", fw_line.strip())
-                    if m:
-                        zone_map[m.group(2)] = m.group(1)
-                # For each zone find its network, then the interface IP
-                for zone_name, uci_key in zone_map.items():
-                    net_line = self._send_command(
-                        f"uci -q get firewall.{uci_key}.network 2>/dev/null || true"
-                    ).strip()
-                    if not net_line:
-                        continue
-                    for net_name in net_line.split():
-                        iface_ip = self._send_command(
-                            f"uci -q get network.{net_name}.ipaddr 2>/dev/null || true"
-                        ).strip()
-                        if iface_ip and "." in iface_ip:
-                            iface_prefix = iface_ip.split("/")[0].rsplit(".", 1)[0]
-                            if peer_prefix and iface_prefix == peer_prefix:
-                                src_zone = zone_name
-                                break
-                    if src_zone != "*":
-                        break
-
-                self._send_command(
-                    f"uci -q delete firewall.snmp_netork 2>/dev/null; "
-                    f"uci set firewall.snmp_netork=rule; "
-                    f"uci set firewall.snmp_netork.name='Allow-SNMP-from-mgmt'; "
-                    f"uci set firewall.snmp_netork.src='{src_zone}'; "
-                    f"uci set firewall.snmp_netork.dest_port='161'; "
-                    f"uci set firewall.snmp_netork.proto='udp'; "
-                    f"uci set firewall.snmp_netork.target='ACCEPT'; "
-                    f"uci commit firewall; "
-                    f"/etc/init.d/firewall reload 2>/dev/null || true"
-                )
-                lines.append(f"[firewall] Added UDP:161 allow rule (src zone: {src_zone}).")
-            else:
-                lines.append("[firewall] Could not detect peer IP via ss — skipping firewall step.")
-        except Exception as exc:
-            lines.append(f"[firewall] skipped — {exc}")
-
-        # 4. Break any crash-loop, then start cleanly
-        self._send_command("/etc/init.d/snmpd stop 2>/dev/null; true")
-        _time.sleep(2)
-        self._send_command("pkill -9 snmpd 2>/dev/null; true")  # kill crash-loop zombie
-        _time.sleep(3)
-        self._send_command("/etc/init.d/snmpd enable 2>/dev/null; true")
-        self._send_command("/etc/init.d/snmpd start 2>/dev/null; true")
-        _time.sleep(4)
-        lines.append("[service] snmpd started via procd.")
-
-        # 5. Check if snmpd is now active (no local snmpget on OpenWrt by default)
-        status = self._send_command(
-            "/etc/init.d/snmpd running 2>/dev/null && echo active || echo inactive"
-        ).strip()
-        success = status == "active"
-        if success:
-            lines.append("[ok] snmpd is active.")
-        else:
-            lines.append(f"[warn] snmpd status: {status}")
-
-        return {"success": success, "output": "\n".join(lines)}
-
     def get_users(self) -> dict[str, Any]:
         """Return users configured on the device.
 
@@ -433,7 +286,13 @@ class OpenWrtSystemMixin:
             if len(parts) < 7:
                 continue
             username, password_hash, uid_str, _, _, home, shell = (
-                parts[0], parts[1], parts[2], parts[3], parts[4], parts[5], parts[6],
+                parts[0],
+                parts[1],
+                parts[2],
+                parts[3],
+                parts[4],
+                parts[5],
+                parts[6],
             )
             if shell not in valid_shells:
                 continue
@@ -476,9 +335,7 @@ class OpenWrtSystemMixin:
         import json as _json
 
         # -- enabled set: names from /etc/rc.d/S<priority><name> symlinks ----
-        rc_out = self._send_command(
+        rc_out = self._send_command("ls /etc/rc.d/ 2>/dev/null | grep '^S' | sed 's/^S[0-9]*//'")
-            "ls /etc/rc.d/ 2>/dev/null | grep '^S' | sed 's/^S[0-9]*//'"
-        )
         enabled: set[str] = {s.strip() for s in rc_out.splitlines() if s.strip()}
 
         # -- running info from procd via ubus ---------------------------------
@@ -496,9 +353,7 @@ class OpenWrtSystemMixin:
                 continue
             instances = svc_info.get("instances", {})
             running = any(
-                inst.get("running", False)
+                inst.get("running", False) for inst in instances.values() if isinstance(inst, dict)
-                for inst in instances.values()
-                if isinstance(inst, dict)
             )
             pid = next(
                 (
@@ -522,90 +377,28 @@ class OpenWrtSystemMixin:
         result: list[dict[str, Any]] = []
         for name in sorted(all_names):
             info = service_map.get(name, {})
-            result.append({
+            result.append(
-                "name": name,
+                {
-                "running": info.get("running", False),
+                    "name": name,
-                "enabled": name in enabled,
+                    "running": info.get("running", False),
-                "pid": info.get("pid", 0),
+                    "enabled": name in enabled,
-            })
+                    "pid": info.get("pid", 0),
+                }
+            )
 
         return result
 
     def manage_service(self, name: str, action: str) -> dict[str, Any]:
         """Execute a lifecycle action (start/stop/restart/enable/disable) on a service."""
         import re as _re
-        if not _re.match(r'^[a-zA-Z0-9_\-]+$', name):
+
+        if not _re.match(r"^[a-zA-Z0-9_\-]+$", name):
             raise ValueError(f"Invalid service name: {name!r}")
-        if action not in ('start', 'stop', 'restart', 'enable', 'disable'):
+        if action not in ("start", "stop", "restart", "enable", "disable"):
             raise ValueError(f"Invalid action: {action!r}")
         output = self._send_command(f"/etc/init.d/{name} {action} 2>&1")
         return {"success": True, "output": output}
 
-    def get_device_warnings(self) -> list[dict[str, Any]]:
-        """Return a list of warning dicts for issues detected on this device.
-
-        Currently detects:
-        - lldpd not installed (LLDP neighbor discovery unavailable)
-        - package updates available (uses local package cache, no network call)
-        - update notifications not configured (auc not installed, opkg only)
-        """
-        warnings: list[dict[str, Any]] = []
-
-        # 1. LLDP daemon
-        lldpd_path = self._send_command("which lldpd 2>/dev/null").strip()
-        if not lldpd_path:
-            warnings.append({"code": "lldpd_not_installed"})
-
-        pm = self._pm_type()
-
-        # 2. Package updates available (local cache only – no opkg update)
-        def _pkg_name(line: str, pm_type: str) -> str:
-            """Extract just the package name from an upgradable line.
-
-            apk:  'luci-app-firewall-26.143~abc < 26.151~def' → 'luci-app-firewall'
-            opkg: 'luci-app-firewall - 1.0 - 1.1'             → 'luci-app-firewall'
-            """
-            import re as _re
-            if pm_type == "apk":
-                # Strip trailing ' < ...' then remove the version suffix (-\d...)
-                name_ver = line.split(" ")[0]
-                m = _re.match(r'^(.*?)-\d', name_ver)
-                return m.group(1) if m else name_ver
-            else:
-                return line.split(" - ")[0].strip()
-
-        try:
-            if pm == "apk":
-                raw_upg = self._send_command("apk version 2>/dev/null | grep '<'")
-            else:
-                raw_upg = self._send_command("opkg list-upgradable 2>/dev/null")
-            upgradable = [ln.strip() for ln in raw_upg.splitlines() if ln.strip()]
-        except Exception:
-            upgradable = []
-
-        if upgradable:
-            pkg_names = [_pkg_name(ln, pm) for ln in upgradable]
-            warnings.append({
-                "code": "updates_available",
-                "meta": {
-                    "count": len(upgradable),
-                    "packages": pkg_names[:10],
-                },
-            })
-
-        # 3. Attended sysupgrade client not installed (opkg systems only)
-        if pm == "opkg":
-            auc_path = self._send_command("which auc 2>/dev/null").strip()
-            if not auc_path:
-                warnings.append({"code": "update_notifications_disabled"})
-
-        # 4. base64 not available — needed for efficient config apply
-        b64_path = self._send_command("command -v base64 2>/dev/null").strip()
-        if not b64_path:
-            warnings.append({"code": "no_base64"})
-
-        return warnings
-
     def get_available_updates(self) -> list[dict[str, Any]]:
         """Return list of upgradable packages from the local package manager cache.
 
@@ -633,11 +426,13 @@ class OpenWrtSystemMixin:
             # Output format: "pkgname-current_ver < new_ver"
             m = _re.match(r"^(.+)-(\d\S*)\s+<\s+(\S+)", line.strip())
             if m:
-                updates.append({
+                updates.append(
-                    "name": m.group(1),
+                    {
-                    "current_version": m.group(2),
+                        "name": m.group(1),
-                    "new_version": m.group(3),
+                        "current_version": m.group(2),
-                })
+                        "new_version": m.group(3),
+                    }
+                )
         return updates
 
     def _updates_opkg(self) -> list[dict[str, Any]]:
@@ -652,11 +447,13 @@ class OpenWrtSystemMixin:
             # opkg output: "pkgname - current_ver - new_ver"
             parts = [p.strip() for p in line.split(" - ")]
             if len(parts) == 3:
-                updates.append({
+                updates.append(
-                    "name": parts[0],
+                    {
-                    "current_version": parts[1],
+                        "name": parts[0],
-                    "new_version": parts[2],
+                        "current_version": parts[1],
-                })
+                        "new_version": parts[2],
+                    }
+                )
         return updates
 
     def refresh_available_updates(self) -> dict[str, Any]:
@@ -668,8 +465,9 @@ class OpenWrtSystemMixin:
     def apply_updates(self, packages: list[str]) -> dict[str, Any]:
         """Upgrade the given packages using the device's package manager."""
         import re as _re
+
         for pkg in packages:
-            if not _re.match(r'^[a-zA-Z0-9_\-\+\.]+$', pkg):
+            if not _re.match(r"^[a-zA-Z0-9_\-\+\.]+$", pkg):
                 raise ValueError(f"Invalid package name: {pkg!r}")
         pm = self._pm_type()
         pkg_args = " ".join(packages)
@@ -751,19 +549,21 @@ class OpenWrtSystemMixin:
             if len(parts) < 10:
                 continue
             try:
-                stats.append({
+                stats.append(
-                    "remote": parts[0],
+                    {
-                    "referenceid": parts[1],
+                        "remote": parts[0],
-                    "synchronized": tally == "*",
+                        "referenceid": parts[1],
-                    "stratum": int(parts[2]),
+                        "synchronized": tally == "*",
-                    "type": parts[3],
+                        "stratum": int(parts[2]),
-                    "when": parts[4],
+                        "type": parts[3],
-                    "hostpoll": int(parts[5]),
+                        "when": parts[4],
-                    "reachability": int(parts[6], 8),  # octal
+                        "hostpoll": int(parts[5]),
-                    "delay": float(parts[7]),
+                        "reachability": int(parts[6], 8),  # octal
-                    "offset": float(parts[8]),
+                        "delay": float(parts[7]),
-                    "jitter": float(parts[9]),
+                        "offset": float(parts[8]),
-                })
+                        "jitter": float(parts[9]),
+                    }
+                )
             except (ValueError, IndexError):
                 continue
         return stats
@@ -780,19 +580,23 @@ class OpenWrtSystemMixin:
                 continue
             tally = m.group(1)
             try:
-                stats.append({
+                stats.append(
-                    "remote": m.group(2),
+                    {
-                    "referenceid": "",
+                        "remote": m.group(2),
-                    "synchronized": tally == "^*",
+                        "referenceid": "",
-                    "stratum": int(m.group(3)),
+                        "synchronized": tally == "^*",
-                    "type": "u",
+                        "stratum": int(m.group(3)),
-                    "when": m.group(6),
+                        "type": "u",
-                    "hostpoll": int(m.group(4)),
+                        "when": m.group(6),
-                    "reachability": int(m.group(5), 8) if re.match(r"^[0-7]+$", m.group(5)) else 0,
+                        "hostpoll": int(m.group(4)),
-                    "delay": 0.0,
+                        "reachability": int(m.group(5), 8)
-                    "offset": 0.0,
+                        if re.match(r"^[0-7]+$", m.group(5))
-                    "jitter": 0.0,
+                        else 0,
-                })
+                        "delay": 0.0,
+                        "offset": 0.0,
+                        "jitter": 0.0,
+                    }
+                )
             except (ValueError, IndexError):
                 continue
         return stats

napalm_openwrt/vlan_mixin.py

@@ -1,4 +1,3 @@
-# -*- 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
@@ -172,7 +171,7 @@ class OpenWrtVLANMixin:
             vm = re.match(r"^([\w-]+)\.(\d+)$", iface)  # allow hyphens (br-ap)
             if not vm:
                 continue
-            parent = vm.group(1)          # e.g. "eth0" or "br-ap"
+            parent = vm.group(1)  # e.g. "eth0" or "br-ap"
             vlan_id = str(int(vm.group(2)))  # e.g. "10"
             vlans.setdefault(vlan_id, {"name": "", "tagged": [], "untagged": []})
             # sub-interface itself -> untagged egress
@@ -210,11 +209,13 @@ class OpenWrtVLANMixin:
             # No explicit bridge-vlan section -- nothing to delete via UCI
             return
 
-        self.cli([
+        self.cli(
-            f"uci delete network.@bridge-vlan[{idx}]",
+            [
-            "uci commit network",
+                f"uci delete network.@bridge-vlan[{idx}]",
-            "/etc/init.d/network reload",
+                "uci commit network",
-        ])
+                "/etc/init.d/network reload",
+            ]
+        )
 
     def get_network_instances(self, name: str = "") -> dict[str, Any]:
         """Return network instances (Linux network namespaces + default).

napalm_openwrt/wireless_mixin.py

@@ -1,4 +1,3 @@
-# -*- 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
@@ -13,9 +12,12 @@
 
 from __future__ import annotations
 
+import logging
 import re
 from typing import Any
 
+logger = logging.getLogger(__name__)
+
 
 class OpenWrtWirelessMixin:
     """Mixin providing wireless/radio NAPALM getters."""
@@ -101,8 +103,12 @@ class OpenWrtWirelessMixin:
             return ""
 
         _ENC_MAP = {
-            "": "Open", "none": "Open", "0": "Open",
+            "": "Open",
-            "wep": "WEP", "wep-open": "WEP (Open)", "wep-shared": "WEP (Shared)",
+            "none": "Open",
+            "0": "Open",
+            "wep": "WEP",
+            "wep-open": "WEP (Open)",
+            "wep-shared": "WEP (Shared)",
             "psk": "WPA-PSK",
             "psk+ccmp": "WPA-PSK",
             "psk-mixed": "WPA/WPA2-PSK",
@@ -152,7 +158,8 @@ class OpenWrtWirelessMixin:
                 if vlan is not None:
                     net_vlan[iface] = vlan
         except Exception:
-            pass  # Non-fatal: VLAN info is optional enrichment
+            # Non-fatal: VLAN info is optional enrichment.
+            logger.debug("network VLANs: ignored", exc_info=True)
 
         # Build result; group entries with the same SSID name, merging bands
         result: dict[str, Any] = {}
@@ -176,11 +183,17 @@ class OpenWrtWirelessMixin:
             ft_over_ds = entry.get("ft_over_ds", "1") == "1"
             client_isolation = entry.get("isolate", "0") == "1"
             _max_raw = entry.get("maxassoc")
-            max_clients: int | None = int(_max_raw) if _max_raw and str(_max_raw).isdigit() else None
+            max_clients: int | None = (
+                int(_max_raw) if _max_raw and str(_max_raw).isdigit() else None
+            )
             _disassoc_raw = entry.get("disassoc_low_ack")
-            disassoc_low_ack: bool | None = (_disassoc_raw == "1") if _disassoc_raw is not None else None
+            disassoc_low_ack: bool | None = (
+                (_disassoc_raw == "1") if _disassoc_raw is not None else None
+            )
             _max_inact_raw = entry.get("max_inactivity")
-            max_inactivity: int | None = int(_max_inact_raw) if _max_inact_raw and str(_max_inact_raw).isdigit() else None
+            max_inactivity: int | None = (
+                int(_max_inact_raw) if _max_inact_raw and str(_max_inact_raw).isdigit() else None
+            )
             key: str = entry.get("key", "") or ""
             acl_mode = _ACL_MODE_MAP.get(entry.get("macfilter", ""), "off")
             mac_list = sorted(set(iface_maclists.get(idx, [])))
@@ -331,16 +344,18 @@ class OpenWrtWirelessMixin:
         def _flush() -> None:
             if current_station and current_station.get("mac"):
                 info = ap_ifaces.get(active_iface, {})
-                results.append(WirelessClientDict(
+                results.append(
-                    mac=current_station["mac"],
+                    WirelessClientDict(
-                    ssid=info.get("ssid", ""),
+                        mac=current_station["mac"],
-                    radio=info.get("band") or info.get("radio", ""),
+                        ssid=info.get("ssid", ""),
-                    signal=current_station.get("signal", 0),
+                        radio=info.get("band") or info.get("radio", ""),
-                    noise=0,
+                        signal=current_station.get("signal", 0),
-                    tx_rate=current_station.get("tx_rate", 0.0),
+                        noise=0,
-                    rx_rate=current_station.get("rx_rate", 0.0),
+                        tx_rate=current_station.get("tx_rate", 0.0),
-                    uptime=current_station.get("uptime", 0),
+                        rx_rate=current_station.get("rx_rate", 0.0),
-                ))
+                        uptime=current_station.get("uptime", 0),
+                    )
+                )
 
         for line in station_out.splitlines():
             stripped = line.strip()
@@ -410,7 +425,6 @@ class OpenWrtWirelessMixin:
         Keys are interface names (e.g. ``"wlan0"``).  Values are lists of
         :class:`~napalm_device_types.models.ChannelScanEntryDict`.
         """
-        from napalm_device_types.models import ChannelScanEntryDict
 
         # Step 1: discover interfaces and filter to AP mode only
         iw_out = self._send_command("iw dev 2>/dev/null || true")
@@ -435,10 +449,7 @@ class OpenWrtWirelessMixin:
         if current_iface:
             iface_order.append((current_iface, current_type))
 
-        ap_ifaces = [
+        ap_ifaces = [name for name, itype in iface_order if itype.upper() in ("AP", "AP/VLAN")]
-            name for name, itype in iface_order
-            if itype.upper() in ("AP", "AP/VLAN")
-        ]
 
         if not ap_ifaces:
             return {}
@@ -489,15 +500,17 @@ class OpenWrtWirelessMixin:
                 # If width is still 0 try to derive from HT/VHT/HE info
                 if width == 0 and current.get("_ht_width"):
                     width = current["_ht_width"]
-                entries.append(ChannelScanEntryDict(
+                entries.append(
-                    bssid=current["bssid"],
+                    ChannelScanEntryDict(
-                    ssid=current.get("ssid", ""),
+                        bssid=current["bssid"],
-                    frequency=freq,
+                        ssid=current.get("ssid", ""),
-                    channel=ch,
+                        frequency=freq,
-                    signal_dbm=current.get("signal_dbm", 0),
+                        channel=ch,
-                    channel_width=width,
+                        signal_dbm=current.get("signal_dbm", 0),
-                    band=_freq_to_band(freq),
+                        channel_width=width,
-                ))
+                        band=_freq_to_band(freq),
+                    )
+                )
 
         for line in scan_out.splitlines():
             stripped = line.strip()
@@ -663,10 +676,22 @@ class OpenWrtWirelessMixin:
 
             # Derive channel_width from htmode string (e.g. VHT80 → 80 MHz)
             _HTMODE_WIDTH = {
-                "HT20": 20, "HT40": 40,
+                "HT20": 20,
-                "VHT20": 20, "VHT40": 40, "VHT80": 80, "VHT80+80": 80, "VHT160": 160,
+                "HT40": 40,
-                "HE20": 20, "HE40": 40, "HE80": 80, "HE160": 160,
+                "VHT20": 20,
-                "EHT20": 20, "EHT40": 40, "EHT80": 80, "EHT160": 160, "EHT320": 320,
+                "VHT40": 40,
+                "VHT80": 80,
+                "VHT80+80": 80,
+                "VHT160": 160,
+                "HE20": 20,
+                "HE40": 40,
+                "HE80": 80,
+                "HE160": 160,
+                "EHT20": 20,
+                "EHT40": 40,
+                "EHT80": 80,
+                "EHT160": 160,
+                "EHT320": 320,
             }
             channel_width = _HTMODE_WIDTH.get(htmode.upper(), 0)
 
@@ -677,7 +702,7 @@ class OpenWrtWirelessMixin:
                     channel=channel,
                     channel_width=channel_width,
                     tx_power=tx_power,
-                    frequency=0.0,    # enriched below via iwinfo
+                    frequency=0.0,  # enriched below via iwinfo
                 ),
                 "htmode": htmode,
                 "country": country,

pyproject.toml

@@ -1,23 +1,20 @@
 [build-system]
-requires = ["setuptools>=64", "wheel"]
+requires = ["setuptools>=77"]
 build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-openwrt"
-version = "0.1.0"
+version = "0.7.0"
 description = "NAPALM driver for OpenWrt routers/access-points"
 readme = "README.md"
-license = { text = "Apache-2.0" }
+license = "Apache-2.0"
-requires-python = ">=3.8"
+requires-python = ">=3.10"
 authors = [
     { name = "Christian Manivong" },
 ]
 classifiers = [
     "Topic :: Utilities",
-    "License :: OSI Approved :: Apache Software License",
     "Programming Language :: Python :: 3",
-    "Programming Language :: Python :: 3.8",
-    "Programming Language :: Python :: 3.9",
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
@@ -26,30 +23,71 @@ classifiers = [
 ]
 dependencies = [
     "napalm>=4.0.0",
-    "napalm_device_types>=0.1.0",
+    "napalm_device_types>=4.21.1",
     "netmiko>=4.0.0",
     "paramiko>=5.0.0",  # CVE-2026-44405
-    "netaddr",
+    "netaddr>=0.7.19",  # IPAddress; below it the package has none (CI lowest job)
 ]
 
 [project.optional-dependencies]
-dev = [
+dev = ["pytest>=8", "pytest-cov>=5", "mypy>=1.11", "ruff>=0.6"]
-    "pytest",
-    "pytest-cov",
-    "black",
-    "ruff",
-]
 
 [project.entry-points."napalm.drivers"]
 openwrt = "napalm_openwrt:OpenWrtDriver"
 
 [project.urls]
-Repository = "https://github.com/napalm-automation-community/napalm-openwrt"
+Repository = "https://git.netork.io/NAPALM/napalm-openwrt"
+Issues = "https://git.netork.io/NAPALM/napalm-openwrt/issues"
+Changelog = "https://git.netork.io/NAPALM/napalm-openwrt/src/branch/main/CHANGELOG.md"
 
 [tool.setuptools.packages.find]
 where = ["."]
 include = ["napalm_openwrt*"]
 
+[tool.setuptools.package-data]
+napalm_openwrt = ["py.typed"]
+
 [tool.ruff]
 line-length = 100
-target-version = "py38"
+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.mypy]
+python_version = "3.10"
+strict = true
+# napalm itself is untyped, so every driver subclasses Any.
+disallow_subclassing_any = false
+
+[[tool.mypy.overrides]]
+module = ["napalm.*", "netmiko.*", "paramiko.*", "netaddr.*"]
+ignore_missing_imports = true
+
+[[tool.mypy.overrides]]
+# Not yet under strict mypy (quality baseline); may only shrink.
+module = [
+    "napalm_openwrt.config_mixin",
+    "napalm_openwrt.interfaces_mixin",
+    "napalm_openwrt.lldp_mixin",
+    "napalm_openwrt.openwrt",
+    "napalm_openwrt.packages_mixin",
+    "napalm_openwrt.routing_mixin",
+    "napalm_openwrt.system_mixin",
+    "napalm_openwrt.vlan_mixin",
+    "napalm_openwrt.wireless_mixin",
+]
+ignore_errors = true
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+
+[tool.coverage.run]
+source = ["napalm_openwrt"]
+
+[tool.coverage.report]
+fail_under = 65

tests/quality_baseline.json

@@ -0,0 +1,55 @@
+{
+ "coverage_floor": 65.0,
+ "function_complexity": {
+  "napalm_openwrt/config_mixin.py::OpenWrtConfigMixin.commit_config": 14,
+  "napalm_openwrt/interfaces_mixin.py::OpenWrtInterfaceMixin.get_arp_table": 13,
+  "napalm_openwrt/openwrt.py::OpenWrtDriver._allow_snmp_sources": 22,
+  "napalm_openwrt/openwrt.py::OpenWrtDriver._write_snmp_agent_config": 22,
+  "napalm_openwrt/packages_mixin.py::OpenWrtPackageMixin.search_packages": 12,
+  "napalm_openwrt/routing_mixin.py::OpenWrtRoutingMixin._parse_traceroute_output": 12,
+  "napalm_openwrt/routing_mixin.py::OpenWrtRoutingMixin.get_route_to": 12,
+  "napalm_openwrt/system_mixin.py::OpenWrtSystemMixin.get_services": 14,
+  "napalm_openwrt/system_mixin.py::OpenWrtSystemMixin.get_system_config": 16,
+  "napalm_openwrt/system_mixin.py::OpenWrtSystemMixin.get_users": 13,
+  "napalm_openwrt/vlan_mixin.py::OpenWrtVLANMixin.get_vlans": 32,
+  "napalm_openwrt/wireless_mixin.py::OpenWrtWirelessMixin.get_channel_scan": 12,
+  "napalm_openwrt/wireless_mixin.py::OpenWrtWirelessMixin.get_radio_status": 19,
+  "napalm_openwrt/wireless_mixin.py::OpenWrtWirelessMixin.get_ssids": 51,
+  "napalm_openwrt/wireless_mixin.py::OpenWrtWirelessMixin.get_wireless_clients": 27
+ },
+ "function_lines": {
+  "napalm_openwrt/interfaces_mixin.py::OpenWrtInterfaceMixin.get_arp_table": 85,
+  "napalm_openwrt/interfaces_mixin.py::OpenWrtInterfaceMixin.get_interfaces_ip": 55,
+  "napalm_openwrt/interfaces_mixin.py::OpenWrtInterfaceMixin.get_mac_address_table": 53,
+  "napalm_openwrt/openwrt.py::OpenWrtDriver._allow_snmp_sources": 90,
+  "napalm_openwrt/openwrt.py::OpenWrtDriver.get_facts": 53,
+  "napalm_openwrt/packages_mixin.py::OpenWrtPackageMixin.search_packages": 59,
+  "napalm_openwrt/routing_mixin.py::OpenWrtRoutingMixin._parse_traceroute_output": 67,
+  "napalm_openwrt/routing_mixin.py::OpenWrtRoutingMixin.get_route_to": 91,
+  "napalm_openwrt/system_mixin.py::OpenWrtSystemMixin.get_services": 61,
+  "napalm_openwrt/system_mixin.py::OpenWrtSystemMixin.get_snmp_information": 56,
+  "napalm_openwrt/system_mixin.py::OpenWrtSystemMixin.get_system_config": 92,
+  "napalm_openwrt/system_mixin.py::OpenWrtSystemMixin.get_users": 64,
+  "napalm_openwrt/vlan_mixin.py::OpenWrtVLANMixin.get_network_instances": 61,
+  "napalm_openwrt/vlan_mixin.py::OpenWrtVLANMixin.get_vlans": 106,
+  "napalm_openwrt/wireless_mixin.py::OpenWrtWirelessMixin._parse_iw_scan": 93,
+  "napalm_openwrt/wireless_mixin.py::OpenWrtWirelessMixin.get_radio_status": 133,
+  "napalm_openwrt/wireless_mixin.py::OpenWrtWirelessMixin.get_ssids": 240,
+  "napalm_openwrt/wireless_mixin.py::OpenWrtWirelessMixin.get_wireless_clients": 152
+ },
+ "mypy_relaxed": [
+  "napalm_openwrt.config_mixin",
+  "napalm_openwrt.interfaces_mixin",
+  "napalm_openwrt.lldp_mixin",
+  "napalm_openwrt.openwrt",
+  "napalm_openwrt.packages_mixin",
+  "napalm_openwrt.routing_mixin",
+  "napalm_openwrt.system_mixin",
+  "napalm_openwrt.vlan_mixin",
+  "napalm_openwrt.wireless_mixin"
+ ],
+ "python_file_lines": {},
+ "untyped_public": [
+  "napalm_openwrt/system_mixin.py::OpenWrtSystemMixin.get_snmp_config"
+ ]
+}

tests/unit/test_command_channel.py

@@ -0,0 +1,112 @@
+"""A public command channel (napalm-device-types ``CommandChannelMixin``, netOrk #1113).
+
+netOrk sent UCI scripts and the wifi channel through the private
+``_send_command``: an interactive shell with no exit code, where a file had to
+be typed in base64 chunks or ``printf`` lines. ``run_command`` runs on an SSH
+exec channel of its own, with stdin, stderr and the exit code -- a script is
+written with ``cat`` from stdin, and ``base64`` is needed for nothing.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+from napalm.base.exceptions import ConnectionClosedException
+from napalm_device_types import CommandResult
+
+from napalm_openwrt.openwrt import OpenWrtDriver
+
+
+def _driver(username: str = "root") -> OpenWrtDriver:
+    with patch("napalm_openwrt.openwrt.ConnectHandler"):
+        drv = OpenWrtDriver(hostname="192.0.2.15", username=username, password="")
+    drv.device = MagicMock()
+    return drv
+
+
+def test_a_command_runs_on_the_ssh_transport_with_its_stdin():
+    drv = _driver()
+    transport = drv.device.remote_conn_pre.get_transport.return_value
+    done = CommandResult("", "", 0)
+    with patch("napalm_openwrt.openwrt.run_on_transport", return_value=done) as run:
+        result = drv.run_command("cat > /tmp/x.sh", stdin=b"uci commit\n", timeout=30)
+
+    assert result is done
+    run.assert_called_once_with(transport, "cat > /tmp/x.sh", stdin=b"uci commit\n", timeout=30)
+
+
+def test_root_runs_privileged_commands_as_they_are():
+    drv = _driver()
+    with patch("napalm_openwrt.openwrt.run_on_transport") as run:
+        drv.run_command("wifi", privileged=True)
+
+    assert run.call_args.args[1] == "wifi"
+
+
+def test_a_login_other_than_root_cannot_gain_root():
+    """OpenWrt has no sudo: refused before anything runs, not half done."""
+    drv = _driver(username="admin")
+    with patch("napalm_openwrt.openwrt.run_on_transport") as run, pytest.raises(PermissionError):
+        drv.run_command("wifi", privileged=True)
+
+    run.assert_not_called()
+
+
+def test_a_stream_on_the_same_transport():
+    drv = _driver()
+    transport = drv.device.remote_conn_pre.get_transport.return_value
+    with patch("napalm_openwrt.openwrt.open_stream_on_transport") as stream:
+        drv.open_stream("logread -f")
+
+    stream.assert_called_once_with(transport, "logread -f")
+
+
+def test_not_connected():
+    drv = _driver()
+    drv.device = None
+    with pytest.raises(ConnectionClosedException):
+        drv.run_command("true")
+
+
+def test_no_warnings_of_its_own():
+    """base64 is no longer needed for anything, so a missing one is no warning;
+    lldpd, auc and the updates are facts the caller reads (netOrk #1111)."""
+    assert not any(
+        "get_device_warnings" in vars(c)
+        for c in OpenWrtDriver.__mro__
+        if c.__module__.startswith("napalm_openwrt")
+    )
+
+
+class TestRebootHost:
+    """``reboot_host`` (napalm-device-types ``HostRebootMixin``): netOrk sent
+    ``/sbin/reboot`` through the private ``_send_command`` and could not tell a
+    refusal from the session dropping (netOrk #1113)."""
+
+    def test_detached_so_the_answer_comes_back_first(self):
+        drv = _driver()
+        with patch(
+            "napalm_openwrt.openwrt.run_on_transport", return_value=CommandResult("", "", 0)
+        ) as run:
+            drv.reboot_host()
+
+        command = run.call_args.args[1]
+        assert "reboot" in command and command.rstrip().endswith("&'")
+        assert "trap" in command and "sleep" in command
+
+    def test_a_refusal_is_raised(self):
+        drv = _driver()
+        failed = CommandResult("", "reboot: Operation not permitted", 1)
+        with (
+            patch("napalm_openwrt.openwrt.run_on_transport", return_value=failed),
+            pytest.raises(RuntimeError, match="not permitted"),
+        ):
+            drv.reboot_host()
+
+    def test_a_login_other_than_root_is_refused(self):
+        drv = _driver(username="admin")
+        with patch("napalm_openwrt.openwrt.run_on_transport") as run, pytest.raises(RuntimeError):
+            drv.reboot_host()
+
+        run.assert_not_called()

tests/unit/test_driver.py

@@ -1,15 +1,16 @@
 """Unit tests for OpenWrtDriver — no real device required."""
 
-import pytest
 from unittest.mock import MagicMock, patch
 
-from napalm_openwrt.openwrt import OpenWrtDriver
+import pytest
 
+from napalm_openwrt.openwrt import OpenWrtDriver
 
 # ---------------------------------------------------------------------------
 # Fixtures
 # ---------------------------------------------------------------------------
 
+
 @pytest.fixture
 def driver():
     """Return a driver instance with a mocked Netmiko connection."""
@@ -130,9 +131,11 @@ HOSTNAME = "OpenWrt"
 # Tests
 # ---------------------------------------------------------------------------
 
+
 class TestGetFacts:
     def _make_send(self):
         """Return a _send_command mock that handles both str and list args."""
+
         def _send(cmd, **kw):
             key = cmd[0] if isinstance(cmd, list) else cmd
             if "openwrt_release" in key:
@@ -146,6 +149,7 @@ class TestGetFacts:
             if "ip link" in key:
                 return IP_LINK_SHOW
             return ""
+
         return _send
 
     def test_returns_required_keys(self, driver):
@@ -153,8 +157,15 @@ class TestGetFacts:
 
         facts = driver.get_facts()
         assert set(facts.keys()) == {
-            "vendor", "model", "hostname", "fqdn", "os_version",
+            "vendor",
-            "serial_number", "uptime", "interface_list", "number_of_interfaces",
+            "model",
+            "hostname",
+            "fqdn",
+            "os_version",
+            "serial_number",
+            "uptime",
+            "interface_list",
+            "number_of_interfaces",
         }
 
     def test_number_of_interfaces_matches_list(self, driver):
@@ -213,8 +224,13 @@ class TestGetInterfaces:
         result = driver.get_interfaces()
         for iface_data in result.values():
             assert set(iface_data.keys()) >= {
-                "is_up", "is_enabled", "description",
+                "is_up",
-                "last_flapped", "speed", "mtu", "mac_address",
+                "is_enabled",
+                "description",
+                "last_flapped",
+                "speed",
+                "mtu",
+                "mac_address",
             }
 
 
@@ -658,6 +674,7 @@ mgmt (id: 2)
 # Tests for new methods
 # ---------------------------------------------------------------------------
 
+
 class TestCli:
     def test_returns_dict_keyed_by_command(self, driver):
         driver._send_command = lambda cmd, **kw: f"output of {cmd}"
@@ -758,8 +775,17 @@ class TestGetNtpStats:
         result = driver.get_ntp_stats()
         for entry in result:
             assert set(entry.keys()) >= {
-                "remote", "referenceid", "synchronized", "stratum",
+                "remote",
-                "type", "when", "hostpoll", "reachability", "delay", "offset", "jitter",
+                "referenceid",
+                "synchronized",
+                "stratum",
+                "type",
+                "when",
+                "hostpoll",
+                "reachability",
+                "delay",
+                "offset",
+                "jitter",
             }
 
     def test_empty_on_no_tool(self, driver):
@@ -885,11 +911,15 @@ class TestGetRouteTo:
 
     def test_required_keys(self, driver):
         driver._send_command = lambda cmd, **kw: IP_ROUTE_SHOW
-        for prefix, routes in driver.get_route_to().items():
+        for _prefix, routes in driver.get_route_to().items():
             for route in routes:
                 assert set(route.keys()) >= {
-                    "protocol", "current_active", "next_hop",
+                    "protocol",
-                    "outgoing_interface", "preference", "routing_table",
+                    "current_active",
+                    "next_hop",
+                    "outgoing_interface",
+                    "preference",
+                    "routing_table",
                 }
 
 
@@ -925,23 +955,17 @@ class TestTraceroute:
 
 class TestGetNetworkInstances:
     def test_default_instance_always_present(self, driver):
-        driver._send_command = lambda cmd, **kw: (
+        driver._send_command = lambda cmd, **kw: IP_LINK_SHOW if "ip link" in cmd else ""
-            IP_LINK_SHOW if "ip link" in cmd else ""
-        )
         result = driver.get_network_instances()
         assert "default" in result
 
     def test_default_instance_type(self, driver):
-        driver._send_command = lambda cmd, **kw: (
+        driver._send_command = lambda cmd, **kw: IP_LINK_SHOW if "ip link" in cmd else ""
-            IP_LINK_SHOW if "ip link" in cmd else ""
-        )
         result = driver.get_network_instances()
         assert result["default"]["type"] == "DEFAULT_INSTANCE"
 
     def test_default_interfaces_populated(self, driver):
-        driver._send_command = lambda cmd, **kw: (
+        driver._send_command = lambda cmd, **kw: IP_LINK_SHOW if "ip link" in cmd else ""
-            IP_LINK_SHOW if "ip link" in cmd else ""
-        )
         result = driver.get_network_instances()
         ifaces = result["default"]["interfaces"]["interface"]
         assert "eth0" in ifaces
@@ -955,6 +979,7 @@ class TestGetNetworkInstances:
             if "ip link" in cmd:
                 return IP_LINK_SHOW
             return ""
+
         driver._send_command = _send
         result = driver.get_network_instances()
         assert "vpn" in result
@@ -962,16 +987,12 @@ class TestGetNetworkInstances:
         assert result["vpn"]["type"] == "L3VRF"
 
     def test_name_filter(self, driver):
-        driver._send_command = lambda cmd, **kw: (
+        driver._send_command = lambda cmd, **kw: IP_LINK_SHOW if "ip link" in cmd else ""
-            IP_LINK_SHOW if "ip link" in cmd else ""
-        )
         result = driver.get_network_instances(name="default")
         assert list(result.keys()) == ["default"]
 
     def test_required_keys(self, driver):
-        driver._send_command = lambda cmd, **kw: (
+        driver._send_command = lambda cmd, **kw: IP_LINK_SHOW if "ip link" in cmd else ""
-            IP_LINK_SHOW if "ip link" in cmd else ""
-        )
         for inst in driver.get_network_instances().values():
             assert set(inst.keys()) >= {"name", "type", "state", "interfaces"}
 
@@ -1057,6 +1078,7 @@ class TestGetChannelScan:
 
     def _make_send(self, mode: str = "active"):
         """Return a _send_command mock routing iw dev / iw scan output."""
+
         def _send(cmd, **kw):
             if cmd.strip().startswith("iw dev") and "scan" not in cmd:
                 return IW_DEV_OUT
@@ -1065,6 +1087,7 @@ class TestGetChannelScan:
             if "wlan1" in cmd and "scan" in cmd:
                 return IW_SCAN_WLAN1
             return ""
+
         return _send
 
     def test_returns_dict(self, driver):
@@ -1089,8 +1112,13 @@ class TestGetChannelScan:
         driver._send_command = self._make_send()
         entry = driver.get_channel_scan()["wlan0"][0]
         assert set(entry.keys()) >= {
-            "bssid", "ssid", "frequency", "channel", "signal_dbm",
+            "bssid",
-            "channel_width", "band",
+            "ssid",
+            "frequency",
+            "channel",
+            "signal_dbm",
+            "channel_width",
+            "band",
         }
 
     def test_wlan0_first_entry_values(self, driver):
@@ -1300,7 +1328,9 @@ class TestGetSsidsAcl:
 
     def test_deny_maps_to_blacklist(self, driver):
         deny_uci = UCI_WIRELESS_ACL.replace("macfilter='allow'", "macfilter='deny'")
-        driver._send_command = lambda cmd, **kw: deny_uci if cmd.strip() == "uci show wireless" else ""
+        driver._send_command = lambda cmd, **kw: (
+            deny_uci if cmd.strip() == "uci show wireless" else ""
+        )
         result = driver.get_ssids()
         assert result["CorpWiFi"]["acl_mode"] == "blacklist"
 
@@ -1308,7 +1338,9 @@ class TestGetSsidsAcl:
 class TestPushMacAcl:
     """Tests for OpenWrtWirelessMixin.push_mac_acl()."""
 
-    def _make_send(self, sections="wireless.@wifi-iface[0]\nwireless.@wifi-iface[1]", ssid_by_section=None):
+    def _make_send(
+        self, sections="wireless.@wifi-iface[0]\nwireless.@wifi-iface[1]", ssid_by_section=None
+    ):
         ssid_by_section = ssid_by_section or {
             "wireless.@wifi-iface[0]": "CorpWiFi",
             "wireless.@wifi-iface[1]": "GuestNet",
@@ -1370,16 +1402,24 @@ class TestPushMacAcl:
         send, issued = self._make_send()
         driver._send_command = send
         driver.push_mac_acl("CorpWiFi", "whitelist", ["AA:BB:CC:DD:EE:01", "AA:BB:CC:DD:EE:02"])
-        assert any("add_list wireless.@wifi-iface[0].maclist='AA:BB:CC:DD:EE:01'" in c for c in issued)
+        assert any(
-        assert any("add_list wireless.@wifi-iface[0].maclist='AA:BB:CC:DD:EE:02'" in c for c in issued)
+            "add_list wireless.@wifi-iface[0].maclist='AA:BB:CC:DD:EE:01'" in c for c in issued
+        )
+        assert any(
+            "add_list wireless.@wifi-iface[0].maclist='AA:BB:CC:DD:EE:02'" in c for c in issued
+        )
 
     def test_maclist_cleared_before_readd(self, driver):
         """Full-rebuild: existing maclist must be deleted before new entries are added."""
         send, issued = self._make_send()
         driver._send_command = send
         driver.push_mac_acl("CorpWiFi", "whitelist", ["AA:BB:CC:DD:EE:01"])
-        delete_idx = next(i for i, c in enumerate(issued) if "delete wireless.@wifi-iface[0].maclist" in c)
+        delete_idx = next(
-        add_idx = next(i for i, c in enumerate(issued) if "add_list wireless.@wifi-iface[0].maclist" in c)
+            i for i, c in enumerate(issued) if "delete wireless.@wifi-iface[0].maclist" in c
+        )
+        add_idx = next(
+            i for i, c in enumerate(issued) if "add_list wireless.@wifi-iface[0].maclist" in c
+        )
         assert delete_idx < add_idx
 
     def test_only_matching_ssid_sections_touched(self, driver):
@@ -1474,8 +1514,17 @@ IP_ADDR_BRLAN = """\
 class _FakeShell:
     """Collects issued commands and answers them from a canned config."""
 
-    def __init__(self, firewall="", network="", ip_addr="", ssh_connection="",
+    def __init__(
-                 reload_out="", nft_hits="1", listen_hits="1", snmpd="running"):
+        self,
+        firewall="",
+        network="",
+        ip_addr="",
+        ssh_connection="",
+        reload_out="",
+        nft_hits="1",
+        listen_hits="1",
+        snmpd="running",
+    ):
         self.firewall = firewall
         self.network = network
         self.ip_addr = ip_addr
@@ -1488,6 +1537,12 @@ class _FakeShell:
 
     def __call__(self, cmd, **kw):
         self.issued.append(cmd)
+        if "/usr/sbin/snmpd" in cmd:
+            return "present"
+        if "__OWRT_RC" in cmd:
+            return "ok\n__OWRT_RC=0"
+        if cmd.startswith("uci show snmpd"):
+            return ""
         if cmd.startswith("uci show firewall"):
             return self.firewall
         if cmd.startswith("uci show network"):
@@ -1507,8 +1562,11 @@ class _FakeShell:
         return ""
 
 
+_SETTINGS = {"community": "c0mm", "allowed_sources": []}
+
+
 class TestFixSnmpZoneDetection:
-    """_action_fix_snmp() must resolve the real zone that owns the mgmt address."""
+    """The firewall step must resolve the real zone that owns the mgmt address."""
 
     def test_static_mgmt_address_selects_owning_zone(self, driver):
         """10.10.0.5 lives on network 'mgmt' → zone 'mgmt', not the 'lan' fallback."""
@@ -1518,7 +1576,7 @@ class TestFixSnmpZoneDetection:
             ssh_connection="10.10.0.1 51234 10.10.0.5 22",
         )
         driver._send_command = shell
-        result = driver._action_fix_snmp()
+        result = driver.configure_snmp_agent(_SETTINGS)
         assert "Management zone: 'mgmt'" in result["output"]
         assert any("firewall.allow_snmp_from_mgmt.src='mgmt'" in c for c in shell.issued)
 
@@ -1530,7 +1588,7 @@ class TestFixSnmpZoneDetection:
             ssh_connection="192.168.1.50 5000 192.168.1.1 22",
         )
         driver._send_command = shell
-        result = driver._action_fix_snmp()
+        result = driver.configure_snmp_agent(_SETTINGS)
         assert "Management zone: 'lan'" in result["output"]
 
     def test_dhcp_mgmt_address_resolved_via_l3_device(self, driver):
@@ -1542,7 +1600,7 @@ class TestFixSnmpZoneDetection:
             ssh_connection="10.10.0.1 51234 10.10.0.5 22",
         )
         driver._send_command = shell
-        result = driver._action_fix_snmp()
+        result = driver.configure_snmp_agent(_SETTINGS)
         assert "Management zone: 'lan'" in result["output"]
 
     def test_nameless_zone_is_reported_and_action_fails(self, driver):
@@ -1553,7 +1611,7 @@ class TestFixSnmpZoneDetection:
             ssh_connection="192.168.1.50 5000 192.168.1.1 22",
         )
         driver._send_command = shell
-        result = driver._action_fix_snmp()
+        result = driver.configure_snmp_agent(_SETTINGS)
         assert result["success"] is False
         assert "@zone[0]" in result["output"]
         assert "name" in result["output"]
@@ -1566,7 +1624,7 @@ class TestFixSnmpZoneDetection:
             ssh_connection="192.168.1.50 5000 192.168.1.1 22",
         )
         driver._send_command = shell
-        driver._action_fix_snmp()
+        driver.configure_snmp_agent(_SETTINGS)
         assert not any("allow_snmp" in c for c in shell.issued)
 
 
@@ -1586,7 +1644,7 @@ class TestFixSnmpRuleWriting:
             ssh_connection="192.168.1.50 5000 192.168.1.1 22",
         )
         driver._send_command = shell
-        driver._action_fix_snmp()
+        driver.configure_snmp_agent(_SETTINGS)
         assert any("firewall.allow_snmp_from_lan.src='lan'" in c for c in shell.issued)
 
     def test_stale_rules_are_deleted_highest_index_first(self, driver):
@@ -1605,7 +1663,7 @@ class TestFixSnmpRuleWriting:
             ssh_connection="192.168.1.50 5000 192.168.1.1 22",
         )
         driver._send_command = shell
-        driver._action_fix_snmp()
+        driver.configure_snmp_agent(_SETTINGS)
         deletes = [c for c in shell.issued if "delete" in c and "@rule" in c]
         joined = " ".join(deletes)
         assert joined.index("@rule[2]") < joined.index("@rule[1]")
@@ -1623,8 +1681,10 @@ class TestFixSnmpRuleWriting:
             ssh_connection="192.168.1.50 5000 192.168.1.1 22",
         )
         driver._send_command = shell
-        driver._action_fix_snmp()
+        driver.configure_snmp_agent(_SETTINGS)
-        delete_idx = next(i for i, c in enumerate(shell.issued) if "snmp_netork" in c and "delete" in c)
+        delete_idx = next(
+            i for i, c in enumerate(shell.issued) if "snmp_netork" in c and "delete" in c
+        )
         assert any("uci commit firewall" in c for c in shell.issued[delete_idx:])
 
     def test_legacy_snmp_netork_rule_is_recognised_as_stale(self, driver):
@@ -1641,7 +1701,7 @@ class TestFixSnmpRuleWriting:
             ssh_connection="192.168.1.50 5000 192.168.1.1 22",
         )
         driver._send_command = shell
-        driver._action_fix_snmp()
+        driver.configure_snmp_agent(_SETTINGS)
         assert any("snmp_netork" in c and "delete" in c for c in shell.issued)
 
 
@@ -1663,7 +1723,7 @@ class TestFixSnmpReloadVerification:
             reload_out=self.FW4_ZONE_ERROR,
         )
         driver._send_command = shell
-        result = driver._action_fix_snmp()
+        result = driver.configure_snmp_agent(_SETTINGS)
         assert result["success"] is False
 
     def test_reload_output_is_not_truncated(self, driver):
@@ -1675,7 +1735,7 @@ class TestFixSnmpReloadVerification:
             reload_out=self.FW4_ZONE_ERROR,
         )
         driver._send_command = shell
-        result = driver._action_fix_snmp()
+        result = driver.configure_snmp_agent(_SETTINGS)
         assert "references unknown zone" in result["output"]
 
     def test_missing_live_rule_fails_the_action(self, driver):
@@ -1687,7 +1747,7 @@ class TestFixSnmpReloadVerification:
             nft_hits="0",
         )
         driver._send_command = shell
-        result = driver._action_fix_snmp()
+        result = driver.configure_snmp_agent(_SETTINGS)
         assert result["success"] is False
 
     def test_snmpd_not_listening_fails_the_action(self, driver):
@@ -1698,7 +1758,7 @@ class TestFixSnmpReloadVerification:
             listen_hits="0",
         )
         driver._send_command = shell
-        result = driver._action_fix_snmp()
+        result = driver.configure_snmp_agent(_SETTINGS)
         assert result["success"] is False
 
     def test_fully_healthy_run_succeeds(self, driver):
@@ -1708,7 +1768,7 @@ class TestFixSnmpReloadVerification:
             ssh_connection="192.168.1.50 5000 192.168.1.1 22",
         )
         driver._send_command = shell
-        result = driver._action_fix_snmp()
+        result = driver.configure_snmp_agent(_SETTINGS)
         assert result["success"] is True
 
 

tests/unit/test_listening_sockets.py

@@ -0,0 +1,89 @@
+"""What listens on an OpenWrt device, and which procd service it is (netOrk #673).
+
+The command and its parse are napalm-device-types' (``ListeningSocketsMixin``):
+OpenWrt has no ``ss``, so the command falls back to busybox ``netstat``, and the
+cgroup of each process names its procd service. The driver only carries the
+command across -- over an SSH exec channel of its own, because busybox ash cuts
+an interactive line at 512 characters and the command is longer.
+
+The netstat output is a real OpenWrt 25.12.2 access point's, with documentation
+addresses.
+"""
+
+from __future__ import annotations
+
+import io
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from napalm_openwrt.openwrt import OpenWrtDriver
+
+REPORT = """SOCK_BEGIN
+[netstat]
+Active Internet connections (only servers)
+Proto Recv-Q Send-Q Local Address           Foreign Address         State       PID/Program name
+tcp        0      0 0.0.0.0:22              0.0.0.0:*               LISTEN      1604/dropbear
+tcp        0      0 0.0.0.0:443             0.0.0.0:*               LISTEN      1969/uhttpd
+tcp        0      0 192.0.2.15:53           0.0.0.0:*               LISTEN      1495/dnsmasq
+tcp        0      0 :::443                  :::*                    LISTEN      1969/uhttpd
+udp        0      0 0.0.0.0:161             0.0.0.0:*                           3173/snmpd
+__SS_RC=0
+[cgroups]
+1495 0::/services/dnsmasq/cfg01411c
+1604 0::/services/dropbear/instance1
+1969 0::/services/uhttpd/instance1
+3173 0::/services/snmpd/instance1
+SOCK_END
+"""
+
+
+def _driver(username: str = "root", report: str = REPORT) -> OpenWrtDriver:
+    with patch("napalm_openwrt.openwrt.ConnectHandler"):
+        drv = OpenWrtDriver(hostname="192.0.2.15", username=username, password="")
+    drv.device = MagicMock()
+    exec_command = drv.device.remote_conn_pre.exec_command
+    exec_command.side_effect = lambda *_a, **_k: (None, io.BytesIO(report.encode()), None)
+    return drv
+
+
+def test_every_socket_with_its_procd_service():
+    reading = _driver().get_listening_sockets()
+
+    assert reading["attributed"] is True
+    services = {(s["proto"], s["port"], s["address"]): s["unit"] for s in reading["sockets"]}
+    assert services == {
+        ("tcp", 22, "0.0.0.0"): "dropbear",
+        ("tcp", 53, "192.0.2.15"): "dnsmasq",
+        ("tcp", 443, "0.0.0.0"): "uhttpd",
+        ("tcp", 443, "::"): "uhttpd",
+        ("udp", 161, "0.0.0.0"): "snmpd",
+    }
+
+
+def test_the_command_goes_over_an_exec_channel_not_the_shell():
+    drv = _driver()
+    drv.get_listening_sockets()
+
+    [call] = drv.device.remote_conn_pre.exec_command.call_args_list
+    assert call.args[0].startswith("sh -c '")
+    drv.device.send_command.assert_not_called()
+
+
+def test_a_login_other_than_root_reads_without_attributing():
+    drv = _driver(username="admin")
+
+    reading = drv.get_listening_sockets()
+
+    assert reading["attributed"] is False
+    assert len(drv.device.remote_conn_pre.exec_command.call_args_list) == 1
+
+
+def test_a_report_cut_short_raises():
+    with pytest.raises(ValueError):
+        _driver(report=REPORT.replace("SOCK_END\n", "")).get_listening_sockets()
+
+
+def test_netork_finds_it():
+    """netOrk asks ``hasattr(driver, "get_listening_sockets")``."""
+    assert hasattr(OpenWrtDriver, "get_listening_sockets")

tests/unit/test_snmp_agent.py

@@ -0,0 +1,96 @@
+"""configure_snmp_agent's values: community, sources, contact, location (netOrk #1105).
+
+The old fix_snmp set no community at all and opened the management zone to every
+source; the rule was named after an application. Every value is the caller's now.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from napalm_openwrt import OpenWrtDriver
+from tests.unit.test_driver import UCI_FIREWALL_HEALTHY, UCI_NETWORK_STATIC, _FakeShell
+
+SETTINGS = {
+    "community": "c0mm",
+    "allowed_sources": ["10.7.224.20", "fd00::20"],
+    "contact": "ops",
+    "location": "rack 1",
+}
+
+
+@pytest.fixture
+def shell():
+    return _FakeShell(
+        firewall=UCI_FIREWALL_HEALTHY,
+        network=UCI_NETWORK_STATIC,
+        ssh_connection="10.7.224.20 51234 192.168.1.1 22",
+    )
+
+
+@pytest.fixture
+def owrt(shell):
+    d = OpenWrtDriver("ap", "root", "pw")
+    d._send_command = shell
+    return d
+
+
+def _joined(shell):
+    return " ".join(shell.issued)
+
+
+def test_the_steps_run_in_order_and_succeed(owrt):
+    result = owrt.configure_snmp_agent(SETTINGS)
+
+    assert result["success"] is True
+    assert [s["step"] for s in result["steps"]] == ["install", "configure", "firewall", "start"]
+
+
+def test_community_per_source_in_uci(owrt, shell):
+    owrt.configure_snmp_agent(SETTINGS)
+
+    uci = _joined(shell)
+    assert "uci set snmpd.public.source='10.7.224.20/32'" in uci
+    assert "uci set snmpd.public.community='c0mm'" in uci
+    assert "uci set snmpd.public6.source='fd00::20/128'" in uci
+    assert "snmpd.@system[0].sysLocation='rack 1'" in uci
+    assert "uci commit snmpd" in uci
+
+
+def test_without_sources_every_address_may_ask(owrt, shell):
+    owrt.configure_snmp_agent({"community": "c0mm", "allowed_sources": []})
+
+    assert "uci set snmpd.public.source='default'" in _joined(shell)
+
+
+def test_ipv4_sources_only_keep_ipv6_to_localhost(owrt, shell):
+    owrt.configure_snmp_agent({"community": "c0mm", "allowed_sources": ["10.7.224.20"]})
+
+    assert "uci set snmpd.public6.source='::1/128'" in _joined(shell)
+
+
+def test_the_firewall_rule_is_limited_to_the_sources_and_neutrally_named(owrt, shell):
+    owrt.configure_snmp_agent(SETTINGS)
+
+    uci = _joined(shell)
+    assert "uci add_list firewall.allow_snmp_from_lan.src_ip='10.7.224.20/32'" in uci
+    assert "firewall.allow_snmp_from_lan.name='Allow-SNMP'" in uci
+    assert "netork" not in uci.lower()
+
+
+def test_the_management_peer_is_the_ssh_client(owrt):
+    assert owrt.get_management_peer() == "10.7.224.20"
+
+
+@pytest.mark.parametrize("community", ["two words", "it's", ""])
+def test_a_community_that_would_break_uci_is_refused(owrt, shell, community):
+    result = owrt.configure_snmp_agent({"community": community, "allowed_sources": []})
+
+    assert result["success"] is False
+    assert "uci commit snmpd" not in _joined(shell)
+
+
+def test_the_old_actions_are_gone():
+    gone = ("run_device_action", "_action_fix_snmp", "_action_install_lldpd", "_action_install_auc")
+    for name in gone:
+        assert not hasattr(OpenWrtDriver, name), name

tests/unit/test_ssh_host_key.py

@@ -0,0 +1,86 @@
+"""open() hands netmiko the host key to expect, and reports the one it saw (NetOrk/netork#1167)."""
+
+from __future__ import annotations
+
+import os
+from unittest.mock import MagicMock, patch
+
+import paramiko
+import pytest
+from napalm_device_types.ssh_host_keys import (
+    KNOWN_KEYS_ARG,
+    OBSERVER_ARG,
+    SshHostKey,
+    SshHostKeyMismatch,
+)
+from netmiko.exceptions import NetmikoTimeoutException
+
+from napalm_openwrt.openwrt import OpenWrtDriver
+
+SERVER_KEY = paramiko.ECDSAKey.generate()
+OTHER_KEY = paramiko.ECDSAKey.generate()
+HANDLER = "napalm_openwrt.openwrt.ConnectHandler"
+
+
+def _device() -> MagicMock:
+    device = MagicMock()
+    device.remote_conn_pre.get_transport.return_value.get_remote_server_key.return_value = (
+        SERVER_KEY
+    )
+    return device
+
+
+def _driver(seen: list, known: SshHostKey | None = None) -> OpenWrtDriver:
+    args: dict = {"port": 2222, OBSERVER_ARG: seen.append}
+    if known is not None:
+        args[KNOWN_KEYS_ARG] = {"2222": known.as_dict()}
+    return OpenWrtDriver("10.0.0.8", "admin", "s3cret", optional_args=args)
+
+
+def test_first_use_lets_netmiko_accept_and_reports_the_key():
+    seen: list = []
+    with patch(HANDLER, return_value=_device()) as handler:
+        _driver(seen).open()
+    kwargs = handler.call_args.kwargs
+    assert kwargs["ssh_strict"] is False
+    assert kwargs["alt_host_keys"] is False
+    assert kwargs["system_host_keys"] is False
+    assert kwargs["port"] == 2222
+    assert seen[0]["accepted"] is True
+    assert seen[0]["seen"]["fingerprint"] == SshHostKey.of(SERVER_KEY).fingerprint
+    assert (seen[0]["host"], seen[0]["port"]) == ("10.0.0.8", 2222)
+
+
+def test_a_known_key_is_the_only_one_netmiko_accepts():
+    seen: list = []
+    known = SshHostKey.of(SERVER_KEY)
+    files: list[str] = []
+
+    def connect(**kwargs):
+        files.append(kwargs["alt_key_file"])
+        with open(kwargs["alt_key_file"]) as f:
+            assert f.read().strip() == known.known_hosts_line("10.0.0.8", 2222)
+        assert kwargs["ssh_strict"] is True and kwargs["alt_host_keys"] is True
+        return _device()
+
+    with patch(HANDLER, side_effect=connect):
+        _driver(seen, known).open()
+    assert not os.path.exists(files[0])
+    assert seen[0]["accepted"] is True
+
+
+def test_a_changed_key_is_refused():
+    seen: list = []
+
+    def refuse(**kwargs):
+        try:
+            raise paramiko.BadHostKeyException("10.0.0.8", SERVER_KEY, OTHER_KEY)
+        except paramiko.SSHException as exc:  # what netmiko does with it
+            raise NetmikoTimeoutException(f"A paramiko SSHException occurred: {exc}") from exc
+
+    driver = _driver(seen, SshHostKey.of(OTHER_KEY))
+    with patch(HANDLER, side_effect=refuse):
+        with pytest.raises(SshHostKeyMismatch):
+            driver.open()
+    assert driver.device is None
+    assert seen[0]["accepted"] is False