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.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, 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: 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[0] == 'uci' and args[1] == '-q': return fake_get(args[3]) if args[:2] == ['uci', 'get']: return fake_get(args[2]) 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' 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, 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'), ({'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', 'uci commit network'): self.assertIn(expected, cmds) 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_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) # Revert must converge runtime like the idle branch: a wifi reload # after the STA disable, with hop resumed only after the reload. cmds = self.cmds reloads = [i for i, c in enumerate(cmds) if c == 'wifi reload'] self.assertEqual(len(reloads), 2, cmds) disable_set = cmds.index('uci set wireless.wlan1up.disabled=1') last_commit = len(cmds) - 1 - cmds[::-1].index('uci commit wireless') resume_idx = self.seq.index('resume') self.assertLess(disable_set, last_commit) self.assertLess(last_commit, reloads[-1]) self.assertLess(reloads[-1], resume_idx) # cmds[i] maps 1:1 onto self.runs[i]; revert reload must be gated. self.assertEqual(self.runs[reloads[-1]], (['wifi', 'reload'], 60)) 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.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(self.mutations(), []) 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') 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()