diff --git a/napalm_openwrt/openwrt.py b/napalm_openwrt/openwrt.py index 9c1fcc2..956bbd3 100644 --- a/napalm_openwrt/openwrt.py +++ b/napalm_openwrt/openwrt.py @@ -299,6 +299,57 @@ class OpenWrtDriver( 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 `` 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 +366,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 `` + `` — 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 - firewall management zone (typically named ``mgmt``) only allows - SSH/HTTP/HTTPS/ICMP by default and default-drops everything else. - snmpd runs but packets are rejected before reaching the process. + On OpenWrt the usual reason SNMP stays unreachable is the firewall: + snmpd listens, but the zone covering the management interface + default-drops everything except SSH/HTTP/HTTPS/ICMP. A rule only + 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 - management interface where SNMP needs to be open too). - 2. Removes any wrongly-scoped SNMP rule (one without src=). - 3. Adds a named UCI rule ``allow_snmp_from_`` with - ``src=`` so it ends up in the correct nftables chain. - 4. Commits and reloads fw4 immediately (no reboot needed). - 5. Ensures snmpd is enabled and running. + + 1. Resolves the management address from ``$SSH_CONNECTION``, maps it + to its ``network`` section and from there to the owning firewall + zone — no guessing from rule names. + 2. Aborts with a diagnosis when that zone has no ``option name``: + fw4 skips such a section, so any rule written against it is dead + on arrival. Naming the zone is left to the operator because an + 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_`` 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 ─────────────────────────────── - # 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). + # ── 1. Resolve the zone that owns the management address ───────── 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 - for line in fw_raw.splitlines(): - # Named rule pattern: firewall.allow_ssh_from_.src='' - if ".src=" in line and "ssh" in line.lower(): - zone_val = line.split("=", 1)[-1].strip().strip("'\"") - if zone_val: - mgmt_zone = zone_val - break + 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 ''}") - lines.append(f"[firewall] Management zone: {mgmt_zone!r}") - - # ── 2. Clean up any wrongly-scoped previous SNMP rule ──────────── - # A rule named Allow-SNMP without src= lands in the global input - # 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" + broken_zone = ( + self._zone_for_network(fw_sections, {key: key for key in nameless}, mgmt_net) + if mgmt_net + else "" ) - 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 +601,21 @@ class OpenWrtDriver( else: lines.append("[snmpd] Service already running") - # ── 6. Local probe (best-effort) ────────────────────────────────── - probe = self._send_command( - "snmpget -v2c -cpublic -t2 -r0 -Ov 127.0.0.1 1.3.6.1.2.1.1.1.0 2>&1" - " || echo 'snmp_client_not_available'" + # ── 6. Verify on the device instead of assuming success ─────────── + listening = self._grep_count( + self._send_command("ss -lun 2>/dev/null | grep -c ':161'") ) - if "snmp_client_not_available" in probe: - 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]}") + 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}") + + 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]: diff --git a/tests/unit/test_driver.py b/tests/unit/test_driver.py index 4f92d08..b10eb24 100644 --- a/tests/unit/test_driver.py +++ b/tests/unit/test_driver.py @@ -1395,3 +1395,342 @@ class TestPushMacAcl: 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"]