From bace45d6e41645dd514bd63bf3c788def1e417d2 Mon Sep 17 00:00:00 2001 From: c4ch3c4d3 Date: Sat, 22 Aug 2026 14:35:12 -0600 Subject: [PATCH] =?UTF-8?q?fix(reliability):=20rfplan=20review=20fixes=20?= =?UTF-8?q?=E2=80=94=20cli=20commit,=20assoc=20poll,=20ensure=5Fattack=20w?= =?UTF-8?q?iring,=20idle=20reload?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../remote_access/pager-webui/mk8_rfplan.py | 64 ++++++-- .../user/remote_access/pager-webui/server.py | 13 ++ tests/test_mk8_rfplan.py | 155 ++++++++++++++++-- 3 files changed, 206 insertions(+), 26 deletions(-) diff --git a/payload/user/remote_access/pager-webui/mk8_rfplan.py b/payload/user/remote_access/pager-webui/mk8_rfplan.py index ccb2b3c..96d4018 100644 --- a/payload/user/remote_access/pager-webui/mk8_rfplan.py +++ b/payload/user/remote_access/pager-webui/mk8_rfplan.py @@ -1,9 +1,16 @@ """Mark VIII RF role manager: radio1/phy1 is shared between an uplink STA (``wlan1up``) and attack work, so the roles are made mutually exclusive. Uplink pauses channel hopping; attack/idle resumes it.""" +import time + ROLE_KEY = 'mk8.rfplan.role' IFACE = 'wlan1up' +# wifi reload returns while wpa_supplicant is still scanning/authenticating; +# poll instead of checking once or every real uplink would false-fail. +ASSOC_ATTEMPTS = 5 +ASSOC_WAIT_SECONDS = 2 + def current_role(): from server import _uci_values @@ -22,7 +29,9 @@ def associated(): for line in (out or '').splitlines(): line = line.strip() if line.startswith('Connected to '): - return line.split()[2] + parts = line.split() + if len(parts) >= 3: + return parts[2] return None @@ -31,6 +40,24 @@ def hop_paused(): return _read_hop() == '0' +def _ensure_cli_network(): + """Make network 'cli' usable for the STA. Returns True when a network + change was staged and still needs ``uci commit network``. Stock firmware + ships 'cli' present but disabled; create a minimal DHCP interface when it + is missing entirely so netifd can bring wlan1up up either way.""" + from server import device_run + rc, _, _ = device_run(['uci', '-q', 'get', 'network.cli']) + if rc != 0: + device_run(['uci', 'set', 'network.cli=interface']) + device_run(['uci', 'set', 'network.cli.proto=dhcp']) + return True + rc, out, _ = device_run(['uci', '-q', 'get', 'network.cli.disabled']) + if rc == 0 and out.strip() == '1': + device_run(['uci', 'set', 'network.cli.disabled=0']) + return True + return False + + def set_role(role, ssid=None, psk=None): from server import device_run, _pause_hop, _resume_hop if role not in ('uplink', 'attack', 'idle'): @@ -52,17 +79,19 @@ def set_role(role, ssid=None, psk=None): cmds.append(['uci', 'set', 'wireless.wlan1up.key=%s' % psk]) for c in cmds: device_run(c) - # The STA rides network 'cli'; stock firmware ships it disabled but - # present. If it is missing entirely, create a minimal DHCP interface - # so netifd can bring wlan1up up. - rc, _, _ = device_run(['uci', 'show', 'network.cli']) - if rc != 0: - device_run(['uci', 'set', 'network.cli=interface']) - device_run(['uci', 'set', 'network.cli.proto=dhcp']) + if _ensure_cli_network(): + # netifd consumes committed config only; staging without commit + # would leave the STA with no L3 attachment. + device_run(['uci', 'commit', 'network']) device_run(['uci', 'commit', 'wireless']) _pause_hop() device_run(['wifi', 'reload'], timeout=60) - assoc = associated() + assoc = None + for _ in range(ASSOC_ATTEMPTS): + time.sleep(ASSOC_WAIT_SECONDS) + assoc = associated() + if assoc: + break if not assoc: disable_uplink() _resume_hop() @@ -71,6 +100,20 @@ def set_role(role, ssid=None, psk=None): # attack/idle: tear down the STA so radio1 is free again. disable_uplink() _resume_hop() + if role == 'attack': + # The deploy path performs its own wifi reload right after; teardown + # converges there without a second reload churn on this phy. + try: + import mk8_events + mk8_events.log_event( + 'rfplan', msg='rfplan role attack applied; STA teardown ' + 'applies at next wifi reload') + except Exception: + pass + else: + # idle has no guaranteed follow-up reload anywhere else, so converge + # now while the gated watchdog is still armed. + device_run(['wifi', 'reload'], timeout=60) return {'ok': True, 'role': role} @@ -81,7 +124,8 @@ def disable_uplink(): def ensure_attack(): - """Exclusivity hook for attack enable paths: switch uplink off first.""" + """Exclusivity hook for radio1 attack-AP enable paths: switch the + uplink off first so one phy never carries STA + AP at once.""" if current_role() == 'uplink': return set_role('attack') return None diff --git a/payload/user/remote_access/pager-webui/server.py b/payload/user/remote_access/pager-webui/server.py index acee8c2..bd5736b 100644 --- a/payload/user/remote_access/pager-webui/server.py +++ b/payload/user/remote_access/pager-webui/server.py @@ -3535,6 +3535,13 @@ def h_pineap_wifi_set_ap(ctx): open_band = channel_band(openap.get('channel')) if openap.get('channel') is not None else None use_radio1 = wpa_band in (BAND_5G, BAND_6G) or open_band in (BAND_5G, BAND_6G) if use_radio1: + # Exclusivity: an uplink STA on radio1 must go down before any + # radio1 AP change so one phy never carries STA + AP at once. + import mk8_rfplan + try: + mk8_rfplan.ensure_attack() + except Exception: + pass wpa_active = (wpa_band in (BAND_5G, BAND_6G) and bool(wpa.get('enabled', True)) and wpa.get('channel') is not None) open_active = (open_band in (BAND_5G, BAND_6G) and bool(openap.get('enabled', True)) @@ -4353,7 +4360,13 @@ def _enterprise_boot_recover(): def h_attacks_deploy(ctx): import mk8_gate + import mk8_rfplan mk8_gate.enter('attack_deploy') + # Exclusivity: tear an active uplink down before deploying attack APs. + try: + mk8_rfplan.ensure_attack() + except Exception: + pass body = ctx.body or {} kind = (body.get('kind') or '').strip().lower() if kind not in ('wpa', 'open', 'enterprise'): diff --git a/tests/test_mk8_rfplan.py b/tests/test_mk8_rfplan.py index c68f000..86f2c2d 100644 --- a/tests/test_mk8_rfplan.py +++ b/tests/test_mk8_rfplan.py @@ -21,14 +21,29 @@ class RfPlanTest(unittest.TestCase): self.seq = [] self.paused = 0 self.resumed = 0 + self.sleeps = [] self.uci_show = {} + self.gets = {} + self.iw_fail_left = 0 self._old = (server.device_run, server._pause_hop, - server._resume_hop, server._read_hop) + server._resume_hop, server._read_hop, + mk8_rfplan.time.sleep) + + def fake_get(key): + if key == 'pineapd.wlan1mon.hop': + return 0, '0\n', '' + return self.gets.get(key, (1, '', 'entry not found')) def fake_run(args, timeout=20, input_data=None): args = list(args) self.seq.append('cmd:' + ' '.join(str(a) for a in args)) self.runs.append((args, timeout)) + if args[0] == 'iw': + if self.iw_fail_left > 0: + self.iw_fail_left -= 1 + return 0, 'Not connected.\n', '' + return 0, ('Connected to aa:bb:cc:dd:ee:ff (on wlan1up)\n' + '\tSSID: Net\n'), '' if args[:2] == ['uci', 'show']: cfg = self.uci_show.get(args[2]) if cfg is None: @@ -36,13 +51,10 @@ class RfPlanTest(unittest.TestCase): out = ''.join("%s.%s='%s'\n" % (args[2], k, v) for k, v in sorted(cfg.items())) return 0, out, '' + if args[0] == 'uci' and args[1] == '-q': + return fake_get(args[3]) if args[:2] == ['uci', 'get']: - if args[2] == 'pineapd.wlan1mon.hop': - return 0, '0\n', '' - return 1, '', 'entry not found' - if args[0] == 'iw': - return 0, ('Connected to aa:bb:cc:dd:ee:ff (on wlan1up)\n' - '\tSSID: Net\n'), '' + return fake_get(args[2]) return 0, '', '' def fake_pause(): @@ -57,15 +69,21 @@ class RfPlanTest(unittest.TestCase): server._pause_hop = fake_pause server._resume_hop = fake_resume server._read_hop = lambda: '0' + mk8_rfplan.time.sleep = lambda s: self.sleeps.append(s) def tearDown(self): (server.device_run, server._pause_hop, - server._resume_hop, server._read_hop) = self._old + server._resume_hop, server._read_hop, + mk8_rfplan.time.sleep) = self._old @property def cmds(self): return [s[len('cmd:'):] for s in self.seq if s.startswith('cmd:')] + def mutations(self): + return [c for c in self.cmds + if c.startswith(('uci set', 'uci commit'))] + def test_current_role_reads_uci(self): cases = [ ({'mode': 'sta', 'disabled': '0'}, 'uplink'), @@ -94,33 +112,71 @@ class RfPlanTest(unittest.TestCase): 'uci set wireless.wlan1up.key=key', 'uci set wireless.wlan1up.disabled=0', 'uci set network.cli=interface', - 'uci set network.cli.proto=dhcp'): + 'uci set network.cli.proto=dhcp', + 'uci commit network'): self.assertIn(expected, cmds) - commit = cmds.index('uci commit wireless') + commit_net = cmds.index('uci commit network') + commit_wireless = cmds.index('uci commit wireless') for c in ('uci set wireless.wlan1up.mode=sta', 'uci set wireless.wlan1up.disabled=0', 'uci set network.cli=interface', 'uci set network.cli.proto=dhcp'): - self.assertLess(cmds.index(c), commit) - reload_at = cmds.index('wifi reload') - self.assertLess(commit, reload_at) + self.assertLess(cmds.index(c), commit_net) + self.assertLess(commit_net, cmds.index('uci commit wireless')) + self.assertLess(commit_wireless, cmds.index('wifi reload')) self.assertEqual(self.paused, 1) self.assertLess(self.seq.index('pause'), self.seq.index('cmd:wifi reload')) self.assertEqual(result['assoc'], 'aa:bb:cc:dd:ee:ff') + def test_uplink_enables_present_but_disabled_cli_network(self): + self.gets['network.cli'] = (0, 'interface\n', '') + self.gets['network.cli.disabled'] = (0, '1\n', '') + result = mk8_rfplan.set_role('uplink', ssid='Net') + self.assertTrue(result.get('ok'), result) + cmds = self.cmds + self.assertIn('uci set network.cli.disabled=0', cmds) + self.assertIn('uci commit network', cmds) + self.assertNotIn('uci set network.cli=interface', cmds) + + def test_assoc_poll_succeeds_on_third_attempt(self): + self.iw_fail_left = 2 + result = mk8_rfplan.set_role('uplink', ssid='Net') + self.assertTrue(result.get('ok'), result) + self.assertEqual(result['assoc'], 'aa:bb:cc:dd:ee:ff') + self.assertEqual(len(self.sleeps), 3) + iw_calls = [c for c in self.cmds + if c.startswith('iw dev wlan1up link')] + self.assertEqual(len(iw_calls), 3) + + def test_assoc_poll_exhaustion_reverts_uplink(self): + self.iw_fail_left = 99 + result = mk8_rfplan.set_role('uplink', ssid='Net') + self.assertFalse(result['ok']) + self.assertIn('association failed', result['error']) + self.assertEqual(len(self.sleeps), 5) + self.assertIn('uci set wireless.wlan1up.disabled=1', self.cmds) + self.assertEqual(self.resumed, 1) + def test_set_role_uplink_requires_ssid(self): result = mk8_rfplan.set_role('uplink') self.assertFalse(result['ok']) self.assertIn('ssid', result['error']) - self.assertEqual(self.runs, []) + self.assertEqual(self.mutations(), []) self.assertEqual(self.paused, 0) + def test_idle_reloads_so_sta_disassociates_now(self): + self.uci_show['wireless.wlan1up'] = {'mode': 'sta', 'disabled': '0'} + result = mk8_rfplan.set_role('idle') + self.assertTrue(result.get('ok'), result) + self.assertIn('uci set wireless.wlan1up.disabled=1', self.cmds) + self.assertIn('wifi reload', self.cmds) + self.assertEqual(self.resumed, 1) + def test_exclusivity_switch(self): self.uci_show['wireless.wlan1up'] = {'mode': 'ap'} mk8_rfplan.ensure_attack() - self.assertEqual([c for c in self.cmds - if c.startswith(('uci set', 'uci commit'))], []) + self.assertEqual(self.mutations(), []) self.assertEqual(self.resumed, 0) self.assertEqual(self.paused, 0) self.uci_show['wireless.wlan1up'] = {'mode': 'sta', 'disabled': '0'} @@ -150,6 +206,73 @@ class RfPlanTest(unittest.TestCase): self.assertEqual(status, 200) self.assertEqual(data['role'], 'idle') + def test_ensure_attack_wired_into_attacks_deploy(self): + calls = [] + + def fake_set_role(role, ssid=None, psk=None): + calls.append(role) + return {'ok': True, 'role': role} + + self._patch_rfplan(lambda: 'uplink', fake_set_role) + old_dep = server._deploy_wpa_open + old_state = server.update_pineap_state + server._deploy_wpa_open = lambda kind, body: {} + server.update_pineap_state = lambda *a, **kw: {} + try: + status, payload = server.h_attacks_deploy( + CtxStub({'kind': 'wpa'})) + finally: + self._unpatch_rfplan() + server._deploy_wpa_open = old_dep + server.update_pineap_state = old_state + self.assertEqual(status, 200) + self.assertEqual(calls, ['attack']) + + def test_attacks_deploy_skips_switch_when_not_uplink(self): + calls = [] + + def fake_set_role(role, ssid=None, psk=None): + calls.append(role) + return {'ok': True} + + self._patch_rfplan(lambda: 'idle', fake_set_role) + old_dep = server._deploy_wpa_open + old_state = server.update_pineap_state + server._deploy_wpa_open = lambda kind, body: {} + server.update_pineap_state = lambda *a, **kw: {} + try: + status, _ = server.h_attacks_deploy(CtxStub({'kind': 'wpa'})) + finally: + self._unpatch_rfplan() + server._deploy_wpa_open = old_dep + server.update_pineap_state = old_state + self.assertEqual(status, 200) + self.assertEqual(calls, []) + + def test_ensure_attack_wired_into_radio1_ap_request(self): + calls = [] + + def fake_set_role(role, ssid=None, psk=None): + calls.append(role) + return {'ok': True, 'role': role} + + self._patch_rfplan(lambda: 'uplink', fake_set_role) + try: + status, payload = server.h_pineap_wifi_set_ap( + CtxStub({'open': {'enabled': False, 'channel': 36}})) + finally: + self._unpatch_rfplan() + self.assertEqual(status, 200) + self.assertEqual(calls, ['attack']) + + def _patch_rfplan(self, current_role, set_role): + self._rf_old = (mk8_rfplan.current_role, mk8_rfplan.set_role) + mk8_rfplan.current_role = current_role + mk8_rfplan.set_role = set_role + + def _unpatch_rfplan(self): + mk8_rfplan.current_role, mk8_rfplan.set_role = self._rf_old + if __name__ == '__main__': unittest.main()