292 lines
11 KiB
Python
292 lines
11 KiB
Python
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()
|