145 lines
5.6 KiB
Python
145 lines
5.6 KiB
Python
import os, sys, unittest
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'payload', 'user', 'remote_access', 'pager-webui'))
|
|
import server
|
|
import mk8_guard
|
|
|
|
HOP_KEY = 'pineapd.wlan1mon.hop'
|
|
|
|
class GuardTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.calls = []
|
|
self.uci = {}
|
|
self.pool = 0
|
|
self.mon_up = {'wlan0mon': True, 'wlan1mon': True}
|
|
self.old_iface_up = server._iface_up
|
|
self.old_server_device_run = server.device_run
|
|
self.old_guard_device_run = mk8_guard.device_run
|
|
server._iface_up = lambda name: self.mon_up.get(name, True)
|
|
|
|
def fake_run(args, timeout=20, input_data=None):
|
|
a = list(args)
|
|
self.calls.append(a)
|
|
if a[:2] == ['uci', 'get']:
|
|
key = a[2]
|
|
if key == 'pineapd.@ssidpool[0].ssid':
|
|
return (0, ''.join('s%d\n' % i for i in range(self.pool)), '')
|
|
return (0, self.uci.get(key, '') + '\n', '')
|
|
if a[:2] == ['uci', 'set']:
|
|
k, _, v = a[2].partition('=')
|
|
self.uci[k] = v
|
|
if a[:2] == ['uci', 'delete']:
|
|
self.pool = 0
|
|
return (0, '', '')
|
|
mk8_guard.device_run = fake_run
|
|
server.device_run = fake_run
|
|
mk8_guard._GR_CACHE['data'] = None
|
|
|
|
def tearDown(self):
|
|
server._iface_up = self.old_iface_up
|
|
server.device_run = self.old_server_device_run
|
|
mk8_guard.device_run = self.old_guard_device_run
|
|
mk8_guard._GR_CACHE['data'] = None
|
|
|
|
def wanted_count(self):
|
|
# Safe set minus wlan1mon.hop (rfplan owns hop), plus WANTED_EXTRA.
|
|
return (len(server.PINEAPD_SAFE_UCI) - 1 +
|
|
len(mk8_guard.WANTED_EXTRA))
|
|
|
|
def test_applies_all_wanted_when_missing(self):
|
|
self.pool = 0
|
|
result = mk8_guard.reconcile(clear_pool=False)
|
|
sets = [c[2] for c in self.calls if c[:2] == ['uci', 'set']]
|
|
self.assertEqual(len(sets), self.wanted_count())
|
|
self.assertTrue(result['changed'])
|
|
self.assertNotIn(HOP_KEY, {c[2].partition('=')[0]
|
|
for c in self.calls
|
|
if c[:2] == ['uci', 'set']})
|
|
|
|
def test_wanted_excludes_hop(self):
|
|
wanted = mk8_guard._wanted()
|
|
self.assertNotIn(HOP_KEY, wanted)
|
|
self.assertEqual(len(wanted), self.wanted_count())
|
|
|
|
def test_clears_large_pool_only(self):
|
|
self.pool = 25
|
|
result = mk8_guard.reconcile(clear_pool=True)
|
|
self.assertTrue(result['pool_cleared'])
|
|
self.assertIn(['uci', 'delete', 'pineapd.@ssidpool[0].ssid'], self.calls)
|
|
|
|
def test_small_pool_untouched(self):
|
|
self.pool = 5
|
|
result = mk8_guard.reconcile(clear_pool=True)
|
|
self.assertFalse(result['pool_cleared'])
|
|
|
|
def test_report_ignores_hop_and_caches(self):
|
|
for key, value in mk8_guard._wanted().items():
|
|
self.uci[key] = value
|
|
self.uci[HOP_KEY] = '1' # attack-role baseline; must stay ignored
|
|
report = mk8_guard.guard_report()
|
|
self.assertTrue(report['in_sync'], report)
|
|
gets = [c[2] for c in self.calls if c[:2] == ['uci', 'get']]
|
|
self.assertNotIn(HOP_KEY, gets)
|
|
n_after_first = len(self.calls)
|
|
self.assertIs(mk8_guard.guard_report(), report)
|
|
self.assertEqual(len(self.calls), n_after_first,
|
|
'guard_report must serve from cache within TTL')
|
|
mk8_guard._GR_CACHE['t'] -= mk8_guard.GR_TTL_SECONDS * 2
|
|
mk8_guard.guard_report()
|
|
self.assertGreater(len(self.calls), n_after_first)
|
|
# reconcile mutates live state; it must invalidate the cached report.
|
|
mk8_guard.guard_report()
|
|
n_cached = len(self.calls)
|
|
mk8_guard.reconcile(clear_pool=False)
|
|
self.assertIsNone(mk8_guard._GR_CACHE['data'])
|
|
mk8_guard.guard_report()
|
|
self.assertGreater(len(self.calls), n_cached)
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|
|
|
|
|
|
class SectionRecreateTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.calls = []
|
|
self.uci = {}
|
|
self.old_server_run = server.device_run
|
|
self.old_iface = server._iface_up
|
|
server._iface_up = lambda name: True
|
|
|
|
def fake_run(args, timeout=20, input_data=None):
|
|
a = list(args)
|
|
self.calls.append(a)
|
|
if a[:3] == ['uci', '-q', 'show']:
|
|
return (1, '', '') # section missing
|
|
if a[:2] == ['uci', 'get']:
|
|
got = self.uci.get(a[2])
|
|
return (0, (got if got is not None else '') + '\n', '')
|
|
if a[:2] == ['uci', 'set']:
|
|
k, _, v = a[2].partition('=')
|
|
self.uci[k] = v
|
|
return (0, '', '')
|
|
|
|
def dual_run(args, timeout=20, input_data=None):
|
|
# install the same fake for server-side helpers
|
|
self.__dict__.setdefault('_srv', server)
|
|
return fake_run(args, timeout=timeout, input_data=input_data)
|
|
|
|
mk8_guard.device_run = fake_run
|
|
server.device_run = fake_run
|
|
|
|
def tearDown(self):
|
|
server.device_run = self.old_server_run
|
|
server._iface_up = self.old_iface
|
|
|
|
def test_recreates_missing_pineapd_section(self):
|
|
result = mk8_guard.reconcile(clear_pool=False)
|
|
self.assertIn(['uci', 'add', 'pineapd', 'pineapd'], self.calls)
|
|
self.assertTrue(any('section recreated' in c for c in result['changed']))
|
|
sets = [c for c in self.calls if c[:2] == ['uci', 'set']]
|
|
self.assertEqual(len(sets),
|
|
len(server.PINEAPD_SAFE_UCI) - 1
|
|
+ len(mk8_guard.WANTED_EXTRA))
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|