diff --git a/napalm_openwrt/wireless_mixin.py b/napalm_openwrt/wireless_mixin.py index 18faa62..33a702a 100644 --- a/napalm_openwrt/wireless_mixin.py +++ b/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,34 @@ 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. + + :param ssid_name: SSID name to match against ``option ssid`` on each wifi-iface section. + :param mode: ``"off"`` | ``"whitelist"`` | ``"blacklist"`` — mapped to UCI + ``macfilter`` ``"disable"``/``"allow"``/``"deny"``. + :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. + """ + uci_mode = {"off": "disable", "whitelist": "allow", "blacklist": "deny"}[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 + 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. diff --git a/tests/unit/test_driver.py b/tests/unit/test_driver.py index 899b390..7a36c9d 100644 --- a/tests/unit/test_driver.py +++ b/tests/unit/test_driver.py @@ -1099,3 +1099,138 @@ 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_sets_macfilter_disable_and_clears_maclist(self, driver): + send, issued = self._make_send() + driver._send_command = send + driver.push_mac_acl("CorpWiFi", "off", []) + assert any("wireless.@wifi-iface[0].macfilter='disable'" 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_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)