Christian Manivong
597a59fa39
fix_snmp reported success on APs where the rule never reached nftables. Five defects stacked up: 1. Zone detection required ".src=" and "ssh" in the same `uci show` line. UCI prints one option per line, so anonymous rules never matched and every device fell through to the hardcoded "lan" fallback. 2. That fallback was never checked against the zones that actually exist. On an AP whose zone section has no `option name`, fw4 skips the section, so `src='lan'` referenced a zone that was not there and the rule was dropped with it. 3. The "already present" guard was a substring test, so a rule written by an earlier broken run was skipped forever instead of repaired. 4. Stale-rule deletion never committed — the only `uci commit firewall` sat in the add branch that the guard had just skipped. 5. `fw4 reload` errors were swallowed by `|| true`, and with no local snmpget the action hardcoded success = True. Now: the management address comes from $SSH_CONNECTION and is mapped to its network section (via ipaddr, or via `ip -o -4 addr` -> device when the interface is DHCP-addressed) and from there to the owning zone. A zone section without a name aborts the action with the repair command rather than writing a dead rule — naming it is left to the operator, since an inert zone becoming active changes what the AP filters. Rules are written in full every run, stale ones are deleted highest anonymous index first (uci renumbers @rule[n] on delete) and committed, reload output is no longer truncated or ignored, and success is verified on the device via `ss -lun` and a udp/161 lookup in the live ruleset.
1737 lines
63 KiB
Python
1737 lines
63 KiB
Python
"""Unit tests for OpenWrtDriver — no real device required."""
|
|
|
|
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from napalm_openwrt.openwrt import OpenWrtDriver
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.fixture
|
|
def driver():
|
|
"""Return a driver instance with a mocked Netmiko connection."""
|
|
with patch("napalm_openwrt.openwrt.ConnectHandler"):
|
|
drv = OpenWrtDriver(
|
|
hostname="192.168.1.1",
|
|
username="root",
|
|
password="",
|
|
)
|
|
drv.device = MagicMock()
|
|
yield drv
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Sample command output (as would be returned by the device)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
OPENWRT_RELEASE = """\
|
|
DISTRIB_ID="OpenWrt"
|
|
DISTRIB_RELEASE="23.05.3"
|
|
DISTRIB_REVISION="r23809-234f1a2efa"
|
|
DISTRIB_TARGET="ath79/generic"
|
|
DISTRIB_ARCH="mips_24kc"
|
|
DISTRIB_CODENAME="Restoring Earth"
|
|
DISTRIB_TAINTS=""
|
|
"""
|
|
|
|
SYSINFO_MODEL = "TP-Link TL-WR1043N/ND v5"
|
|
|
|
UPTIME = "352467.12 345823.44"
|
|
|
|
IP_LINK_SHOW = """\
|
|
1: lo: <LOOPBACK,UP,LOWER_UP> mtu 65536 qdisc noqueue state UNKNOWN mode DEFAULT group default qlen 1000
|
|
link/loopback 00:00:00:00:00:00 brd 00:00:00:00:00:00
|
|
2: eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc fq_codel state UP mode DEFAULT group default qlen 1000
|
|
link/ether b0:95:75:aa:bb:cc brd ff:ff:ff:ff:ff:ff
|
|
3: eth1: <BROADCAST,MULTICAST> mtu 1500 qdisc noop state DOWN mode DEFAULT group default qlen 1000
|
|
link/ether b0:95:75:aa:bb:dd brd ff:ff:ff:ff:ff:ff
|
|
4: br-lan: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc noqueue state UP mode DEFAULT group default qlen 1000
|
|
link/ether b0:95:75:aa:bb:cc brd ff:ff:ff:ff:ff:ff
|
|
"""
|
|
|
|
IP_ADDR_SHOW = """\
|
|
1: lo: <LOOPBACK,UP,LOWER_UP> mtu 65536 qdisc noqueue state UNKNOWN group default qlen 1000
|
|
link/loopback 00:00:00:00:00:00 brd 00:00:00:00:00:00
|
|
inet 127.0.0.1/8 scope host lo
|
|
valid_lft forever preferred_lft forever
|
|
2: eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc fq_codel state UP group default qlen 1000
|
|
link/ether b0:95:75:aa:bb:cc brd ff:ff:ff:ff:ff:ff
|
|
4: br-lan: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc noqueue state UP group default qlen 1000
|
|
link/ether b0:95:75:aa:bb:cc brd ff:ff:ff:ff:ff:ff
|
|
inet 192.168.1.1/24 brd 192.168.1.255 scope global br-lan
|
|
valid_lft forever preferred_lft forever
|
|
inet6 fd00::1/64 scope global
|
|
valid_lft forever preferred_lft forever
|
|
"""
|
|
|
|
IP_NEIGH_SHOW = """\
|
|
192.168.1.100 dev br-lan lladdr aa:bb:cc:dd:ee:ff REACHABLE
|
|
192.168.1.101 dev br-lan lladdr 11:22:33:44:55:66 STALE
|
|
192.168.1.102 dev br-lan FAILED
|
|
"""
|
|
|
|
BRIDGE_FDB = """\
|
|
aa:bb:cc:dd:ee:ff dev br-lan master br-lan permanent
|
|
11:22:33:44:55:66 dev eth0 vlan 1 master br-lan
|
|
33:33:00:00:00:01 dev br-lan self permanent
|
|
"""
|
|
|
|
UCI_EXPORT = """\
|
|
package system
|
|
|
|
config system
|
|
\toption hostname 'OpenWrt'
|
|
\toption timezone 'UTC'
|
|
|
|
package network
|
|
|
|
config interface 'loopback'
|
|
\toption device 'lo'
|
|
\toption proto 'static'
|
|
\toption ipaddr '127.0.0.1'
|
|
\toption netmask '255.0.0.0'
|
|
|
|
config interface 'lan'
|
|
\toption device 'br-lan'
|
|
\toption proto 'static'
|
|
\toption ipaddr '192.168.1.1'
|
|
\toption netmask '255.255.255.0'
|
|
"""
|
|
|
|
LLDPCTL_KV = """\
|
|
lldp.eth0.via=LLDP
|
|
lldp.eth0.rid=1
|
|
lldp.eth0.age=0 day, 01:23:45
|
|
lldp.eth0.chassis.mac=00:aa:bb:cc:dd:ee
|
|
lldp.eth0.chassis.name=core-router
|
|
lldp.eth0.chassis.descr=RouterOS 7.x
|
|
lldp.eth0.chassis.mgmt-ip=10.0.0.1
|
|
lldp.eth0.chassis.cap.available=Router, Bridge
|
|
lldp.eth0.chassis.cap.enabled=Router
|
|
lldp.eth0.port.ifname=ether1
|
|
lldp.eth0.port.descr=uplink
|
|
"""
|
|
|
|
PROC_NET_DEV = """\
|
|
Inter-| Receive | Transmit
|
|
face |bytes packets errs drop fifo frame compressed multicast|bytes packets errs drop fifo colls carrier compressed
|
|
lo: 1234 12 0 0 0 0 0 0 1234 12 0 0 0 0 0 0
|
|
eth0: 9876543 12345 0 0 0 0 0 100 1234567 9876 0 0 0 0 0 0
|
|
br-lan: 8765432 11234 0 0 0 0 0 50 1123456 8765 0 0 0 0 0 0
|
|
"""
|
|
|
|
HOSTNAME = "OpenWrt"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestGetFacts:
|
|
def _make_send(self):
|
|
"""Return a _send_command mock that handles both str and list args."""
|
|
def _send(cmd, **kw):
|
|
key = cmd[0] if isinstance(cmd, list) else cmd
|
|
if "openwrt_release" in key:
|
|
return OPENWRT_RELEASE
|
|
if "sysinfo/model" in key:
|
|
return SYSINFO_MODEL
|
|
if "uptime" in key:
|
|
return UPTIME
|
|
if "hostname" in key.lower() or "system.@system" in key:
|
|
return HOSTNAME
|
|
if "ip link" in key:
|
|
return IP_LINK_SHOW
|
|
return ""
|
|
return _send
|
|
|
|
def test_returns_required_keys(self, driver):
|
|
driver._send_command = self._make_send()
|
|
|
|
facts = driver.get_facts()
|
|
assert set(facts.keys()) == {
|
|
"vendor", "model", "hostname", "fqdn", "os_version",
|
|
"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"
|
|
|
|
def test_os_version(self, driver):
|
|
driver._send_command = self._make_send()
|
|
assert driver.get_facts()["os_version"] == "23.05.3"
|
|
|
|
def test_uptime_parsed(self, driver):
|
|
driver._send_command = self._make_send()
|
|
facts = driver.get_facts()
|
|
assert facts["uptime"] == pytest.approx(352467.12)
|
|
|
|
|
|
class TestParseOpenwrtRelease:
|
|
def test_parses_release(self):
|
|
result = OpenWrtDriver._parse_openwrt_release(OPENWRT_RELEASE)
|
|
assert result["DISTRIB_RELEASE"] == "23.05.3"
|
|
assert result["DISTRIB_ID"] == "OpenWrt"
|
|
assert result["DISTRIB_TARGET"] == "ath79/generic"
|
|
|
|
|
|
class TestGetInterfaces:
|
|
def test_returns_dict(self, driver):
|
|
driver._send_command = lambda cmd, **kw: IP_LINK_SHOW
|
|
result = driver.get_interfaces()
|
|
assert isinstance(result, dict)
|
|
|
|
def test_eth0_is_up(self, driver):
|
|
driver._send_command = lambda cmd, **kw: IP_LINK_SHOW
|
|
result = driver.get_interfaces()
|
|
assert "eth0" in result
|
|
assert result["eth0"]["is_up"] is True
|
|
assert result["eth0"]["is_enabled"] is True
|
|
|
|
def test_eth1_is_down(self, driver):
|
|
driver._send_command = lambda cmd, **kw: IP_LINK_SHOW
|
|
result = driver.get_interfaces()
|
|
assert "eth1" in result
|
|
assert result["eth1"]["is_up"] is False
|
|
|
|
def test_mac_address_populated(self, driver):
|
|
driver._send_command = lambda cmd, **kw: IP_LINK_SHOW
|
|
result = driver.get_interfaces()
|
|
assert result["eth0"]["mac_address"] != ""
|
|
|
|
def test_required_keys(self, driver):
|
|
driver._send_command = lambda cmd, **kw: IP_LINK_SHOW
|
|
result = driver.get_interfaces()
|
|
for iface_data in result.values():
|
|
assert set(iface_data.keys()) >= {
|
|
"is_up", "is_enabled", "description",
|
|
"last_flapped", "speed", "mtu", "mac_address",
|
|
}
|
|
|
|
|
|
class TestGetInterfacesIP:
|
|
def test_br_lan_ipv4(self, driver):
|
|
driver._send_command = lambda cmd, **kw: IP_ADDR_SHOW
|
|
result = driver.get_interfaces_ip()
|
|
assert "br-lan" in result
|
|
assert "192.168.1.1" in result["br-lan"]["ipv4"]
|
|
assert result["br-lan"]["ipv4"]["192.168.1.1"]["prefix_length"] == 24
|
|
|
|
def test_br_lan_ipv6(self, driver):
|
|
driver._send_command = lambda cmd, **kw: IP_ADDR_SHOW
|
|
result = driver.get_interfaces_ip()
|
|
assert "ipv6" in result["br-lan"]
|
|
assert "fd00::1" in result["br-lan"]["ipv6"]
|
|
|
|
|
|
class TestGetArpTable:
|
|
def test_returns_list(self, driver):
|
|
driver._send_command = lambda cmd, **kw: IP_NEIGH_SHOW
|
|
result = driver.get_arp_table()
|
|
assert isinstance(result, list)
|
|
|
|
def test_entries_count(self, driver):
|
|
driver._send_command = lambda cmd, **kw: IP_NEIGH_SHOW
|
|
result = driver.get_arp_table()
|
|
# FAILED entry should be skipped
|
|
assert len(result) == 2
|
|
|
|
def test_entry_keys(self, driver):
|
|
driver._send_command = lambda cmd, **kw: IP_NEIGH_SHOW
|
|
result = driver.get_arp_table()
|
|
for entry in result:
|
|
assert set(entry.keys()) >= {"interface", "mac", "ip", "age"}
|
|
|
|
def test_ip_value(self, driver):
|
|
driver._send_command = lambda cmd, **kw: IP_NEIGH_SHOW
|
|
result = driver.get_arp_table()
|
|
ips = {e["ip"] for e in result}
|
|
assert "192.168.1.100" in ips
|
|
assert "192.168.1.101" in ips
|
|
|
|
|
|
class TestGetMacAddressTable:
|
|
def test_returns_list(self, driver):
|
|
driver._send_command = lambda cmd, **kw: BRIDGE_FDB
|
|
result = driver.get_mac_address_table()
|
|
assert isinstance(result, list)
|
|
|
|
def test_multicast_skipped(self, driver):
|
|
driver._send_command = lambda cmd, **kw: BRIDGE_FDB
|
|
result = driver.get_mac_address_table()
|
|
macs = [e["mac"] for e in result]
|
|
assert not any("33:33" in m for m in macs)
|
|
|
|
def test_vlan_parsed(self, driver):
|
|
driver._send_command = lambda cmd, **kw: BRIDGE_FDB
|
|
result = driver.get_mac_address_table()
|
|
vlan1_entries = [e for e in result if e["vlan"] == 1]
|
|
assert len(vlan1_entries) >= 1
|
|
|
|
|
|
class TestGetConfig:
|
|
def test_returns_uci_export(self, driver):
|
|
driver._send_command = lambda cmd, **kw: UCI_EXPORT
|
|
result = driver.get_config()
|
|
assert "running" in result
|
|
assert "startup" in result
|
|
assert "candidate" in result
|
|
assert "package" in result["running"]
|
|
|
|
def test_candidate_always_empty(self, driver):
|
|
driver._send_command = lambda cmd, **kw: UCI_EXPORT
|
|
result = driver.get_config()
|
|
assert result["candidate"] == ""
|
|
|
|
|
|
class TestGetLldpNeighbors:
|
|
def test_returns_dict(self, driver):
|
|
driver._send_command = lambda cmd, **kw: LLDPCTL_KV
|
|
result = driver.get_lldp_neighbors()
|
|
assert isinstance(result, dict)
|
|
|
|
def test_eth0_neighbor(self, driver):
|
|
driver._send_command = lambda cmd, **kw: LLDPCTL_KV
|
|
result = driver.get_lldp_neighbors()
|
|
assert "eth0" in result
|
|
assert result["eth0"][0]["hostname"] == "core-router"
|
|
assert result["eth0"][0]["port"] == "ether1"
|
|
|
|
|
|
BRIDGE_VLAN_SHOW = """\
|
|
port vlan-id
|
|
eth0 1 PVID Egress Untagged
|
|
10
|
|
20
|
|
br-lan 1 PVID Egress Untagged
|
|
10
|
|
20
|
|
eth1 20 PVID Egress Untagged
|
|
"""
|
|
|
|
UCI_NETWORK_VLANS = """\
|
|
network.@bridge-vlan[0]=bridge-vlan
|
|
network.@bridge-vlan[0].device='br-lan'
|
|
network.@bridge-vlan[0].vlan='10'
|
|
network.@bridge-vlan[0].name='management'
|
|
network.@bridge-vlan[1]=bridge-vlan
|
|
network.@bridge-vlan[1].device='br-lan'
|
|
network.@bridge-vlan[1].vlan='20'
|
|
network.@bridge-vlan[1].name='iot'
|
|
"""
|
|
|
|
|
|
class TestGetVlans:
|
|
def _send(self, cmd, **kw):
|
|
if "bridge vlan" in cmd:
|
|
return BRIDGE_VLAN_SHOW
|
|
if "uci show network" in cmd:
|
|
return UCI_NETWORK_VLANS
|
|
return ""
|
|
|
|
def test_returns_dict(self, driver):
|
|
driver._send_command = self._send
|
|
result = driver.get_vlans()
|
|
assert isinstance(result, dict)
|
|
|
|
def test_vlan_ids_present(self, driver):
|
|
driver._send_command = self._send
|
|
result = driver.get_vlans()
|
|
assert "1" in result
|
|
assert "10" in result
|
|
assert "20" in result
|
|
|
|
def test_required_keys(self, driver):
|
|
driver._send_command = self._send
|
|
result = driver.get_vlans()
|
|
for vlan_data in result.values():
|
|
assert "name" in vlan_data
|
|
assert "tagged" in vlan_data
|
|
assert "untagged" in vlan_data
|
|
|
|
def test_pvid_port_is_untagged_member(self, driver):
|
|
driver._send_command = self._send
|
|
result = driver.get_vlans()
|
|
assert set(result["1"]["untagged"]) == {"eth0", "br-lan"}
|
|
assert result["1"]["tagged"] == []
|
|
|
|
def test_continuation_lines_are_tagged_members(self, driver):
|
|
driver._send_command = self._send
|
|
result = driver.get_vlans()
|
|
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
|
|
result = driver.get_vlans()
|
|
assert result["10"]["name"] == "management"
|
|
assert result["20"]["name"] == "iot"
|
|
|
|
def test_vlan_without_uci_name_is_empty_string(self, driver):
|
|
driver._send_command = self._send
|
|
result = driver.get_vlans()
|
|
assert result["1"]["name"] == ""
|
|
|
|
def test_no_duplicate_ports(self, driver):
|
|
driver._send_command = self._send
|
|
result = driver.get_vlans()
|
|
for vlan_data in result.values():
|
|
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: ""
|
|
result = driver.get_vlans()
|
|
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'")
|
|
assert driver._candidate_config is not None
|
|
assert driver._candidate_mode == "merge"
|
|
|
|
def test_load_replace_candidate(self, driver):
|
|
driver.load_replace_candidate(config=UCI_EXPORT)
|
|
assert driver._candidate_config is not None
|
|
assert driver._candidate_mode == "replace"
|
|
|
|
def test_discard_config(self, driver):
|
|
driver.load_merge_candidate(config="uci set system.@system[0].hostname='test'")
|
|
driver.discard_config()
|
|
assert driver._candidate_config is None
|
|
assert not driver.has_pending_commit()
|
|
|
|
def test_compare_merge_candidate(self, driver):
|
|
driver._send_command = lambda cmd, **kw: UCI_EXPORT
|
|
driver.load_merge_candidate(config="uci set system.@system[0].hostname='test'")
|
|
diff = driver.compare_config()
|
|
assert diff.startswith("+")
|
|
|
|
def test_compare_no_candidate(self, driver):
|
|
assert driver.compare_config() == ""
|
|
|
|
def test_has_pending_commit_false_initially(self, driver):
|
|
assert not driver.has_pending_commit()
|
|
|
|
def test_has_pending_commit_true_after_load(self, driver):
|
|
driver.load_merge_candidate(config="uci set system.@system[0].hostname='test'")
|
|
assert driver.has_pending_commit()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Sample data for new methods
|
|
# ---------------------------------------------------------------------------
|
|
|
|
PASSWD = """\
|
|
root:$1$xyz:0:0:root:/root:/bin/ash
|
|
daemon:*:1:1:daemon:/var:/bin/false
|
|
nobody:*:65534:65534:nobody:/var:/bin/false
|
|
alice:$6$abc:1001:1001:Alice:/home/alice:/bin/ash
|
|
"""
|
|
|
|
ROOT_AUTHORIZED_KEYS = "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABroot@host"
|
|
ALICE_AUTHORIZED_KEYS = "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5alice@host"
|
|
|
|
UCI_SYSTEM_NTP = """\
|
|
system.@system[0]=system
|
|
system.@system[0].hostname='OpenWrt'
|
|
system.@system[0].timezone='UTC'
|
|
system.ntp=timeserver
|
|
system.ntp.server='0.openwrt.pool.ntp.org 1.openwrt.pool.ntp.org 2.openwrt.pool.ntp.org'
|
|
system.ntp.enabled='1'
|
|
system.ntp.enable_server='0'
|
|
"""
|
|
|
|
NTPQ_OUTPUT = """\
|
|
remote refid st t when poll reach delay offset jitter
|
|
==============================================================================
|
|
*188.114.101.4 188.114.100.1 4 u 107 256 377 164.228 -13.866 2.695
|
|
+37.187.56.220 192.53.103.108 2 u 22 64 377 30.112 5.123 1.100
|
|
"""
|
|
|
|
UCI_SNMPD = """\
|
|
snmpd.@agent[0]=agent
|
|
snmpd.@agent[0].agentaddress='UDP:161'
|
|
snmpd.@com2sec[0]=com2sec
|
|
snmpd.@com2sec[0].secname='public'
|
|
snmpd.@com2sec[0].source='default'
|
|
snmpd.@com2sec[0].community='public'
|
|
snmpd.@com2sec[1]=com2sec
|
|
snmpd.@com2sec[1].secname='private'
|
|
snmpd.@com2sec[1].source='10.0.0.0/8'
|
|
snmpd.@com2sec[1].community='private'
|
|
snmpd.@system[0]=system
|
|
snmpd.@system[0].sysContact='admin@example.com'
|
|
snmpd.@system[0].sysLocation='Server Room'
|
|
snmpd.@system[0].sysName='MyRouter'
|
|
"""
|
|
|
|
PING_SUCCESS = """\
|
|
PING 8.8.8.8 (8.8.8.8): 56 data bytes
|
|
64 bytes from 8.8.8.8: seq=0 ttl=120 time=7.123 ms
|
|
64 bytes from 8.8.8.8: seq=1 ttl=120 time=6.987 ms
|
|
64 bytes from 8.8.8.8: seq=2 ttl=120 time=7.234 ms
|
|
--- 8.8.8.8 ping statistics ---
|
|
3 packets transmitted, 3 packets received, 0% packet loss
|
|
round-trip min/avg/max = 6.987/7.115/7.234 ms
|
|
"""
|
|
|
|
PING_LOSS = """\
|
|
PING 10.0.0.99 (10.0.0.99): 56 data bytes
|
|
--- 10.0.0.99 ping statistics ---
|
|
3 packets transmitted, 0 packets received, 100% packet loss
|
|
"""
|
|
|
|
PING_ERROR = "ping: bad address 'invalid.host'"
|
|
|
|
IPV6_NEIGH = """\
|
|
2001:db8::1 dev eth0 lladdr aa:bb:cc:dd:ee:ff REACHABLE
|
|
fe80::1 dev br-lan lladdr 11:22:33:44:55:66 STALE
|
|
2001:db8::2 dev eth0 FAILED
|
|
"""
|
|
|
|
IP_ROUTE_SHOW = """\
|
|
default via 192.168.1.1 dev br-wan proto dhcp src 203.0.113.1 metric 100
|
|
192.168.1.0/24 dev br-lan proto kernel scope link src 192.168.1.1
|
|
10.0.0.0/8 via 192.168.1.254 dev br-wan proto static metric 50
|
|
"""
|
|
|
|
TRACEROUTE_OUTPUT = """\
|
|
traceroute to 8.8.8.8 (8.8.8.8), 30 hops max, 38 byte packets
|
|
1 192.168.1.1 (192.168.1.1) 1.123 ms 1.456 ms 1.789 ms
|
|
2 10.0.0.1 (10.0.0.1) 5.123 ms 5.456 ms 5.789 ms
|
|
3 * * *
|
|
4 8.8.8.8 (8.8.8.8) 7.001 ms 6.999 ms 7.100 ms
|
|
"""
|
|
|
|
TRACEROUTE_ERROR = "traceroute: unknown host invalid.host"
|
|
|
|
IP_NETNS_LIST = """\
|
|
vpn (id: 1)
|
|
mgmt (id: 2)
|
|
"""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests for new methods
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCli:
|
|
def test_returns_dict_keyed_by_command(self, driver):
|
|
driver._send_command = lambda cmd, **kw: f"output of {cmd}"
|
|
result = driver.cli(["uname -a", "uptime"])
|
|
assert set(result.keys()) == {"uname -a", "uptime"}
|
|
|
|
def test_output_content(self, driver):
|
|
driver._send_command = lambda cmd, **kw: "Linux OpenWrt"
|
|
result = driver.cli(["uname -a"])
|
|
assert result["uname -a"] == "Linux OpenWrt"
|
|
|
|
def test_empty_command_list(self, driver):
|
|
result = driver.cli([])
|
|
assert result == {}
|
|
|
|
|
|
class TestGetUsers:
|
|
def _send(self, cmd, **kw):
|
|
key = cmd[0] if isinstance(cmd, list) else cmd
|
|
if "/etc/passwd" in key:
|
|
return PASSWD
|
|
if "root/.ssh/authorized_keys" in key or "dropbear/authorized_keys" in key:
|
|
return ROOT_AUTHORIZED_KEYS
|
|
if "alice" in key:
|
|
return ALICE_AUTHORIZED_KEYS
|
|
return ""
|
|
|
|
def test_returns_dict(self, driver):
|
|
driver._send_command = self._send
|
|
assert isinstance(driver.get_users(), dict)
|
|
|
|
def test_root_level_15(self, driver):
|
|
driver._send_command = self._send
|
|
users = driver.get_users()
|
|
assert "root" in users
|
|
assert users["root"]["level"] == 15
|
|
|
|
def test_regular_user_level_1(self, driver):
|
|
driver._send_command = self._send
|
|
users = driver.get_users()
|
|
assert "alice" in users
|
|
assert users["alice"]["level"] == 1
|
|
|
|
def test_system_accounts_excluded(self, driver):
|
|
driver._send_command = self._send
|
|
users = driver.get_users()
|
|
assert "daemon" not in users
|
|
assert "nobody" not in users
|
|
|
|
def test_root_has_ssh_key(self, driver):
|
|
driver._send_command = self._send
|
|
users = driver.get_users()
|
|
assert len(users["root"]["sshkeys"]) >= 1
|
|
assert users["root"]["sshkeys"][0].startswith("ssh-rsa")
|
|
|
|
def test_required_keys(self, driver):
|
|
driver._send_command = self._send
|
|
users = driver.get_users()
|
|
for data in users.values():
|
|
assert "level" in data
|
|
assert "password" in data
|
|
assert "sshkeys" in data
|
|
|
|
|
|
class TestGetNtpServers:
|
|
def test_returns_dict(self, driver):
|
|
driver._send_command = lambda cmd, **kw: UCI_SYSTEM_NTP
|
|
result = driver.get_ntp_servers()
|
|
assert isinstance(result, dict)
|
|
|
|
def test_servers_present(self, driver):
|
|
driver._send_command = lambda cmd, **kw: UCI_SYSTEM_NTP
|
|
result = driver.get_ntp_servers()
|
|
assert "0.openwrt.pool.ntp.org" in result
|
|
assert "1.openwrt.pool.ntp.org" in result
|
|
assert "2.openwrt.pool.ntp.org" in result
|
|
|
|
def test_empty_when_no_ntp(self, driver):
|
|
driver._send_command = lambda cmd, **kw: ""
|
|
assert driver.get_ntp_servers() == {}
|
|
|
|
|
|
class TestGetNtpStats:
|
|
def test_returns_list(self, driver):
|
|
driver._send_command = lambda cmd, **kw: NTPQ_OUTPUT
|
|
result = driver.get_ntp_stats()
|
|
assert isinstance(result, list)
|
|
|
|
def test_synchronized_entry(self, driver):
|
|
driver._send_command = lambda cmd, **kw: NTPQ_OUTPUT
|
|
result = driver.get_ntp_stats()
|
|
synced = [e for e in result if e["synchronized"]]
|
|
assert len(synced) == 1
|
|
assert synced[0]["remote"] == "188.114.101.4"
|
|
|
|
def test_required_keys(self, driver):
|
|
driver._send_command = lambda cmd, **kw: NTPQ_OUTPUT
|
|
result = driver.get_ntp_stats()
|
|
for entry in result:
|
|
assert set(entry.keys()) >= {
|
|
"remote", "referenceid", "synchronized", "stratum",
|
|
"type", "when", "hostpoll", "reachability", "delay", "offset", "jitter",
|
|
}
|
|
|
|
def test_empty_on_no_tool(self, driver):
|
|
driver._send_command = lambda cmd, **kw: "sh: ntpq: not found"
|
|
result = driver.get_ntp_stats()
|
|
assert result == []
|
|
|
|
|
|
class TestGetSnmpInformation:
|
|
def test_returns_dict(self, driver):
|
|
driver._send_command = lambda cmd, **kw: UCI_SNMPD
|
|
result = driver.get_snmp_information()
|
|
assert isinstance(result, dict)
|
|
|
|
def test_required_keys(self, driver):
|
|
driver._send_command = lambda cmd, **kw: UCI_SNMPD
|
|
result = driver.get_snmp_information()
|
|
assert set(result.keys()) >= {"chassis_id", "community", "contact", "location"}
|
|
|
|
def test_contact_and_location(self, driver):
|
|
driver._send_command = lambda cmd, **kw: UCI_SNMPD
|
|
result = driver.get_snmp_information()
|
|
assert result["contact"] == "admin@example.com"
|
|
assert result["location"] == "Server Room"
|
|
|
|
def test_community_entries(self, driver):
|
|
driver._send_command = lambda cmd, **kw: UCI_SNMPD
|
|
result = driver.get_snmp_information()
|
|
assert "public" in result["community"]
|
|
assert "private" in result["community"]
|
|
|
|
def test_community_mode(self, driver):
|
|
driver._send_command = lambda cmd, **kw: UCI_SNMPD
|
|
result = driver.get_snmp_information()
|
|
assert result["community"]["public"]["mode"] == "ro"
|
|
assert result["community"]["private"]["mode"] == "rw"
|
|
|
|
|
|
class TestPing:
|
|
def test_success_result(self, driver):
|
|
driver._send_command = lambda cmd, **kw: PING_SUCCESS
|
|
result = driver.ping("8.8.8.8")
|
|
assert "success" in result
|
|
assert result["success"]["probes_sent"] == 3
|
|
assert result["success"]["packet_loss"] == 0
|
|
|
|
def test_rtt_values(self, driver):
|
|
driver._send_command = lambda cmd, **kw: PING_SUCCESS
|
|
result = driver.ping("8.8.8.8")
|
|
s = result["success"]
|
|
assert s["rtt_min"] == 6.987
|
|
assert s["rtt_max"] == 7.234
|
|
assert s["rtt_avg"] == 7.115
|
|
|
|
def test_probe_results(self, driver):
|
|
driver._send_command = lambda cmd, **kw: PING_SUCCESS
|
|
result = driver.ping("8.8.8.8")
|
|
assert len(result["success"]["results"]) == 3
|
|
assert result["success"]["results"][0]["ip_address"] == "8.8.8.8"
|
|
|
|
def test_100_percent_loss(self, driver):
|
|
driver._send_command = lambda cmd, **kw: PING_LOSS
|
|
result = driver.ping("10.0.0.99")
|
|
assert "success" in result
|
|
assert result["success"]["packet_loss"] == 3
|
|
|
|
def test_error_on_bad_host(self, driver):
|
|
driver._send_command = lambda cmd, **kw: PING_ERROR
|
|
result = driver.ping("invalid.host")
|
|
assert "error" in result
|
|
|
|
|
|
class TestGetIpv6NeighborsTable:
|
|
def test_returns_list(self, driver):
|
|
driver._send_command = lambda cmd, **kw: IPV6_NEIGH
|
|
assert isinstance(driver.get_ipv6_neighbors_table(), list)
|
|
|
|
def test_failed_entries_excluded(self, driver):
|
|
driver._send_command = lambda cmd, **kw: IPV6_NEIGH
|
|
result = driver.get_ipv6_neighbors_table()
|
|
assert len(result) == 2
|
|
|
|
def test_required_keys(self, driver):
|
|
driver._send_command = lambda cmd, **kw: IPV6_NEIGH
|
|
for entry in driver.get_ipv6_neighbors_table():
|
|
assert set(entry.keys()) >= {"interface", "mac", "ip", "age", "state"}
|
|
|
|
def test_state_values(self, driver):
|
|
driver._send_command = lambda cmd, **kw: IPV6_NEIGH
|
|
states = {e["state"] for e in driver.get_ipv6_neighbors_table()}
|
|
assert "REACHABLE" in states
|
|
assert "STALE" in states
|
|
|
|
|
|
class TestGetRouteTo:
|
|
def test_returns_dict(self, driver):
|
|
driver._send_command = lambda cmd, **kw: IP_ROUTE_SHOW
|
|
assert isinstance(driver.get_route_to(), dict)
|
|
|
|
def test_default_route_present(self, driver):
|
|
driver._send_command = lambda cmd, **kw: IP_ROUTE_SHOW
|
|
result = driver.get_route_to()
|
|
assert "0.0.0.0/0" in result
|
|
|
|
def test_next_hop(self, driver):
|
|
driver._send_command = lambda cmd, **kw: IP_ROUTE_SHOW
|
|
result = driver.get_route_to()
|
|
default = result["0.0.0.0/0"][0]
|
|
assert default["next_hop"] == "192.168.1.1"
|
|
assert default["outgoing_interface"] == "br-wan"
|
|
|
|
def test_connected_route(self, driver):
|
|
driver._send_command = lambda cmd, **kw: IP_ROUTE_SHOW
|
|
result = driver.get_route_to()
|
|
assert "192.168.1.0/24" in result
|
|
assert result["192.168.1.0/24"][0]["protocol"] == "connected"
|
|
|
|
def test_protocol_filter(self, driver):
|
|
driver._send_command = lambda cmd, **kw: IP_ROUTE_SHOW
|
|
result = driver.get_route_to(protocol="static")
|
|
assert "10.0.0.0/8" in result
|
|
assert "192.168.1.0/24" not in result
|
|
|
|
def test_required_keys(self, driver):
|
|
driver._send_command = lambda cmd, **kw: IP_ROUTE_SHOW
|
|
for prefix, routes in driver.get_route_to().items():
|
|
for route in routes:
|
|
assert set(route.keys()) >= {
|
|
"protocol", "current_active", "next_hop",
|
|
"outgoing_interface", "preference", "routing_table",
|
|
}
|
|
|
|
|
|
class TestTraceroute:
|
|
def test_success_result(self, driver):
|
|
driver._send_command = lambda cmd, **kw: TRACEROUTE_OUTPUT
|
|
result = driver.traceroute("8.8.8.8")
|
|
assert "success" in result
|
|
|
|
def test_hop_count(self, driver):
|
|
driver._send_command = lambda cmd, **kw: TRACEROUTE_OUTPUT
|
|
result = driver.traceroute("8.8.8.8")
|
|
assert len(result["success"]) == 4
|
|
|
|
def test_hop_1_rtt(self, driver):
|
|
driver._send_command = lambda cmd, **kw: TRACEROUTE_OUTPUT
|
|
result = driver.traceroute("8.8.8.8")
|
|
hop1 = result["success"][1]["probes"][1]
|
|
assert hop1["rtt"] == 1.123
|
|
assert hop1["ip_address"] == "192.168.1.1"
|
|
|
|
def test_star_hop(self, driver):
|
|
driver._send_command = lambda cmd, **kw: TRACEROUTE_OUTPUT
|
|
result = driver.traceroute("8.8.8.8")
|
|
hop3 = result["success"][3]["probes"][1]
|
|
assert hop3["ip_address"] == "*"
|
|
|
|
def test_error_on_unknown_host(self, driver):
|
|
driver._send_command = lambda cmd, **kw: TRACEROUTE_ERROR
|
|
result = driver.traceroute("invalid.host")
|
|
assert "error" in result
|
|
|
|
|
|
class TestGetNetworkInstances:
|
|
def test_default_instance_always_present(self, driver):
|
|
driver._send_command = lambda cmd, **kw: (
|
|
IP_LINK_SHOW if "ip link" in cmd else ""
|
|
)
|
|
result = driver.get_network_instances()
|
|
assert "default" in result
|
|
|
|
def test_default_instance_type(self, driver):
|
|
driver._send_command = lambda cmd, **kw: (
|
|
IP_LINK_SHOW if "ip link" in cmd else ""
|
|
)
|
|
result = driver.get_network_instances()
|
|
assert result["default"]["type"] == "DEFAULT_INSTANCE"
|
|
|
|
def test_default_interfaces_populated(self, driver):
|
|
driver._send_command = lambda cmd, **kw: (
|
|
IP_LINK_SHOW if "ip link" in cmd else ""
|
|
)
|
|
result = driver.get_network_instances()
|
|
ifaces = result["default"]["interfaces"]["interface"]
|
|
assert "eth0" in ifaces
|
|
|
|
def test_named_namespaces(self, driver):
|
|
def _send(cmd, **kw):
|
|
if "netns list" in cmd:
|
|
return IP_NETNS_LIST
|
|
if "netns exec" in cmd:
|
|
return "" # empty namespace
|
|
if "ip link" in cmd:
|
|
return IP_LINK_SHOW
|
|
return ""
|
|
driver._send_command = _send
|
|
result = driver.get_network_instances()
|
|
assert "vpn" in result
|
|
assert "mgmt" in result
|
|
assert result["vpn"]["type"] == "L3VRF"
|
|
|
|
def test_name_filter(self, driver):
|
|
driver._send_command = lambda cmd, **kw: (
|
|
IP_LINK_SHOW if "ip link" in cmd else ""
|
|
)
|
|
result = driver.get_network_instances(name="default")
|
|
assert list(result.keys()) == ["default"]
|
|
|
|
def test_required_keys(self, driver):
|
|
driver._send_command = lambda cmd, **kw: (
|
|
IP_LINK_SHOW if "ip link" in cmd else ""
|
|
)
|
|
for inst in driver.get_network_instances().values():
|
|
assert set(inst.keys()) >= {"name", "type", "state", "interfaces"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Sample data for get_channel_scan()
|
|
# ---------------------------------------------------------------------------
|
|
|
|
IW_DEV_OUT = """\
|
|
phy#0
|
|
\tInterface wlan0
|
|
\t\tifindex 4
|
|
\t\twdev 0x1
|
|
\t\taddr aa:bb:cc:dd:ee:f0
|
|
\t\tssid HomeNet
|
|
\t\ttype AP
|
|
\t\tchannel 6 (2437 MHz), width: 20 MHz, center1: 2437 MHz
|
|
phy#1
|
|
\tInterface wlan1
|
|
\t\tifindex 5
|
|
\t\twdev 0x2
|
|
\t\taddr aa:bb:cc:dd:ee:f1
|
|
\t\tssid HomeNet
|
|
\t\ttype AP
|
|
\t\tchannel 36 (5180 MHz), width: 80 MHz, center1: 5210 MHz
|
|
\tInterface wlan2
|
|
\t\tifindex 6
|
|
\t\twdev 0x3
|
|
\t\taddr aa:bb:cc:dd:ee:f2
|
|
\t\tssid GuestNet
|
|
\t\ttype managed
|
|
"""
|
|
|
|
# Minimal iw scan output for wlan0 (2.4 GHz AP, ch 6 + ch 1)
|
|
IW_SCAN_WLAN0 = """\
|
|
BSS aa:bb:cc:dd:01:ff(on wlan0)
|
|
\tfreq: 2437
|
|
\tsignal: -72.00 dBm
|
|
\tSSID: NeighborNet1
|
|
\tHT operation:
|
|
\t\t * primary channel: 6
|
|
\t\t * secondary channel offset: no secondary
|
|
\t\t * STA channel width: 20 MHz
|
|
BSS aa:bb:cc:dd:02:ff(on wlan0)
|
|
\tfreq: 2412
|
|
\tsignal: -85.00 dBm
|
|
\tSSID: FarAway
|
|
\tHT operation:
|
|
\t\t * primary channel: 1
|
|
\t\t * secondary channel offset: no secondary
|
|
\t\t * STA channel width: 20 MHz
|
|
BSS aa:bb:cc:dd:03:ff(on wlan0)
|
|
\tfreq: 2452
|
|
\tsignal: -90.00 dBm
|
|
\tSSID: Hidden
|
|
\tHT operation:
|
|
\t\t * primary channel: 9
|
|
\t\t * secondary channel offset: no secondary
|
|
\t\t * STA channel width: 20 MHz
|
|
"""
|
|
|
|
# Minimal iw scan output for wlan1 (5 GHz AP, ch 36 + 44)
|
|
IW_SCAN_WLAN1 = """\
|
|
BSS 11:22:33:44:01:ff(on wlan1)
|
|
\tfreq: 5180
|
|
\tsignal: -58.00 dBm
|
|
\tSSID: Corp5G
|
|
\tVHT operation:
|
|
\t\t * channel width: 80 MHz
|
|
\t\t * center freq segment 1: 42
|
|
\t\t * center freq segment 2: 0
|
|
BSS 11:22:33:44:02:ff(on wlan1)
|
|
\tfreq: 5220
|
|
\tsignal: -77.00 dBm
|
|
\tSSID: Office5G
|
|
\tVHT operation:
|
|
\t\t * channel width: 80 MHz
|
|
"""
|
|
|
|
|
|
class TestGetChannelScan:
|
|
"""Tests for OpenWrtWirelessMixin.get_channel_scan()."""
|
|
|
|
def _make_send(self, mode: str = "active"):
|
|
"""Return a _send_command mock routing iw dev / iw scan output."""
|
|
def _send(cmd, **kw):
|
|
if cmd.strip().startswith("iw dev") and "scan" not in cmd:
|
|
return IW_DEV_OUT
|
|
if "wlan0" in cmd and "scan" in cmd:
|
|
return IW_SCAN_WLAN0
|
|
if "wlan1" in cmd and "scan" in cmd:
|
|
return IW_SCAN_WLAN1
|
|
return ""
|
|
return _send
|
|
|
|
def test_returns_dict(self, driver):
|
|
driver._send_command = self._make_send()
|
|
result = driver.get_channel_scan()
|
|
assert isinstance(result, dict)
|
|
|
|
def test_only_ap_interfaces_included(self, driver):
|
|
"""wlan2 is managed (client) mode — must not appear in results."""
|
|
driver._send_command = self._make_send()
|
|
result = driver.get_channel_scan()
|
|
assert "wlan0" in result
|
|
assert "wlan1" in result
|
|
assert "wlan2" not in result
|
|
|
|
def test_wlan0_entry_count(self, driver):
|
|
driver._send_command = self._make_send()
|
|
result = driver.get_channel_scan()
|
|
assert len(result["wlan0"]) == 3
|
|
|
|
def test_wlan0_entry_required_keys(self, driver):
|
|
driver._send_command = self._make_send()
|
|
entry = driver.get_channel_scan()["wlan0"][0]
|
|
assert set(entry.keys()) >= {
|
|
"bssid", "ssid", "frequency", "channel", "signal_dbm",
|
|
"channel_width", "band",
|
|
}
|
|
|
|
def test_wlan0_first_entry_values(self, driver):
|
|
driver._send_command = self._make_send()
|
|
entry = driver.get_channel_scan()["wlan0"][0]
|
|
assert entry["bssid"] == "aa:bb:cc:dd:01:ff"
|
|
assert entry["ssid"] == "NeighborNet1"
|
|
assert entry["frequency"] == 2437
|
|
assert entry["channel"] == 6
|
|
assert entry["signal_dbm"] == -72
|
|
assert entry["channel_width"] == 20
|
|
assert entry["band"] == "2.4GHz"
|
|
|
|
def test_wlan0_ch1_entry(self, driver):
|
|
driver._send_command = self._make_send()
|
|
entries = driver.get_channel_scan()["wlan0"]
|
|
ch1 = next(e for e in entries if e["channel"] == 1)
|
|
assert ch1["signal_dbm"] == -85
|
|
assert ch1["band"] == "2.4GHz"
|
|
|
|
def test_wlan0_hidden_ssid(self, driver):
|
|
"""Empty SSID (hidden) should be stored as empty string."""
|
|
driver._send_command = self._make_send()
|
|
entries = driver.get_channel_scan()["wlan0"]
|
|
ch9 = next(e for e in entries if e["channel"] == 9)
|
|
assert ch9["ssid"] == "Hidden"
|
|
|
|
def test_wlan1_band_is_5ghz(self, driver):
|
|
driver._send_command = self._make_send()
|
|
entries = driver.get_channel_scan()["wlan1"]
|
|
assert all(e["band"] == "5GHz" for e in entries)
|
|
|
|
def test_wlan1_80mhz_width(self, driver):
|
|
driver._send_command = self._make_send()
|
|
entry = driver.get_channel_scan()["wlan1"][0]
|
|
assert entry["channel_width"] == 80
|
|
|
|
def test_passive_mode_uses_passive_flag(self, driver):
|
|
"""With mode='passive', the scan command must contain 'passive'."""
|
|
issued_cmds: list[str] = []
|
|
|
|
def _send(cmd, **kw):
|
|
issued_cmds.append(cmd)
|
|
if cmd.strip().startswith("iw dev") and "scan" not in cmd:
|
|
return IW_DEV_OUT
|
|
return ""
|
|
|
|
driver._send_command = _send
|
|
driver.get_channel_scan(mode="passive")
|
|
scan_cmds = [c for c in issued_cmds if "scan" in c]
|
|
assert all("passive" in c for c in scan_cmds)
|
|
|
|
def test_active_mode_no_passive_flag(self, driver):
|
|
"""With mode='active' (default), scan command must NOT contain 'passive'."""
|
|
issued_cmds: list[str] = []
|
|
|
|
def _send(cmd, **kw):
|
|
issued_cmds.append(cmd)
|
|
if cmd.strip().startswith("iw dev") and "scan" not in cmd:
|
|
return IW_DEV_OUT
|
|
return ""
|
|
|
|
driver._send_command = _send
|
|
driver.get_channel_scan(mode="active")
|
|
scan_cmds = [c for c in issued_cmds if "scan" in c]
|
|
assert all("passive" not in c for c in scan_cmds)
|
|
|
|
def test_no_ap_interfaces_returns_empty(self, driver):
|
|
"""If no AP-mode interfaces exist, return empty dict."""
|
|
no_ap = IW_DEV_OUT.replace("type AP", "type managed")
|
|
driver._send_command = lambda cmd, **kw: (
|
|
no_ap if "iw dev" in cmd and "scan" not in cmd else ""
|
|
)
|
|
result = driver.get_channel_scan()
|
|
assert result == {}
|
|
|
|
def test_empty_iw_output_returns_empty(self, driver):
|
|
driver._send_command = lambda cmd, **kw: ""
|
|
result = driver.get_channel_scan()
|
|
assert result == {}
|
|
|
|
def test_6ghz_band_detected(self, driver):
|
|
iw_dev_6g = """\
|
|
phy#0
|
|
\tInterface wlan0
|
|
\t\taddr aa:bb:cc:dd:ee:f0
|
|
\t\tssid Net6G
|
|
\t\ttype AP
|
|
"""
|
|
scan_6g = """\
|
|
BSS aa:bb:cc:dd:01:ff(on wlan0)
|
|
\tfreq: 5975
|
|
\tsignal: -65.00 dBm
|
|
\tSSID: Neighbor6G
|
|
"""
|
|
|
|
def _send(cmd, **kw):
|
|
if "iw dev" in cmd and "scan" not in cmd:
|
|
return iw_dev_6g
|
|
if "scan" in cmd:
|
|
return scan_6g
|
|
return ""
|
|
|
|
driver._send_command = _send
|
|
entries = driver.get_channel_scan().get("wlan0", [])
|
|
assert len(entries) == 1
|
|
assert entries[0]["band"] == "6GHz"
|
|
assert entries[0]["frequency"] == 5975
|
|
|
|
|
|
class TestPushRadioChannel:
|
|
"""Tests for OpenWrtWirelessMixin.push_radio_channel()."""
|
|
|
|
def test_issues_uci_set(self, driver):
|
|
"""Must call 'uci set wireless.radio0.channel=11'."""
|
|
issued: list[str] = []
|
|
driver._send_command = lambda cmd, **kw: issued.append(cmd) or ""
|
|
driver.push_radio_channel("radio0", 11)
|
|
assert any("uci set wireless.radio0.channel=11" in c for c in issued)
|
|
|
|
def test_issues_uci_commit(self, driver):
|
|
issued: list[str] = []
|
|
driver._send_command = lambda cmd, **kw: issued.append(cmd) or ""
|
|
driver.push_radio_channel("radio0", 11)
|
|
assert any("uci commit wireless" in c for c in issued)
|
|
|
|
def test_issues_wifi_restart(self, driver):
|
|
"""Must call 'wifi' (full restart) — NOT just 'wifi reload'."""
|
|
issued: list[str] = []
|
|
driver._send_command = lambda cmd, **kw: issued.append(cmd) or ""
|
|
driver.push_radio_channel("radio1", 36)
|
|
# 'wifi' alone is a full restart; 'wifi reload' is insufficient for channel changes
|
|
assert any(c.strip() == "wifi" for c in issued)
|
|
|
|
def test_auto_channel_zero(self, driver):
|
|
"""channel=0 should set 'auto' in UCI."""
|
|
issued: list[str] = []
|
|
driver._send_command = lambda cmd, **kw: issued.append(cmd) or ""
|
|
driver.push_radio_channel("radio0", 0)
|
|
assert any("channel=auto" in c or "channel='auto'" in c for c in issued)
|
|
|
|
def test_commit_before_restart(self, driver):
|
|
"""UCI commit must happen before wifi restart."""
|
|
issued: list[str] = []
|
|
driver._send_command = lambda cmd, **kw: issued.append(cmd) or ""
|
|
driver.push_radio_channel("radio0", 6)
|
|
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"]
|