feat: get_ssids() macfilter/maclist parsing + push_mac_acl()
Adds MAC ACL (whitelist/blacklist) read+write support for wireless SSIDs, mirroring push_radio_channel's UCI write style. Backs the new Global MAC ACL feature in netOrk.
This commit is contained in:
Christian Manivong
@@ -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.
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user