From 15cd3c5eb8b6f879a6ca3d8a38d00960cc70142c Mon Sep 17 00:00:00 2001 From: c4ch3c4d3 Date: Sat, 22 Aug 2026 14:20:05 -0600 Subject: [PATCH] feat(reliability): phy1 RF role manager with uplink-on-radio1 --- .../remote_access/pager-webui/mk8_rfplan.py | 87 ++++++++++ .../user/remote_access/pager-webui/server.py | 40 +++++ tests/test_mk8_rfplan.py | 155 ++++++++++++++++++ 3 files changed, 282 insertions(+) create mode 100644 payload/user/remote_access/pager-webui/mk8_rfplan.py create mode 100644 tests/test_mk8_rfplan.py diff --git a/payload/user/remote_access/pager-webui/mk8_rfplan.py b/payload/user/remote_access/pager-webui/mk8_rfplan.py new file mode 100644 index 0000000..ccb2b3c --- /dev/null +++ b/payload/user/remote_access/pager-webui/mk8_rfplan.py @@ -0,0 +1,87 @@ +"""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.""" +ROLE_KEY = 'mk8.rfplan.role' +IFACE = 'wlan1up' + + +def current_role(): + from server import _uci_values + cfg = _uci_values('wireless.%s' % IFACE) or {} + if cfg.get('disabled') != '1' and cfg.get('mode') == 'sta': + return 'uplink' + return 'idle' + + +def associated(): + """BSSID of the uplink AP when wlan1up is associated, else None.""" + from server import device_run + rc, out, err = device_run(['iw', 'dev', IFACE, 'link'], timeout=10) + if rc != 0 or 'Connected' not in (out or ''): + return None + for line in (out or '').splitlines(): + line = line.strip() + if line.startswith('Connected to '): + return line.split()[2] + return None + + +def hop_paused(): + from server import _read_hop + return _read_hop() == '0' + + +def set_role(role, ssid=None, psk=None): + from server import device_run, _pause_hop, _resume_hop + if role not in ('uplink', 'attack', 'idle'): + return {'ok': False, 'error': 'role must be uplink, attack or idle'} + if role == 'uplink': + if not ssid: + return {'ok': False, 'error': 'ssid required'} + cmds = [ + ['uci', 'set', 'wireless.wlan1up=wifi-iface'], + ['uci', 'set', 'wireless.wlan1up.device=radio1'], + ['uci', 'set', 'wireless.wlan1up.mode=sta'], + ['uci', 'set', 'wireless.wlan1up.network=cli'], + ['uci', 'set', 'wireless.wlan1up.ssid=%s' % ssid], + ['uci', 'set', 'wireless.wlan1up.encryption=%s' + % ('psk2' if psk else 'none')], + ['uci', 'set', 'wireless.wlan1up.disabled=0'], + ] + if psk: + 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']) + device_run(['uci', 'commit', 'wireless']) + _pause_hop() + device_run(['wifi', 'reload'], timeout=60) + assoc = associated() + if not assoc: + disable_uplink() + _resume_hop() + return {'ok': False, 'error': 'association failed; reverted'} + return {'ok': True, 'role': 'uplink', 'assoc': assoc} + # attack/idle: tear down the STA so radio1 is free again. + disable_uplink() + _resume_hop() + return {'ok': True, 'role': role} + + +def disable_uplink(): + from server import device_run + device_run(['uci', 'set', 'wireless.wlan1up.disabled=1']) + device_run(['uci', 'commit', 'wireless']) + + +def ensure_attack(): + """Exclusivity hook for attack enable paths: switch uplink off first.""" + 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 2525811..acee8c2 100644 --- a/payload/user/remote_access/pager-webui/server.py +++ b/payload/user/remote_access/pager-webui/server.py @@ -4812,6 +4812,44 @@ def start_health_monitor(): threading.Thread(target=_health_loop, daemon=True).start() +# -------------------------------------------------------------------------- +# RF role manager: radio1 shared between uplink STA and attack work. +# -------------------------------------------------------------------------- + +def h_rfplan_get(ctx): + import mk8_rfplan + return 200, {'role': mk8_rfplan.current_role(), + 'assoc': mk8_rfplan.associated(), + 'hop_paused': mk8_rfplan.hop_paused()} + + +def h_rfplan_post(ctx): + import mk8_gate + import mk8_events + import mk8_rfplan + body = ctx.body or {} + role = (body.get('role') or '').strip().lower() + if role not in ('uplink', 'attack', 'idle'): + return 400, {'error': 'role must be uplink, attack or idle'} + mk8_gate.enter('rfplan_' + role) + try: + result = mk8_rfplan.set_role( + role, + ssid=(body.get('ssid') or '').strip() or None, + psk=(body.get('psk') or '').strip() or None) + except Exception as exc: + result = {'ok': False, 'error': str(exc)} + ok = bool(result.get('ok')) + try: + mk8_events.log_event('rfplan', sev='info' if ok else 'warn', + msg='rfplan role %s %s' + % (role, 'applied' if ok else 'failed'), + meta=result) + except Exception: + pass + return (200, result) if ok else (502, result) + + # -------------------------------------------------------------------------- # Startup environment check: run once at service startup (and via # ``server.py --env-check`` on the payload screen) to make the device match @@ -7194,6 +7232,8 @@ def h_mode_release(ctx): ROUTER.add('GET', r'/api/mode', h_mode_get) ROUTER.add('POST', r'/api/mode/release', h_mode_release) +ROUTER.add('GET', r'/api/rfplan', h_rfplan_get) +ROUTER.add('POST', r'/api/rfplan/role', h_rfplan_post) def serve(): diff --git a/tests/test_mk8_rfplan.py b/tests/test_mk8_rfplan.py new file mode 100644 index 0000000..c68f000 --- /dev/null +++ b/tests/test_mk8_rfplan.py @@ -0,0 +1,155 @@ +import os, sys, unittest +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'payload', + 'user', 'remote_access', 'pager-webui')) +import importlib +import server +import mk8_rfplan + + +def setUpModule(): + importlib.reload(server) + + +class CtxStub(object): + def __init__(self, body=None): + self.body = body or {} + + +class RfPlanTest(unittest.TestCase): + def setUp(self): + self.runs = [] + self.seq = [] + self.paused = 0 + self.resumed = 0 + self.uci_show = {} + self._old = (server.device_run, server._pause_hop, + server._resume_hop, server._read_hop) + + 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[:2] == ['uci', 'show']: + cfg = self.uci_show.get(args[2]) + if cfg is None: + return 1, '', 'entry not found' + out = ''.join("%s.%s='%s'\n" % (args[2], k, v) + for k, v in sorted(cfg.items())) + return 0, out, '' + 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 0, '', '' + + def fake_pause(): + self.paused += 1 + self.seq.append('pause') + + def fake_resume(): + self.resumed += 1 + self.seq.append('resume') + + server.device_run = fake_run + server._pause_hop = fake_pause + server._resume_hop = fake_resume + server._read_hop = lambda: '0' + + def tearDown(self): + (server.device_run, server._pause_hop, + server._resume_hop, server._read_hop) = self._old + + @property + def cmds(self): + return [s[len('cmd:'):] for s in self.seq if s.startswith('cmd:')] + + def test_current_role_reads_uci(self): + cases = [ + ({'mode': 'sta', 'disabled': '0'}, 'uplink'), + ({'mode': 'sta', 'disabled': '1'}, 'idle'), + ({'mode': 'ap'}, 'idle'), + ({}, 'idle'), + ] + for cfg, want in cases: + self.uci_show.clear() + self.uci_show['wireless.wlan1up'] = cfg + self.assertEqual(mk8_rfplan.current_role(), want, cfg) + self.uci_show.clear() + self.assertEqual(mk8_rfplan.current_role(), 'idle') + + def test_uplink_sets_sta_section_and_pauses_hop(self): + result = mk8_rfplan.set_role('uplink', ssid='Net', psk='key') + self.assertTrue(result.get('ok'), result) + cmds = self.cmds + for expected in ( + 'uci set wireless.wlan1up=wifi-iface', + 'uci set wireless.wlan1up.device=radio1', + 'uci set wireless.wlan1up.mode=sta', + 'uci set wireless.wlan1up.network=cli', + 'uci set wireless.wlan1up.ssid=Net', + 'uci set wireless.wlan1up.encryption=psk2', + 'uci set wireless.wlan1up.key=key', + 'uci set wireless.wlan1up.disabled=0', + 'uci set network.cli=interface', + 'uci set network.cli.proto=dhcp'): + self.assertIn(expected, cmds) + commit = 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.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_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.paused, 0) + + 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.resumed, 0) + self.assertEqual(self.paused, 0) + self.uci_show['wireless.wlan1up'] = {'mode': 'sta', 'disabled': '0'} + mk8_rfplan.ensure_attack() + cmds = self.cmds + self.assertIn('uci set wireless.wlan1up.disabled=1', cmds) + self.assertIn('uci commit wireless', cmds) + self.assertEqual(self.resumed, 1) + + def test_hop_paused_helper(self): + self.assertTrue(mk8_rfplan.hop_paused()) + server._read_hop = lambda: None + self.assertFalse(mk8_rfplan.hop_paused()) + server._read_hop = lambda: '1' + self.assertFalse(mk8_rfplan.hop_paused()) + + def test_h_rfplan_get_and_post(self): + self.uci_show['wireless.wlan1up'] = {'mode': 'sta', 'disabled': '0'} + status, data = server.h_rfplan_get(CtxStub()) + self.assertEqual(status, 200) + self.assertEqual(data['role'], 'uplink') + self.assertEqual(data['assoc'], 'aa:bb:cc:dd:ee:ff') + self.assertTrue(data['hop_paused']) + status, data = server.h_rfplan_post(CtxStub({'role': 'nope'})) + self.assertEqual(status, 400) + status, data = server.h_rfplan_post(CtxStub({'role': 'idle'})) + self.assertEqual(status, 200) + self.assertEqual(data['role'], 'idle') + + +if __name__ == '__main__': + unittest.main()