fix(reliability): resolve STA netdev by phy membership (netifd ignores ifname); sae-mixed for PSK uplinks

This commit is contained in:
2026-08-22 20:37:11 -06:00
parent 132cf4d77a
commit 501fa455ce
2 changed files with 46 additions and 6 deletions
@@ -20,10 +20,39 @@ def current_role():
return 'idle' return 'idle'
def associated(): def _sta_netdev():
"""BSSID of the uplink AP when wlan1up is associated, else None.""" """Actual netdev carrying the radio1 STA. OpenWrt ignores a requested
ifname for mac80211 STA ifaces (comes up as phy1-sta0), so resolve by
phy membership + managed type instead of by name."""
from server import device_run from server import device_run
rc, out, err = device_run(['iw', 'dev', IFACE, 'link'], timeout=10) rc, out, err = device_run(['iw', 'dev'], timeout=10)
if rc != 0:
return None
current = None
managed = []
for line in (out or '').splitlines():
line = line.strip()
if line.startswith('Interface '):
current = line.split()[1]
elif line.startswith('type managed') and current:
if not current.startswith('wlan0'):
managed.append(current)
current = None
for name in managed:
rc2, o2, _ = device_run(
['readlink', '/sys/class/net/%s/phy80211' % name], timeout=10)
if rc2 == 0 and 'phy1' in (o2 or ''):
return name
return None
def associated():
"""BSSID of the uplink AP when the radio1 STA is associated, else None."""
from server import device_run
dev = _sta_netdev()
if not dev:
return None
rc, out, err = device_run(['iw', 'dev', dev, 'link'], timeout=10)
if rc != 0 or 'Connected' not in (out or ''): if rc != 0 or 'Connected' not in (out or ''):
return None return None
for line in (out or '').splitlines(): for line in (out or '').splitlines():
@@ -71,8 +100,10 @@ def set_role(role, ssid=None, psk=None):
['uci', 'set', 'wireless.wlan1up.mode=sta'], ['uci', 'set', 'wireless.wlan1up.mode=sta'],
['uci', 'set', 'wireless.wlan1up.network=cli'], ['uci', 'set', 'wireless.wlan1up.network=cli'],
['uci', 'set', 'wireless.wlan1up.ssid=%s' % ssid], ['uci', 'set', 'wireless.wlan1up.ssid=%s' % ssid],
# sae-mixed = WPA2-PSK/WPA3-SAE transition; associates with
# either security mode. 'none' for open networks.
['uci', 'set', 'wireless.wlan1up.encryption=%s' ['uci', 'set', 'wireless.wlan1up.encryption=%s'
% ('psk2' if psk else 'none')], % ('sae-mixed' if psk else 'none')],
['uci', 'set', 'wireless.wlan1up.disabled=0'], ['uci', 'set', 'wireless.wlan1up.disabled=0'],
] ]
if psk: if psk:
+11 -2
View File
@@ -38,7 +38,16 @@ class RfPlanTest(unittest.TestCase):
args = list(args) args = list(args)
self.seq.append('cmd:' + ' '.join(str(a) for a in args)) self.seq.append('cmd:' + ' '.join(str(a) for a in args))
self.runs.append((args, timeout)) self.runs.append((args, timeout))
if args == ['iw', 'dev']:
# interface enumeration: radio1 STA comes up as phy1-sta0
return 0, ('Interface wlan1mon\n'
'\ttype monitor\n'
'Interface phy1-sta0\n'
'\ttype managed\n'), ''
if args[0] == 'readlink' and 'phy80211' in args[1]:
return 0, '../../devices/platform/usb/phy1\n', ''
if args[0] == 'iw': if args[0] == 'iw':
# iw dev <iface> link
if self.iw_fail_left > 0: if self.iw_fail_left > 0:
self.iw_fail_left -= 1 self.iw_fail_left -= 1
return 0, 'Not connected.\n', '' return 0, 'Not connected.\n', ''
@@ -108,7 +117,7 @@ class RfPlanTest(unittest.TestCase):
'uci set wireless.wlan1up.mode=sta', 'uci set wireless.wlan1up.mode=sta',
'uci set wireless.wlan1up.network=cli', 'uci set wireless.wlan1up.network=cli',
'uci set wireless.wlan1up.ssid=Net', 'uci set wireless.wlan1up.ssid=Net',
'uci set wireless.wlan1up.encryption=psk2', 'uci set wireless.wlan1up.encryption=sae-mixed',
'uci set wireless.wlan1up.key=key', 'uci set wireless.wlan1up.key=key',
'uci set wireless.wlan1up.disabled=0', 'uci set wireless.wlan1up.disabled=0',
'uci set network.cli=interface', 'uci set network.cli=interface',
@@ -146,7 +155,7 @@ class RfPlanTest(unittest.TestCase):
self.assertEqual(result['assoc'], 'aa:bb:cc:dd:ee:ff') self.assertEqual(result['assoc'], 'aa:bb:cc:dd:ee:ff')
self.assertEqual(len(self.sleeps), 3) self.assertEqual(len(self.sleeps), 3)
iw_calls = [c for c in self.cmds iw_calls = [c for c in self.cmds
if c.startswith('iw dev wlan1up link')] if c.startswith('iw dev phy1-sta0 link')]
self.assertEqual(len(iw_calls), 3) self.assertEqual(len(iw_calls), 3)
def test_assoc_poll_exhaustion_reverts_uplink(self): def test_assoc_poll_exhaustion_reverts_uplink(self):