feat(system): report remote syslog, LuCI state and bridge STP

get_system_config() already parsed the whole system section but returned
only a slice of it, so netOrk had no IST side for three AP-profile fields
and could not compare them at all — the fields were editable, stored and
silently ineffective.

Adds:
  * syslog_remote / syslog_ip / syslog_port / syslog_proto — read from the
    system section that was already being parsed. log_remote gates the
    others: OpenWrt ships nothing without it, so a leftover log_ip must
    not read as an active target.
  * luci_enabled — from the uhttpd init script rather than its listener
    config, so the answer stays "is the web UI served" and re-enabling
    restores whatever was configured before.
  * bridge_stp — None when there is no br-ap at all, which is a different
    statement from "bridge without STP" and has to stay distinguishable.

netOrk #164

napalm_openwrt/openwrt.py

@@ -278,26 +278,63 @@ class OpenWrtDriver(
     # defines NotImplementedError stubs for these)
     # ------------------------------------------------------------------
 
-    def get_services(self) -> list[dict[str, Any]]:
-        return OpenWrtSystemMixin.get_services(self)
 
-    def manage_service(self, name: str, action: str) -> dict[str, Any]:
-        return OpenWrtSystemMixin.manage_service(self, name, action)
 
-    def get_available_updates(self) -> list[dict[str, Any]]:
-        return OpenWrtSystemMixin.get_available_updates(self)
 
-    def apply_updates(self, packages: list[str]) -> dict[str, Any]:
-        return OpenWrtSystemMixin.apply_updates(self, packages)
 
-    def get_packages(self) -> list[dict[str, Any]]:
-        return OpenWrtPackageMixin.get_packages(self)
 
-    def install_package(self, name: str) -> dict[str, Any]:
-        return OpenWrtPackageMixin.install_package(self, name)
 
-    def remove_package(self, name: str) -> dict[str, Any]:
+
-        return OpenWrtPackageMixin.uninstall_package(self, name)
+    # ------------------------------------------------------------------
+    # UCI parsing helpers
+    # ------------------------------------------------------------------
+
+    _UCI_TOKEN_RE = re.compile(r"'([^']*)'|\"([^\"]*)\"|(\S+)")
+
+    @classmethod
+    def _uci_tokens(cls, value: str) -> list[str]:
+        """Split a ``uci show`` value into its (possibly quoted) tokens.
+
+        List options are printed as ``opt='a' 'b'``; scalars as ``opt='a'``.
+        """
+        return [a or b or c for a, b, c in cls._UCI_TOKEN_RE.findall(value)]
+
+    @classmethod
+    def _uci_scalar(cls, value: str) -> str:
+        """Return the first token of a ``uci show`` value, unquoted."""
+        tokens = cls._uci_tokens(value)
+        return tokens[0] if tokens else ""
+
+    @classmethod
+    def _parse_uci_sections(cls, raw: str) -> dict[str, dict[str, str]]:
+        """Parse ``uci show <pkg>`` into ``{section: {"_type": t, opt: raw_value}}``.
+
+        Option values are kept verbatim so that list options survive; use
+        :meth:`_uci_scalar` / :meth:`_uci_tokens` to read them.
+        """
+        sections: dict[str, dict[str, str]] = {}
+        for line in raw.splitlines():
+            line = line.strip()
+            if not line or "=" not in line:
+                continue
+            key, _, value = line.partition("=")
+            parts = key.split(".")
+            if len(parts) == 2:  # firewall.@zone[0]=zone
+                sections.setdefault(parts[1], {})["_type"] = cls._uci_scalar(value)
+            elif len(parts) >= 3:  # firewall.@zone[0].name='lan'
+                sections.setdefault(parts[1], {})[parts[2]] = value.strip()
+        return sections
+
+    @staticmethod
+    def _uci_delete_order(section: str) -> tuple[int, int]:
+        """Sort key that deletes named sections first, then anonymous descending.
+
+        ``uci delete firewall.@rule[1]`` renumbers every later ``@rule[n]``,
+        so a batch of deletes only stays correct when the highest index goes
+        first.
+        """
+        match = re.fullmatch(r"@[\w-]+\[(-?\d+)\]", section)
+        return (1, -int(match.group(1))) if match else (0, 0)
 
     # ------------------------------------------------------------------
     # Device actions
@@ -315,83 +352,229 @@ class OpenWrtDriver(
             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.
+
+        ``$SSH_CONNECTION`` is ``<client ip> <client port> <server ip>
+        <server port>`` — the third field is the address the device is
+        managed on, which is what the firewall rule has to cover.  Falls
+        back to the address NAPALM connected to.
+        """
+        parts = self._send_command("echo $SSH_CONNECTION").split()
+        return parts[2] if len(parts) >= 3 else self.hostname
+
+    def _l3_device_for_address(self, address: str) -> str:
+        """Return the interface holding ``address`` (``br-lan``, ``eth0.9``, …)."""
+        raw = self._send_command("ip -o -4 addr show 2>/dev/null")
+        for line in raw.splitlines():
+            parts = line.split()
+            if len(parts) >= 4 and parts[2] == "inet" and parts[3].split("/")[0] == address:
+                return parts[1]
+        return ""
+
+    def _mgmt_network(self, net_sections: dict[str, dict[str, str]], address: str) -> str:
+        """Return the ``network`` section that carries the management address."""
+        for key, opts in net_sections.items():
+            if opts.get("_type") != "interface":
+                continue
+            if self._uci_scalar(opts.get("ipaddr", "")) == address:
+                return key
+        # DHCP-addressed management interface: address → L3 device → section
+        device = self._l3_device_for_address(address)
+        if device:
+            for key, opts in net_sections.items():
+                if opts.get("_type") != "interface":
+                    continue
+                if self._uci_scalar(opts.get("device", "")) == device:
+                    return key
+        return ""
+
+    @classmethod
+    def _firewall_zones(
+        cls, fw_sections: dict[str, dict[str, str]]
+    ) -> tuple[dict[str, str], list[str]]:
+        """Split zone sections into ``{name: section}`` and the nameless ones.
+
+        A ``config zone`` without ``option name`` is invalid for fw4: the
+        section is skipped and every rule whose ``src`` points at it is
+        dropped along with it.
+        """
+        named: dict[str, str] = {}
+        nameless: list[str] = []
+        for key, opts in fw_sections.items():
+            if opts.get("_type") != "zone":
+                continue
+            name = cls._uci_scalar(opts.get("name", ""))
+            if name:
+                named[name] = key
+            else:
+                nameless.append(key)
+        return named, sorted(nameless)
+
+    @classmethod
+    def _zone_for_network(
+        cls,
+        fw_sections: dict[str, dict[str, str]],
+        candidates: dict[str, str],
+        network: str,
+    ) -> str:
+        """Return the label of the candidate zone whose ``network`` list holds ``network``.
+
+        ``candidates`` maps a label (zone name, or section key for zones that
+        have none) to the UCI section it lives in.
+        """
+        for label, section in candidates.items():
+            if network in cls._uci_tokens(fw_sections.get(section, {}).get("network", "")):
+                return label
+        return ""
+
+    @classmethod
+    def _stale_snmp_rules(
+        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
+            for key, opts in fw_sections.items()
+            if opts.get("_type") == "rule"
+            and key != keep
+            and (
+                "snmp" in cls._uci_scalar(opts.get("name", "")).lower()
+                or "snmp" in key.lower()
+                or cls._uci_scalar(opts.get("dest_port", "")) == "161"
+            )
+        ]
+        return sorted(hits, key=cls._uci_delete_order)
+
+    _FW_RELOAD_ERRORS = (
+        "mandatory but not set",
+        "skipped due to invalid",
+        "references unknown",
+        "is not a valid",
+        "error:",
+    )
+
+    @classmethod
+    def _fw_reload_failed(cls, output: str) -> bool:
+        """True when fw4/fw3 refused part of the ruleset."""
+        low = output.lower()
+        return any(token in low for token in cls._FW_RELOAD_ERRORS)
+
+    @staticmethod
+    def _grep_count(output: str) -> int:
+        """Read the count printed by ``grep -c``; 0 when the command failed."""
+        for line in reversed(output.strip().splitlines()):
+            line = line.strip()
+            if line.isdigit():
+                return int(line)
+        return 0
+
     def _action_fix_snmp(self) -> dict[str, Any]:
         """Ensure snmpd is running and reachable on UDP/161.
 
-        On OpenWRT the most common reason SNMP is unreachable is that the
+        On OpenWrt the usual reason SNMP stays unreachable is the firewall:
-        firewall management zone (typically named ``mgmt``) only allows
+        snmpd listens, but the zone covering the management interface
-        SSH/HTTP/HTTPS/ICMP by default and default-drops everything else.
+        default-drops everything except SSH/HTTP/HTTPS/ICMP.  A rule only
-        snmpd runs but packets are rejected before reaching the process.
+        lands in the right nftables chain when its ``src`` names the zone
+        that actually owns the management network.
 
         This action:
-        1. Detects the management zone name from UCI (looks for the zone
+
-           whose named rules already allow SSH — that zone handles the
+        1. Resolves the management address from ``$SSH_CONNECTION``, maps it
-           management interface where SNMP needs to be open too).
+           to its ``network`` section and from there to the owning firewall
-        2. Removes any wrongly-scoped SNMP rule (one without src=<zone>).
+           zone — no guessing from rule names.
-        3. Adds a named UCI rule ``allow_snmp_from_<zone>`` with
+        2. Aborts with a diagnosis when that zone has no ``option name``:
-           ``src=<zone>`` so it ends up in the correct nftables chain.
+           fw4 skips such a section, so any rule written against it is dead
-        4. Commits and reloads fw4 immediately (no reboot needed).
+           on arrival.  Naming the zone is left to the operator because an
-        5. Ensures snmpd is enabled and running.
+           inert zone becoming active changes what the AP filters.
+        3. Removes stale SNMP rules from earlier versions (highest anonymous
+           index first) and commits the deletion.
+        4. Writes the ``allow_snmp_from_<zone>`` rule in full every run, so a
+           half-written rule from a previous attempt gets repaired.
+        5. Reloads the firewall and fails on any reload complaint.
+        6. Verifies on the device that snmpd listens and that a udp/161
+           accept rule is live in the packet filter.
         """
         lines: list[str] = []
 
-        # ── 1. Detect management zone name ───────────────────────────────
+        # ── 1. Resolve the zone that owns the management address ─────────
-        # Find the zone whose allow-SSH rule already exists — that is the
-        # management zone. Falls back to "lan" if nothing more specific
-        # is found (on vanilla APs without a dedicated mgmt zone the lan
-        # zone has input=ACCEPT anyway).
         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)
 
-        mgmt_zone = "lan"  # safe fallback — lan zone usually has ACCEPT
+        address = self._mgmt_address()
-        for line in fw_raw.splitlines():
+        net_sections = self._parse_uci_sections(self._send_command("uci show network 2>/dev/null"))
-            # Named rule pattern: firewall.allow_ssh_from_<zone>.src='<zone>'
+        mgmt_net = self._mgmt_network(net_sections, address)
-            if ".src=" in line and "ssh" in line.lower():
+        lines.append(f"[firewall] Management address {address} on network {mgmt_net or '<unknown>'}")
-                zone_val = line.split("=", 1)[-1].strip().strip("'\"")
-                if zone_val:
-                    mgmt_zone = zone_val
-                    break
 
-        lines.append(f"[firewall] Management zone: {mgmt_zone!r}")
+        broken_zone = (
-
+            self._zone_for_network(fw_sections, {key: key for key in nameless}, mgmt_net)
-        # ── 2. Clean up any wrongly-scoped previous SNMP rule ────────────
+            if mgmt_net
-        # A rule named Allow-SNMP without src= lands in the global input
+            else ""
-        # chain which is never reached for managed-zone traffic.
-        existing_names = [
-            ln.split("=")[0].strip()
-            for ln in fw_raw.splitlines()
-            if ".name='Allow-SNMP'" in ln or ".name='allow_snmp" in ln.lower()
-        ]
-        for uci_key in existing_names:
-            # Check whether this rule has the correct src
-            src_line = next(
-                (l for l in fw_raw.splitlines() if uci_key.replace(".name", ".src") in l),
-                "",
-            )
-            if f"='{mgmt_zone}'" not in src_line and f'="{mgmt_zone}"' not in src_line:
-                self._send_command(f"uci delete {uci_key.replace('.name', '')} 2>/dev/null || true")
-                lines.append(f"[firewall] Removed mis-scoped rule {uci_key}")
-
-        # ── 3. Add correctly-scoped rule if not already present ──────────
-        named_key = f"allow_snmp_from_{mgmt_zone}"
-        if f"firewall.{named_key}" in fw_raw:
-            lines.append(f"[firewall] Rule {named_key!r} already present — skipping add")
-        else:
-            rule_out = self._send_command(
-                f"uci set firewall.{named_key}=rule"
-                f" && uci set firewall.{named_key}.name='Allow-SNMP-from-{mgmt_zone}'"
-                f" && uci set firewall.{named_key}.src='{mgmt_zone}'"
-                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 commit firewall 2>&1"
-            )
-            lines.append(f"[firewall] Added rule {named_key!r}: {rule_out.strip()[:80] or 'ok'}")
-
-        # ── 4. Reload firewall ────────────────────────────────────────────
-        reload_out = self._send_command(
-            "fw4 reload 2>&1 || /etc/init.d/firewall reload 2>&1 || true"
         )
-        lines.append(f"[firewall] Reload: {reload_out.strip()[:120] or 'ok'}")
+        if broken_zone:
+            lines.append(
+                f"[firewall] Zone {broken_zone} owns network {mgmt_net!r} but has no "
+                f"'name' option — fw4 skips the section and drops every rule that "
+                f"references it. Fix on the device, then re-run:"
+            )
+            lines.append(
+                f"[firewall]   uci set firewall.{broken_zone}.name='{mgmt_net}' "
+                f"&& uci commit firewall && fw4 reload"
+            )
+            return {"success": 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}")
+        elif zones:
+            lines.append(
+                f"[firewall] No zone covers the management network — zones present: "
+                f"{', '.join(sorted(zones))}"
+            )
+            return {"success": False, "output": "\n".join(lines)}
+        else:
+            lines.append("[firewall] 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(
+                "; ".join(f"uci -q delete firewall.{key}" for key in stale)
+                + "; uci commit firewall"
+            )
+            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}.target='ACCEPT'",
+            f"uci set firewall.{named_key}.proto='udp'",
+            f"uci set firewall.{named_key}.dest_port='161'",
+        ]
+        if mgmt_zone:
+            rule_cmds.append(f"uci set firewall.{named_key}.src='{mgmt_zone}'")
+        else:
+            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'}")
+
+        # ── 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'}")
+        if not fw_ok:
+            lines.append("[firewall] 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")
@@ -404,19 +587,21 @@ class OpenWrtDriver(
         else:
             lines.append("[snmpd] Service already running")
 
-        # ── 6. Local probe (best-effort) ──────────────────────────────────
+        # ── 6. Verify on the device instead of assuming success ───────────
-        probe = self._send_command(
+        listening = self._grep_count(
-            "snmpget -v2c -cpublic -t2 -r0 -Ov 127.0.0.1 1.3.6.1.2.1.1.1.0 2>&1"
+            self._send_command("ss -lun 2>/dev/null | grep -c ':161'")
-            " || echo 'snmp_client_not_available'"
         )
-        if "snmp_client_not_available" in probe:
+        lines.append(f"[probe] snmpd listening on udp/161: {'yes' if listening else 'no'}")
-            lines.append("[probe] No local SNMP client — cannot verify locally")
-            success = True  # firewall rule was added; remote poll will confirm
-        else:
-            ok_tokens = ("STRING:", "INTEGER:", "OID:", "Timeticks:", "Hex-STRING:", "IpAddress:")
-            success = any(t in probe for t in ok_tokens)
-            lines.append(f"[probe] {'ok' if success else 'FAILED'}: {probe.strip()[:120]}")
 
+        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}")
+
+        success = fw_ok and bool(listening) and bool(live)
         return {"success": success, "output": "\n".join(lines)}
 
     def _action_install_coreutils_base64(self) -> dict[str, Any]:

napalm_openwrt/system_mixin.py

@@ -75,9 +75,17 @@ class OpenWrtSystemMixin:
         * dropbear_port (int) — SSH port
         * dropbear_password_auth (bool) — whether password login is allowed
         * dropbear_root_password_auth (bool)
+        * syslog_remote (bool) — whether logs are shipped off the device
+        * syslog_ip (str), syslog_port (int), syslog_proto (str)
+        * luci_enabled (bool) — whether the uhttpd service serving LuCI is enabled
+        * bridge_stp (bool | None) — STP on br-ap; None when there is no br-ap
         """
         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(
+            "/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] = {}
         for line in sys_out.splitlines():
@@ -109,6 +117,31 @@ class OpenWrtSystemMixin:
         def _bool_uci(val: str, default: bool = True) -> bool:
             return val.lower() not in ("0", "off", "false", "no") if val else default
 
+        # Remote syslog. OpenWrt only ships logs when log_remote is set, so a
+        # leftover log_ip without it means nothing is being sent.
+        syslog_remote = _bool_uci(sys_cfg.get("log_remote", ""), default=False)
+        try:
+            syslog_port = int(sys_cfg.get("log_port", "514") or "514")
+        except (ValueError, TypeError):
+            syslog_port = 514
+
+        # STP on the AP bridge. None when there is no br-ap device section at
+        # all — "no bridge" is a different statement from "bridge without STP".
+        bridge_stp: bool | None = None
+        br_section: str | None = None
+        for line in net_out.splitlines():
+            m = re.match(r"network\.(\w+)\.name='br-ap'", line.strip())
+            if m:
+                br_section = m.group(1)
+                break
+        if br_section:
+            bridge_stp = False
+            for line in net_out.splitlines():
+                m = re.match(rf"network\.{br_section}\.stp='([^']*)'", line.strip())
+                if m:
+                    bridge_stp = _bool_uci(m.group(1), default=False)
+                    break
+
         return {
             "hostname": sys_cfg.get("hostname", ""),
             "timezone": sys_cfg.get("timezone", ""),
@@ -117,6 +150,12 @@ class OpenWrtSystemMixin:
             "dropbear_port": ssh_port,
             "dropbear_password_auth": _bool_uci(db_cfg.get("PasswordAuth", "on")),
             "dropbear_root_password_auth": _bool_uci(db_cfg.get("RootPasswordAuth", "on")),
+            "syslog_remote": syslog_remote,
+            "syslog_ip": sys_cfg.get("log_ip", "") if syslog_remote else "",
+            "syslog_port": syslog_port,
+            "syslog_proto": sys_cfg.get("log_proto", "udp"),
+            "luci_enabled": luci_out.strip().endswith("1"),
+            "bridge_stp": bridge_stp,
         }
 
     def get_snmp_information(self) -> dict[str, Any]:
@@ -501,11 +540,7 @@ class OpenWrtSystemMixin:
         # 1. LLDP daemon
         lldpd_path = self._send_command("which lldpd 2>/dev/null").strip()
         if not lldpd_path:
-            warnings.append({
+            warnings.append({"code": "lldpd_not_installed"})
-                "code": "lldpd_not_installed",
-                "severity": "warning",
-                "action": "install_lldpd",
-            })
 
         pm = self._pm_type()
 
@@ -538,8 +573,6 @@ class OpenWrtSystemMixin:
             pkg_names = [_pkg_name(ln, pm) for ln in upgradable]
             warnings.append({
                 "code": "updates_available",
-                "severity": "info",
-                "action": None,
                 "meta": {
                     "count": len(upgradable),
                     "packages": pkg_names[:10],
@@ -550,20 +583,12 @@ class OpenWrtSystemMixin:
         if pm == "opkg":
             auc_path = self._send_command("which auc 2>/dev/null").strip()
             if not auc_path:
-                warnings.append({
+                warnings.append({"code": "update_notifications_disabled"})
-                    "code": "update_notifications_disabled",
-                    "severity": "warning",
-                    "action": "install_auc",
-                })
 
         # 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({
+            warnings.append({"code": "no_base64"})
-                "code": "no_base64",
-                "severity": "warning",
-                "action": "install_coreutils_base64",
-            })
 
         return warnings
 

napalm_openwrt/vlan_mixin.py

@@ -17,6 +17,63 @@ import re
 from typing import Any
 
 
+def _split_uci_list(value: str) -> list[str]:
+    """Split a ``uci show`` option value into its items.
+
+    List options are rendered space-separated with each item quoted
+    (``ports='lan1:t' 'lan2:t'``); scalar options carry a single quoted value.
+    """
+    items = re.findall(r"'([^']*)'", value)
+    if items:
+        return [i for i in items if i]
+    stripped = value.strip().strip("'")
+    return [stripped] if stripped else []
+
+
+def _uci_scalar(value: str) -> str:
+    """Return the single unquoted value of a scalar ``uci show`` option."""
+    items = _split_uci_list(value)
+    return items[0] if items else ""
+
+
+def _parse_port_spec(spec: str) -> tuple[str, bool]:
+    """Return ``(port, is_tagged)`` for an OpenWrt bridge-vlan port spec.
+
+    ``eth0:t`` is a tagged member; ``eth0:u*`` (untagged + PVID), ``eth0:*``
+    and a bare ``eth0`` are untagged members.
+    """
+    port, _, flags = spec.partition(":")
+    return port, "t" in flags
+
+
+def _uci_sections_of_type(uci_out: str, sec_type: str) -> list[dict[str, str]]:
+    """Collect all ``network`` sections of *sec_type* from ``uci show network``.
+
+    Handles both anonymous (``network.@bridge-vlan[0]``) and named
+    (``network.apbr_vlan10``) sections — only the former was recognised before,
+    so tool- or hand-provisioned bridges were invisible.
+
+    Option values are kept raw (quotes included) so that list options stay
+    distinguishable from scalars; use :func:`_uci_scalar` or
+    :func:`_split_uci_list` to read them.
+    """
+    decl = re.compile(rf"^network\.(\S+)={re.escape(sec_type)}$")
+    option = re.compile(r"^network\.(\S+?)\.(\w+)=(.*)$")
+
+    section_ids: set[str] = set()
+    for line in uci_out.splitlines():
+        m = decl.match(line.strip())
+        if m:
+            section_ids.add(m.group(1))
+
+    sections: dict[str, dict[str, str]] = {sid: {} for sid in section_ids}
+    for line in uci_out.splitlines():
+        m = option.match(line.strip())
+        if m and m.group(1) in section_ids:
+            sections[m.group(1)][m.group(2)] = m.group(3).strip()
+    return list(sections.values())
+
+
 class OpenWrtVLANMixin:
     """Mixin providing VLAN and network-instance NAPALM getters."""
 
@@ -68,19 +125,28 @@ class OpenWrtVLANMixin:
                 if current_port not in vlans[vlan_id]["tagged"]:
                     vlans[vlan_id]["tagged"].append(current_port)
 
-        # Enrich with UCI VLAN names from explicit bridge-vlan sections
+        # Enrich with the configured membership from UCI bridge-vlan sections.
-        uci_entries: dict[str, dict[str, str]] = {}
+        # On devices whose BusyBox ships without the `bridge` utility this is
-        for line in uci_out.splitlines():
+        # the only source that names the physical ports at all — the
-            m = re.match(r"network\.@bridge-vlan\[(\d+)\]\.(\w+)='([^']*)'", line.strip())
+        # sub-interface fallback below can only ever report br-ap/br-ap.N.
-            if m:
+        # Runtime data from `bridge vlan show` describes what the kernel
-                idx, key, value = m.group(1), m.group(2), m.group(3)
+        # actually enforces, so it keeps precedence where both are present.
-                uci_entries.setdefault(idx, {})[key] = value
+        for entry in _uci_sections_of_type(uci_out, "bridge-vlan"):
-
+            if "vlan" not in entry:
-        for entry in uci_entries.values():
+                continue
-            if "vlan" in entry and "name" in entry:
+            try:
-                vlan_id = str(int(entry["vlan"]))
+                vlan_id = str(int(_uci_scalar(entry["vlan"])))
-                if vlan_id in vlans:
+            except ValueError:
-                    vlans[vlan_id]["name"] = entry["name"]
+                continue
+            vlan = vlans.setdefault(vlan_id, {"name": "", "tagged": [], "untagged": []})
+            uci_name = _uci_scalar(entry.get("name", ""))
+            if uci_name and not vlan["name"]:
+                vlan["name"] = uci_name
+            for spec in _split_uci_list(entry.get("ports", "")):
+                port, tagged = _parse_port_spec(spec)
+                if port in vlan["tagged"] or port in vlan["untagged"]:
+                    continue
+                vlan["tagged" if tagged else "untagged"].append(port)
 
         # Also derive VLAN names from UCI network interface sections that
         # reference subinterfaces like eth0.N or br-ap.N:

napalm_openwrt/wireless_mixin.py

@@ -39,6 +39,10 @@ class OpenWrtWirelessMixin:
         # Collect radio band info: radio0 → "2g", radio1 → "5g", …
         radio_bands: dict[str, str] = {}
         iface_entries: dict[str, dict[str, str]] = {}
+        # UCI list values (e.g. "list maclist 'AA:...'") repeat the same key
+        # across multiple lines — tracked separately since the single-value
+        # iface_entries dict would only keep the last one.
+        iface_maclists: dict[str, list[str]] = {}
 
         # First pass: identify named sections that are wifi-iface types and
         # collect radio band info.
@@ -72,13 +76,19 @@ class OpenWrtWirelessMixin:
             im = re.match(r"wireless\.@wifi-iface\[(\d+)\]\.(\w+)='([^']*)'", line_s)
             if im:
                 idx, key, val = im.group(1), im.group(2), im.group(3)
-                iface_entries.setdefault(idx, {})[key] = val
+                if key == "maclist":
+                    iface_maclists.setdefault(idx, []).append(val)
+                else:
+                    iface_entries.setdefault(idx, {})[key] = val
                 continue
             # named wifi-iface values: wireless.managed_family_2g.ssid='manivong'
             nm = re.match(r"wireless\.(\w+)\.(\w+)='([^']*)'", line_s)
             if nm and nm.group(1) in named_iface_sections:
                 section, key, val = nm.group(1), nm.group(2), nm.group(3)
-                iface_entries.setdefault(section, {})[key] = val
+                if key == "maclist":
+                    iface_maclists.setdefault(section, []).append(val)
+                else:
+                    iface_entries.setdefault(section, {})[key] = val
 
         def _band_label(radio: str) -> str:
             raw = radio_bands.get(radio, "").lower()
@@ -148,7 +158,8 @@ class OpenWrtWirelessMixin:
         result: dict[str, Any] = {}
         # Intermediate: ssid -> list of bands seen
         ssid_bands: dict[str, list[str]] = {}
-        for entry in iface_entries.values():
+        _ACL_MODE_MAP = {"allow": "whitelist", "deny": "blacklist"}
+        for idx, entry in iface_entries.items():
             ssid = entry.get("ssid")
             if not ssid:
                 continue
@@ -171,6 +182,8 @@ class OpenWrtWirelessMixin:
             _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
             key: str = entry.get("key", "") or ""
+            acl_mode = _ACL_MODE_MAP.get(entry.get("macfilter", ""), "off")
+            mac_list = sorted(set(iface_maclists.get(idx, [])))
 
             if ssid in result:
                 # Merge: append band if not already present
@@ -205,6 +218,12 @@ class OpenWrtWirelessMixin:
                 # key: keep first non-empty value seen
                 if key and not result[ssid].get("key"):
                     result[ssid]["key"] = key
+                # acl_mode/mac_list: keep first non-"off" value seen — all
+                # wifi-iface sections for one SSID carry identical ACL config
+                # after a push, so any explicit value wins over the default.
+                if acl_mode != "off" and result[ssid].get("acl_mode", "off") == "off":
+                    result[ssid]["acl_mode"] = acl_mode
+                    result[ssid]["mac_list"] = mac_list
             else:
                 ssid_bands[ssid] = [band] if band else []
                 result[ssid] = {
@@ -226,6 +245,8 @@ class OpenWrtWirelessMixin:
                     "disassoc_low_ack": disassoc_low_ack,
                     "max_inactivity": max_inactivity,
                     "key": key,
+                    "acl_mode": acl_mode,
+                    "mac_list": mac_list,
                 }
         return result
 
@@ -543,6 +564,47 @@ class OpenWrtWirelessMixin:
         self._send_command("uci commit wireless")
         self._send_command("wifi")
 
+    def push_mac_acl(self, ssid_name: str, mode: str, macs: list[str]) -> None:
+        """Rewrite macfilter mode + maclist entries on every wifi-iface matching *ssid_name*.
+
+        Full-rebuild, not diff — always deletes the existing maclist before
+        re-adding, so the result is idempotent regardless of prior state.
+
+        OpenWrt's wifi-scripts validator rejects any ``macfilter`` value other
+        than ``"allow"``/``"deny"`` outright (confirmed on real hardware:
+        setting ``macfilter='disable'`` puts netifd in a permanent restart
+        crash loop with the radio stuck down) — there is no "off" value; the
+        only way to disable filtering is to omit the option entirely.
+
+        A whitelist ("allow") with zero MACs blocks every client outright, so
+        it is treated as equivalent to "off" instead of being pushed as-is.
+
+        :param ssid_name: SSID name to match against ``option ssid`` on each wifi-iface section.
+        :param mode: ``"off"`` | ``"whitelist"`` | ``"blacklist"``.
+        :param macs: MAC addresses to set as the maclist. Only the entries for the
+            active mode's list are ever passed in — the caller resolves whitelist
+            vs. blacklist before calling.
+        """
+        effective_mode = "off" if (mode == "whitelist" and not macs) else mode
+        sections = self._send_command(
+            "uci show wireless | grep -oE '^wireless\\.[^.]+' | sort -u"
+        ).split()
+        for sec in sections:
+            ssid_val = self._send_command(f"uci -q get {sec}.ssid 2>/dev/null || true").strip()
+            if ssid_val != ssid_name:
+                continue
+            if effective_mode == "off":
+                self._send_command(f"uci -q delete {sec}.macfilter || true")
+                self._send_command(f"uci -q delete {sec}.maclist || true")
+                continue
+            uci_mode = "allow" if effective_mode == "whitelist" else "deny"
+            self._send_command(f"uci set {sec}.macfilter='{uci_mode}'")
+            self._send_command(f"uci delete {sec}.maclist 2>/dev/null || true")
+            for mac in macs:
+                self._send_command(f"uci add_list {sec}.maclist='{mac}'")
+        self._send_command("uci commit wireless")
+        self._send_command("wifi reload")
+
     def get_radio_status(self) -> dict[str, Any]:
         """Return radio status from UCI and iwinfo.
 

pyproject.toml

@@ -28,6 +28,7 @@ dependencies = [
     "napalm>=4.0.0",
     "napalm_device_types>=0.1.0",
     "netmiko>=4.0.0",
+    "paramiko>=5.0.0",  # CVE-2026-44405
     "netaddr",
 ]
 

tests/unit/test_driver.py

@@ -154,9 +154,14 @@ class TestGetFacts:
         facts = driver.get_facts()
         assert set(facts.keys()) == {
             "vendor", "model", "hostname", "fqdn", "os_version",
-            "serial_number", "uptime", "interface_list",
+            "serial_number", "uptime", "interface_list", "number_of_interfaces",
         }
 
+    def test_number_of_interfaces_matches_list(self, driver):
+        driver._send_command = self._make_send()
+        facts = driver.get_facts()
+        assert facts["number_of_interfaces"] == len(facts["interface_list"])
+
     def test_vendor(self, driver):
         driver._send_command = self._make_send()
         assert driver.get_facts()["vendor"] == "OpenWrt"
@@ -350,17 +355,26 @@ class TestGetVlans:
         result = driver.get_vlans()
         for vlan_data in result.values():
             assert "name" in vlan_data
-            assert "interfaces" in vlan_data
+            assert "tagged" in vlan_data
+            assert "untagged" in vlan_data
 
-    def test_interfaces_for_vlan10(self, driver):
+    def test_pvid_port_is_untagged_member(self, driver):
         driver._send_command = self._send
         result = driver.get_vlans()
-        assert set(result["10"]["interfaces"]) == {"eth0", "br-lan"}
+        assert set(result["1"]["untagged"]) == {"eth0", "br-lan"}
+        assert result["1"]["tagged"] == []
 
-    def test_interfaces_for_vlan20(self, driver):
+    def test_continuation_lines_are_tagged_members(self, driver):
         driver._send_command = self._send
         result = driver.get_vlans()
-        assert set(result["20"]["interfaces"]) == {"eth0", "br-lan", "eth1"}
+        assert set(result["10"]["tagged"]) == {"eth0", "br-lan"}
+        assert result["10"]["untagged"] == []
+
+    def test_same_vlan_can_mix_tagged_and_untagged_ports(self, driver):
+        driver._send_command = self._send
+        result = driver.get_vlans()
+        assert set(result["20"]["tagged"]) == {"eth0", "br-lan"}
+        assert set(result["20"]["untagged"]) == {"eth1"}
 
     def test_uci_names_applied(self, driver):
         driver._send_command = self._send
@@ -373,11 +387,18 @@ class TestGetVlans:
         result = driver.get_vlans()
         assert result["1"]["name"] == ""
 
-    def test_no_duplicate_interfaces(self, driver):
+    def test_no_duplicate_ports(self, driver):
         driver._send_command = self._send
         result = driver.get_vlans()
         for vlan_data in result.values():
-            assert len(vlan_data["interfaces"]) == len(set(vlan_data["interfaces"]))
+            for key in ("tagged", "untagged"):
+                assert len(vlan_data[key]) == len(set(vlan_data[key]))
+
+    def test_port_is_never_both_tagged_and_untagged(self, driver):
+        driver._send_command = self._send
+        result = driver.get_vlans()
+        for vlan_data in result.values():
+            assert not set(vlan_data["tagged"]) & set(vlan_data["untagged"])
 
     def test_empty_bridge_output(self, driver):
         driver._send_command = lambda cmd, **kw: ""
@@ -385,6 +406,126 @@ class TestGetVlans:
         assert result == {}
 
 
+# A VLAN-aware bridge as configured by an AP profile: management VLAN untagged
+# (native/PVID) on the uplink, SSID VLANs tagged.  Sections are *named*, which
+# is what OpenWrt writes for a hand-built or tool-provisioned bridge.
+UCI_NAMED_BRIDGE_VLANS = """\
+network.apbr=device
+network.apbr.name='br-ap'
+network.apbr.type='bridge'
+network.apbr.ports='eth0'
+network.apbr.vlan_filtering='1'
+network.apbr_vlan10=bridge-vlan
+network.apbr_vlan10.device='br-ap'
+network.apbr_vlan10.vlan='10'
+network.apbr_vlan10.ports='eth0:u*'
+network.apbr_vlan30=bridge-vlan
+network.apbr_vlan30.device='br-ap'
+network.apbr_vlan30.vlan='30'
+network.apbr_vlan30.ports='eth0:t'
+network.mgmt=interface
+network.mgmt.device='br-ap.10'
+network.mgmt.proto='dhcp'
+network.esche=interface
+network.esche.device='br-ap.30'
+network.esche.proto='none'
+"""
+
+IP_LINK_AP = """\
+1: lo: <LOOPBACK,UP,LOWER_UP> mtu 65536 qdisc noqueue
+2: eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc fq_codel master br-ap
+3: br-ap: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc noqueue
+4: br-ap.10@br-ap: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc noqueue
+5: br-ap.30@br-ap: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc noqueue
+"""
+
+
+class TestGetVlansWithoutBridgeBinary:
+    """Membership must come from UCI when `bridge` is absent (BusyBox-only APs).
+
+    Devices such as the Sophos AP100 ship BusyBox without the `bridge`/`ip-full`
+    packages, so `bridge vlan show` fails.  Falling back to sub-interface
+    topology alone reports `br-ap`/`br-ap.10` and never names the uplink port,
+    which is the only port whose tagging actually matters.
+    """
+
+    def _send(self, cmd, **kw):
+        if "bridge vlan" in cmd:
+            return "ash: bridge: not found"
+        if "uci show network" in cmd:
+            return UCI_NAMED_BRIDGE_VLANS
+        if "ip link show" in cmd:
+            return IP_LINK_AP
+        return ""
+
+    def test_untagged_pvid_uplink_membership(self, driver):
+        driver._send_command = self._send
+        result = driver.get_vlans()
+        assert "eth0" in result["10"]["untagged"]
+        assert "eth0" not in result["10"]["tagged"]
+
+    def test_tagged_uplink_membership(self, driver):
+        driver._send_command = self._send
+        result = driver.get_vlans()
+        assert "eth0" in result["30"]["tagged"]
+        assert "eth0" not in result["30"]["untagged"]
+
+    def test_named_sections_are_parsed(self, driver):
+        """The old parser only matched anonymous @bridge-vlan[N] sections."""
+        driver._send_command = self._send
+        result = driver.get_vlans()
+        assert {"10", "30"} <= set(result)
+
+    def test_shell_error_is_not_parsed_as_membership(self, driver):
+        driver._send_command = self._send
+        result = driver.get_vlans()
+        for vlan_data in result.values():
+            assert "ash:" not in vlan_data["tagged"] + vlan_data["untagged"]
+
+    def test_name_falls_back_to_interface_section(self, driver):
+        driver._send_command = self._send
+        result = driver.get_vlans()
+        assert result["10"]["name"] == "mgmt"
+        assert result["30"]["name"] == "esche"
+
+    def test_runtime_membership_wins_over_uci(self, driver):
+        """`bridge vlan show` describes what the kernel actually does."""
+
+        def _send(cmd, **kw):
+            if "bridge vlan" in cmd:
+                return "port              vlan-id\neth0              10 PVID Egress Untagged\n"
+            if "uci show network" in cmd:
+                # UCI claims tagged; the kernel says untagged.
+                return (
+                    "network.apbr_vlan10=bridge-vlan\n"
+                    "network.apbr_vlan10.device='br-ap'\n"
+                    "network.apbr_vlan10.vlan='10'\n"
+                    "network.apbr_vlan10.ports='eth0:t'\n"
+                )
+            return ""
+
+        driver._send_command = _send
+        result = driver.get_vlans()
+        assert "eth0" in result["10"]["untagged"]
+        assert "eth0" not in result["10"]["tagged"]
+
+    def test_multi_port_list_is_split(self, driver):
+        def _send(cmd, **kw):
+            if "uci show network" in cmd:
+                return (
+                    "network.brv=bridge-vlan\n"
+                    "network.brv.device='br-lan'\n"
+                    "network.brv.vlan='20'\n"
+                    "network.brv.ports='lan1:t' 'lan2:t' 'lan3'\n"
+                )
+            return ""
+
+        driver._send_command = _send
+        result = driver.get_vlans()
+        assert set(result["20"]["tagged"]) == {"lan1", "lan2"}
+        assert set(result["20"]["untagged"]) == {"lan3"}
+
+
 class TestConfigManagement:
     def test_load_merge_candidate(self, driver):
         driver.load_merge_candidate(config="uci set system.@system[0].hostname='MyRouter'")
@@ -1099,3 +1240,630 @@ class TestPushRadioChannel:
         commit_idx = next(i for i, c in enumerate(issued) if "commit" in c)
         wifi_idx = next(i for i, c in enumerate(issued) if c.strip() == "wifi")
         assert commit_idx < wifi_idx
+
+
+UCI_WIRELESS_ACL = """\
+wireless.radio0=wifi-device
+wireless.radio0.band='2g'
+wireless.radio1=wifi-device
+wireless.radio1.band='5g'
+wireless.@wifi-iface[0]=wifi-iface
+wireless.@wifi-iface[0].device='radio0'
+wireless.@wifi-iface[0].ssid='CorpWiFi'
+wireless.@wifi-iface[0].encryption='psk2'
+wireless.@wifi-iface[0].macfilter='allow'
+wireless.@wifi-iface[0].maclist='AA:BB:CC:DD:EE:01'
+wireless.@wifi-iface[0].maclist='AA:BB:CC:DD:EE:02'
+wireless.@wifi-iface[1]=wifi-iface
+wireless.@wifi-iface[1].device='radio1'
+wireless.@wifi-iface[1].ssid='CorpWiFi'
+wireless.@wifi-iface[1].encryption='psk2'
+wireless.@wifi-iface[2]=wifi-iface
+wireless.@wifi-iface[2].device='radio0'
+wireless.@wifi-iface[2].ssid='GuestNet'
+wireless.@wifi-iface[2].encryption='none'
+"""
+
+
+class TestGetSsidsAcl:
+    """Tests for the macfilter/maclist parsing in OpenWrtWirelessMixin.get_ssids()."""
+
+    def _send(self, cmd, **kw):
+        if cmd.strip() == "uci show wireless":
+            return UCI_WIRELESS_ACL
+        return ""
+
+    def test_whitelist_mode_parsed(self, driver):
+        driver._send_command = self._send
+        result = driver.get_ssids()
+        assert result["CorpWiFi"]["acl_mode"] == "whitelist"
+
+    def test_maclist_multiple_entries_parsed(self, driver):
+        """UCI list values repeat the same key across lines — must not overwrite."""
+        driver._send_command = self._send
+        result = driver.get_ssids()
+        assert result["CorpWiFi"]["mac_list"] == ["AA:BB:CC:DD:EE:01", "AA:BB:CC:DD:EE:02"]
+
+    def test_acl_mode_merged_across_radios(self, driver):
+        """Only wifi-iface[0] has macfilter set; wifi-iface[1] (same SSID) must inherit it."""
+        driver._send_command = self._send
+        result = driver.get_ssids()
+        # Both wifi-iface sections belong to CorpWiFi — the merged result carries one acl_mode.
+        assert result["CorpWiFi"]["acl_mode"] == "whitelist"
+        assert result["CorpWiFi"]["mac_list"] == ["AA:BB:CC:DD:EE:01", "AA:BB:CC:DD:EE:02"]
+
+    def test_no_macfilter_defaults_to_off(self, driver):
+        driver._send_command = self._send
+        result = driver.get_ssids()
+        assert result["GuestNet"]["acl_mode"] == "off"
+        assert result["GuestNet"]["mac_list"] == []
+
+    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 ""
+        result = driver.get_ssids()
+        assert result["CorpWiFi"]["acl_mode"] == "blacklist"
+
+
+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):
+        ssid_by_section = ssid_by_section or {
+            "wireless.@wifi-iface[0]": "CorpWiFi",
+            "wireless.@wifi-iface[1]": "GuestNet",
+        }
+        issued: list[str] = []
+
+        def _send(cmd, **kw):
+            issued.append(cmd)
+            if "grep -oE" in cmd and "sort -u" in cmd:
+                return sections
+            for sec, ssid in ssid_by_section.items():
+                if f"uci -q get {sec}.ssid" in cmd:
+                    return ssid
+            return ""
+
+        return _send, issued
+
+    def test_whitelist_sets_macfilter_allow(self, driver):
+        send, issued = self._make_send()
+        driver._send_command = send
+        driver.push_mac_acl("CorpWiFi", "whitelist", ["AA:BB:CC:DD:EE:01"])
+        assert any("wireless.@wifi-iface[0].macfilter='allow'" in c for c in issued)
+
+    def test_blacklist_sets_macfilter_deny(self, driver):
+        send, issued = self._make_send()
+        driver._send_command = send
+        driver.push_mac_acl("CorpWiFi", "blacklist", ["AA:BB:CC:DD:EE:01"])
+        assert any("wireless.@wifi-iface[0].macfilter='deny'" in c for c in issued)
+
+    def test_off_deletes_macfilter_option_instead_of_setting_disable(self, driver):
+        """OpenWrt's validator rejects macfilter='disable' outright (confirmed on
+        real hardware — it puts netifd in a permanent restart crash loop with
+        the radio stuck down). "off" must delete the option, never set it."""
+        send, issued = self._make_send()
+        driver._send_command = send
+        driver.push_mac_acl("CorpWiFi", "off", [])
+        assert not any("macfilter='disable'" in c for c in issued)
+        assert any("delete wireless.@wifi-iface[0].macfilter" in c for c in issued)
+        assert any("delete wireless.@wifi-iface[0].maclist" in c for c in issued)
+        assert not any("add_list wireless.@wifi-iface[0].maclist" in c for c in issued)
+
+    def test_whitelist_with_zero_macs_treated_as_off(self, driver):
+        """An empty whitelist blocks every client outright — must not be pushed
+        as macfilter='allow' with an empty list."""
+        send, issued = self._make_send()
+        driver._send_command = send
+        driver.push_mac_acl("CorpWiFi", "whitelist", [])
+        assert not any("macfilter='allow'" in c for c in issued)
+        assert any("delete wireless.@wifi-iface[0].macfilter" in c for c in issued)
+
+    def test_blacklist_with_zero_macs_still_pushed(self, driver):
+        """An empty blacklist is safe (blocks nobody) — no guard needed."""
+        send, issued = self._make_send()
+        driver._send_command = send
+        driver.push_mac_acl("CorpWiFi", "blacklist", [])
+        assert any("macfilter='deny'" in c for c in issued)
+
+    def test_maclist_entries_added(self, driver):
+        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("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)
+        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):
+        """GuestNet section must not be modified when pushing CorpWiFi's ACL."""
+        send, issued = self._make_send()
+        driver._send_command = send
+        driver.push_mac_acl("CorpWiFi", "whitelist", ["AA:BB:CC:DD:EE:01"])
+        assert not any("wireless.@wifi-iface[1].macfilter" in c for c in issued)
+
+    def test_issues_uci_commit_and_wifi_reload(self, driver):
+        send, issued = self._make_send()
+        driver._send_command = send
+        driver.push_mac_acl("CorpWiFi", "whitelist", ["AA:BB:CC:DD:EE:01"])
+        assert any("uci commit wireless" in c for c in issued)
+        assert any(c.strip() == "wifi reload" for c in issued)
+
+
+# ---------------------------------------------------------------------------
+# fix_snmp — firewall zone handling
+# ---------------------------------------------------------------------------
+
+UCI_FIREWALL_HEALTHY = """\
+firewall.@defaults[0]=defaults
+firewall.@defaults[0].input='REJECT'
+firewall.@zone[0]=zone
+firewall.@zone[0].name='lan'
+firewall.@zone[0].network='lan'
+firewall.@zone[0].input='ACCEPT'
+firewall.@zone[1]=zone
+firewall.@zone[1].name='wan'
+firewall.@zone[1].network='wan' 'wan6'
+firewall.@zone[1].input='REJECT'
+firewall.@rule[0]=rule
+firewall.@rule[0].name='Allow-DHCP-Renew'
+firewall.@rule[0].src='wan'
+firewall.@rule[0].dest_port='68'
+"""
+
+# The zone that owns the management network lost its 'name' — fw4 skips the
+# whole section and every rule pointing at it.
+UCI_FIREWALL_NAMELESS_ZONE = """\
+firewall.@defaults[0]=defaults
+firewall.@defaults[0].input='REJECT'
+firewall.@zone[0]=zone
+firewall.@zone[0].network='lan'
+firewall.@zone[0].input='ACCEPT'
+firewall.@rule[0]=rule
+firewall.@rule[0].name='Allow-DHCP-Renew'
+firewall.@rule[0].src='wan'
+"""
+
+# A dedicated management zone — the AP layout the action is meant to handle.
+UCI_FIREWALL_MGMT_ZONE = """\
+firewall.@zone[0]=zone
+firewall.@zone[0].name='lan'
+firewall.@zone[0].network='lan'
+firewall.@zone[1]=zone
+firewall.@zone[1].name='mgmt'
+firewall.@zone[1].network='mgmt'
+firewall.@zone[1].input='REJECT'
+firewall.@rule[0]=rule
+firewall.@rule[0].name='Allow-SSH'
+firewall.@rule[0].src='mgmt'
+firewall.@rule[0].dest_port='22'
+"""
+
+UCI_NETWORK_STATIC = """\
+network.loopback=interface
+network.loopback.device='lo'
+network.lan=interface
+network.lan.device='br-lan'
+network.lan.proto='static'
+network.lan.ipaddr='192.168.1.1'
+network.mgmt=interface
+network.mgmt.device='br-lan.9'
+network.mgmt.proto='static'
+network.mgmt.ipaddr='10.10.0.5'
+"""
+
+UCI_NETWORK_DHCP = """\
+network.lan=interface
+network.lan.device='br-lan'
+network.lan.proto='dhcp'
+"""
+
+IP_ADDR_BRLAN = """\
+1: lo    inet 127.0.0.1/8 scope host lo\\       valid_lft forever preferred_lft forever
+7: br-lan    inet 10.10.0.5/24 brd 10.10.0.255 scope global br-lan\\       valid_lft forever
+"""
+
+
+class _FakeShell:
+    """Collects issued commands and answers them from a canned config."""
+
+    def __init__(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
+        self.ssh_connection = ssh_connection
+        self.reload_out = reload_out
+        self.nft_hits = nft_hits
+        self.listen_hits = listen_hits
+        self.snmpd = snmpd
+        self.issued: list[str] = []
+
+    def __call__(self, cmd, **kw):
+        self.issued.append(cmd)
+        if cmd.startswith("uci show firewall"):
+            return self.firewall
+        if cmd.startswith("uci show network"):
+            return self.network
+        if "$SSH_CONNECTION" in cmd:
+            return self.ssh_connection
+        if "ip -o -4 addr" in cmd:
+            return self.ip_addr
+        if "fw4 reload" in cmd or "firewall reload" in cmd:
+            return self.reload_out
+        if "dport 161" in cmd:
+            return self.nft_hits
+        if ":161" in cmd:
+            return self.listen_hits
+        if "snmpd status" in cmd:
+            return self.snmpd
+        return ""
+
+
+class TestFixSnmpZoneDetection:
+    """_action_fix_snmp() 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."""
+        shell = _FakeShell(
+            firewall=UCI_FIREWALL_MGMT_ZONE,
+            network=UCI_NETWORK_STATIC,
+            ssh_connection="10.10.0.1 51234 10.10.0.5 22",
+        )
+        driver._send_command = shell
+        result = driver._action_fix_snmp()
+        assert "Management zone: 'mgmt'" in result["output"]
+        assert any("firewall.allow_snmp_from_mgmt.src='mgmt'" in c for c in shell.issued)
+
+    def test_anonymous_ssh_rule_does_not_decide_the_zone(self, driver):
+        """The old per-line 'src= and ssh' heuristic never matched anonymous rules."""
+        shell = _FakeShell(
+            firewall=UCI_FIREWALL_HEALTHY,
+            network=UCI_NETWORK_STATIC,
+            ssh_connection="192.168.1.50 5000 192.168.1.1 22",
+        )
+        driver._send_command = shell
+        result = driver._action_fix_snmp()
+        assert "Management zone: 'lan'" in result["output"]
+
+    def test_dhcp_mgmt_address_resolved_via_l3_device(self, driver):
+        """No ipaddr in UCI → resolve address → device → network section → zone."""
+        shell = _FakeShell(
+            firewall=UCI_FIREWALL_HEALTHY,
+            network=UCI_NETWORK_DHCP,
+            ip_addr=IP_ADDR_BRLAN,
+            ssh_connection="10.10.0.1 51234 10.10.0.5 22",
+        )
+        driver._send_command = shell
+        result = driver._action_fix_snmp()
+        assert "Management zone: 'lan'" in result["output"]
+
+    def test_nameless_zone_is_reported_and_action_fails(self, driver):
+        """A zone without 'name' is skipped by fw4 — say so instead of writing a dead rule."""
+        shell = _FakeShell(
+            firewall=UCI_FIREWALL_NAMELESS_ZONE,
+            network=UCI_NETWORK_STATIC,
+            ssh_connection="192.168.1.50 5000 192.168.1.1 22",
+        )
+        driver._send_command = shell
+        result = driver._action_fix_snmp()
+        assert result["success"] is False
+        assert "@zone[0]" in result["output"]
+        assert "name" in result["output"]
+
+    def test_nameless_zone_does_not_get_a_rule_written(self, driver):
+        """No SNMP rule may be committed while the owning zone is invalid."""
+        shell = _FakeShell(
+            firewall=UCI_FIREWALL_NAMELESS_ZONE,
+            network=UCI_NETWORK_STATIC,
+            ssh_connection="192.168.1.50 5000 192.168.1.1 22",
+        )
+        driver._send_command = shell
+        driver._action_fix_snmp()
+        assert not any("allow_snmp" in c for c in shell.issued)
+
+
+class TestFixSnmpRuleWriting:
+    """The rule must be (re)written idempotently, not skipped when present."""
+
+    def test_existing_rule_is_repaired_not_skipped(self, driver):
+        """A rule that exists but lacks src must be rewritten, not left broken."""
+        broken = UCI_FIREWALL_HEALTHY + (
+            "firewall.allow_snmp_from_lan=rule\n"
+            "firewall.allow_snmp_from_lan.name='Allow-SNMP-from-lan'\n"
+            "firewall.allow_snmp_from_lan.dest_port='161'\n"
+        )
+        shell = _FakeShell(
+            firewall=broken,
+            network=UCI_NETWORK_STATIC,
+            ssh_connection="192.168.1.50 5000 192.168.1.1 22",
+        )
+        driver._send_command = shell
+        driver._action_fix_snmp()
+        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):
+        """Anonymous sections shift on delete — descending order keeps the keys valid."""
+        stale = UCI_FIREWALL_HEALTHY + (
+            "firewall.@rule[1]=rule\n"
+            "firewall.@rule[1].name='Allow-SNMP'\n"
+            "firewall.@rule[1].dest_port='161'\n"
+            "firewall.@rule[2]=rule\n"
+            "firewall.@rule[2].name='Allow-SNMP-netOrk'\n"
+            "firewall.@rule[2].dest_port='161'\n"
+        )
+        shell = _FakeShell(
+            firewall=stale,
+            network=UCI_NETWORK_STATIC,
+            ssh_connection="192.168.1.50 5000 192.168.1.1 22",
+        )
+        driver._send_command = shell
+        driver._action_fix_snmp()
+        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]")
+
+    def test_deletion_is_committed(self, driver):
+        """Old code staged deletes in /tmp/.uci and never committed them."""
+        stale = UCI_FIREWALL_HEALTHY + (
+            "firewall.snmp_netork=rule\n"
+            "firewall.snmp_netork.name='Allow-SNMP-from-mgmt'\n"
+            "firewall.snmp_netork.dest_port='161'\n"
+        )
+        shell = _FakeShell(
+            firewall=stale,
+            network=UCI_NETWORK_STATIC,
+            ssh_connection="192.168.1.50 5000 192.168.1.1 22",
+        )
+        driver._send_command = shell
+        driver._action_fix_snmp()
+        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):
+        """The pre-0.x rule was named 'Allow-SNMP-from-mgmt' — hyphens, not underscores."""
+        stale = UCI_FIREWALL_HEALTHY + (
+            "firewall.snmp_netork=rule\n"
+            "firewall.snmp_netork.name='Allow-SNMP-from-mgmt'\n"
+            "firewall.snmp_netork.src='*'\n"
+            "firewall.snmp_netork.dest_port='161'\n"
+        )
+        shell = _FakeShell(
+            firewall=stale,
+            network=UCI_NETWORK_STATIC,
+            ssh_connection="192.168.1.50 5000 192.168.1.1 22",
+        )
+        driver._send_command = shell
+        driver._action_fix_snmp()
+        assert any("snmp_netork" in c and "delete" in c for c in shell.issued)
+
+
+class TestFixSnmpReloadVerification:
+    """A failing fw4 reload must fail the action, not be swallowed."""
+
+    FW4_ZONE_ERROR = (
+        "Section @zone[0] option 'name' is mandatory but not set\n"
+        "Section @zone[0] skipped due to invalid options\n"
+        "Section @rule[0] references unknown zone 'lan'\n"
+        "Section @rule[0] skipped due to invalid options"
+    )
+
+    def test_reload_error_fails_the_action(self, driver):
+        shell = _FakeShell(
+            firewall=UCI_FIREWALL_HEALTHY,
+            network=UCI_NETWORK_STATIC,
+            ssh_connection="192.168.1.50 5000 192.168.1.1 22",
+            reload_out=self.FW4_ZONE_ERROR,
+        )
+        driver._send_command = shell
+        result = driver._action_fix_snmp()
+        assert result["success"] is False
+
+    def test_reload_output_is_not_truncated(self, driver):
+        """The old 120-char cap hid the 'references unknown zone' line."""
+        shell = _FakeShell(
+            firewall=UCI_FIREWALL_HEALTHY,
+            network=UCI_NETWORK_STATIC,
+            ssh_connection="192.168.1.50 5000 192.168.1.1 22",
+            reload_out=self.FW4_ZONE_ERROR,
+        )
+        driver._send_command = shell
+        result = driver._action_fix_snmp()
+        assert "references unknown zone" in result["output"]
+
+    def test_missing_live_rule_fails_the_action(self, driver):
+        """snmpd up + clean reload, but no udp/161 accept in the packet filter."""
+        shell = _FakeShell(
+            firewall=UCI_FIREWALL_HEALTHY,
+            network=UCI_NETWORK_STATIC,
+            ssh_connection="192.168.1.50 5000 192.168.1.1 22",
+            nft_hits="0",
+        )
+        driver._send_command = shell
+        result = driver._action_fix_snmp()
+        assert result["success"] is False
+
+    def test_snmpd_not_listening_fails_the_action(self, driver):
+        shell = _FakeShell(
+            firewall=UCI_FIREWALL_HEALTHY,
+            network=UCI_NETWORK_STATIC,
+            ssh_connection="192.168.1.50 5000 192.168.1.1 22",
+            listen_hits="0",
+        )
+        driver._send_command = shell
+        result = driver._action_fix_snmp()
+        assert result["success"] is False
+
+    def test_fully_healthy_run_succeeds(self, driver):
+        shell = _FakeShell(
+            firewall=UCI_FIREWALL_HEALTHY,
+            network=UCI_NETWORK_STATIC,
+            ssh_connection="192.168.1.50 5000 192.168.1.1 22",
+        )
+        driver._send_command = shell
+        result = driver._action_fix_snmp()
+        assert result["success"] is True
+
+
+class TestUciSectionParser:
+    """_parse_uci_sections() underpins all of the above."""
+
+    def test_section_type_captured(self, driver):
+        parsed = driver._parse_uci_sections(UCI_FIREWALL_HEALTHY)
+        assert parsed["@zone[0]"]["_type"] == "zone"
+
+    def test_option_value_unquoted_on_read(self, driver):
+        parsed = driver._parse_uci_sections(UCI_FIREWALL_HEALTHY)
+        assert driver._uci_scalar(parsed["@zone[0]"]["name"]) == "lan"
+
+    def test_list_values_split_into_tokens(self, driver):
+        parsed = driver._parse_uci_sections(UCI_FIREWALL_HEALTHY)
+        assert driver._uci_tokens(parsed["@zone[1]"]["network"]) == ["wan", "wan6"]
+
+    def test_value_containing_equals_is_kept_whole(self, driver):
+        parsed = driver._parse_uci_sections("firewall.x=rule\nfirewall.x.name='a=b'\n")
+        assert driver._uci_scalar(parsed["x"]["name"]) == "a=b"
+
+    def test_blank_and_malformed_lines_ignored(self, driver):
+        parsed = driver._parse_uci_sections("\n\nnot a uci line\nfirewall.x=rule\n")
+        assert list(parsed) == ["x"]
+
+
+# ---------------------------------------------------------------------------
+# get_system_config — remote syslog, LuCI and bridge STP (netOrk #164)
+# ---------------------------------------------------------------------------
+
+UCI_SYSTEM_WITH_SYSLOG = """\
+system.@system[0]=system
+system.@system[0].hostname='ap-eze-Garten'
+system.@system[0].timezone='CET-1CEST,M3.5.0,M10.5.0/3'
+system.@system[0].zonename='Europe/Berlin'
+system.@system[0].log_remote='1'
+system.@system[0].log_ip='10.10.40.2'
+system.@system[0].log_port='5514'
+system.@system[0].log_proto='udp'
+system.ntp=timeserver
+system.ntp.server='0.openwrt.pool.ntp.org' '1.openwrt.pool.ntp.org'
+"""
+
+UCI_SYSTEM_NO_SYSLOG = """\
+system.@system[0]=system
+system.@system[0].hostname='ap-eze-Parkplatz'
+system.@system[0].timezone='GMT0'
+system.@system[0].zonename='UTC'
+system.ntp=timeserver
+system.ntp.server='0.openwrt.pool.ntp.org'
+"""
+
+UCI_DROPBEAR_SYS = """\
+dropbear.@dropbear[0]=dropbear
+dropbear.@dropbear[0].Port='22'
+dropbear.@dropbear[0].PasswordAuth='on'
+dropbear.@dropbear[0].RootPasswordAuth='on'
+"""
+
+UCI_NETWORK_STP_ON = """\
+network.ap_bridge=device
+network.ap_bridge.name='br-ap'
+network.ap_bridge.type='bridge'
+network.ap_bridge.stp='1'
+network.lan=interface
+"""
+
+UCI_NETWORK_STP_OFF = """\
+network.ap_bridge=device
+network.ap_bridge.name='br-ap'
+network.ap_bridge.type='bridge'
+network.lan=interface
+"""
+
+
+def _system_send(system_out=UCI_SYSTEM_WITH_SYSLOG, luci_out="1", network_out=UCI_NETWORK_STP_ON):
+    """Dispatch _send_command by the command it receives."""
+
+    def _send(cmd, **kw):
+        if "uci show system" in cmd:
+            return system_out
+        if "uci show dropbear" in cmd:
+            return UCI_DROPBEAR_SYS
+        if "uci show network" in cmd:
+            return network_out
+        if "uhttpd" in cmd:
+            return luci_out
+        return ""
+
+    return _send
+
+
+class TestGetSystemConfigSyslog:
+    def test_remote_syslog_target_is_reported(self, driver):
+        driver._send_command = _system_send()
+        cfg = driver.get_system_config()
+        assert cfg["syslog_remote"] is True
+        assert cfg["syslog_ip"] == "10.10.40.2"
+        assert cfg["syslog_port"] == 5514
+
+    def test_absent_syslog_reports_as_disabled(self, driver):
+        driver._send_command = _system_send(UCI_SYSTEM_NO_SYSLOG)
+        cfg = driver.get_system_config()
+        assert cfg["syslog_remote"] is False
+        assert cfg["syslog_ip"] == ""
+
+    def test_log_ip_without_log_remote_is_not_active(self, driver):
+        # OpenWrt only ships logs when log_remote is set, whatever log_ip says.
+        out = UCI_SYSTEM_NO_SYSLOG + "system.@system[0].log_ip='10.10.40.2'\n"
+        driver._send_command = _system_send(out)
+        assert driver.get_system_config()["syslog_remote"] is False
+
+    def test_port_falls_back_to_the_openwrt_default(self, driver):
+        out = UCI_SYSTEM_NO_SYSLOG + (
+            "system.@system[0].log_remote='1'\nsystem.@system[0].log_ip='10.10.40.2'\n"
+        )
+        driver._send_command = _system_send(out)
+        assert driver.get_system_config()["syslog_port"] == 514
+
+    def test_existing_fields_are_untouched(self, driver):
+        driver._send_command = _system_send()
+        cfg = driver.get_system_config()
+        assert cfg["timezone"] == "CET-1CEST,M3.5.0,M10.5.0/3"
+        assert cfg["zonename"] == "Europe/Berlin"
+        assert cfg["dropbear_port"] == 22
+        assert cfg["ntp_servers"] == ["0.openwrt.pool.ntp.org", "1.openwrt.pool.ntp.org"]
+
+
+class TestGetSystemConfigLuci:
+    def test_enabled_uhttpd_reports_luci_as_reachable(self, driver):
+        driver._send_command = _system_send(luci_out="1")
+        assert driver.get_system_config()["luci_enabled"] is True
+
+    def test_disabled_uhttpd_reports_luci_as_unreachable(self, driver):
+        driver._send_command = _system_send(luci_out="0")
+        assert driver.get_system_config()["luci_enabled"] is False
+
+    def test_missing_uhttpd_reports_as_unreachable(self, driver):
+        # No uhttpd installed at all — LuCI cannot be served.
+        driver._send_command = _system_send(luci_out="")
+        assert driver.get_system_config()["luci_enabled"] is False
+
+
+class TestGetSystemConfigBridgeStp:
+    def test_stp_enabled_is_reported(self, driver):
+        driver._send_command = _system_send(network_out=UCI_NETWORK_STP_ON)
+        assert driver.get_system_config()["bridge_stp"] is True
+
+    def test_absent_stp_option_means_disabled(self, driver):
+        # UCI defaults stp to 0 when the option is not present.
+        driver._send_command = _system_send(network_out=UCI_NETWORK_STP_OFF)
+        assert driver.get_system_config()["bridge_stp"] is False
+
+    def test_no_ap_bridge_reports_none(self, driver):
+        # Nothing to have an opinion about — distinct from "STP is off".
+        driver._send_command = _system_send(network_out="network.lan=interface\n")
+        assert driver.get_system_config()["bridge_stp"] is None