fix: harden UI actions and daemon calls for reliable control (v1.3.2)
Retry and serialize pineapd/hak5 calls, queue virtual-pager keys, and grey out buttons until the pager finishes. Deploy now installs python3-light after factory firmware. Bump version to 1.3.2. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Cursor
@@ -0,0 +1,159 @@
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'payload', 'user', 'remote_access', 'pager-webui'))
|
||||
import server
|
||||
|
||||
|
||||
class PagerTruthTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.uci = {
|
||||
'pineapd.wlan1mon.hop': '1',
|
||||
'pineapd.wlan2mon.disable': '0',
|
||||
'wireless.dummy_radio0.disabled': '0',
|
||||
'wireless.radio1.channel': 'auto',
|
||||
'wireless.radio1.band': '5g',
|
||||
'wireless.wlan1open': None,
|
||||
'wireless.wlan1wpa': None,
|
||||
}
|
||||
self.ifaces = {}
|
||||
fd, self.snap = tempfile.mkstemp(suffix='.json')
|
||||
os.close(fd)
|
||||
os.unlink(self.snap)
|
||||
self.old_file = server.PAGER_SNAPSHOT_FILE
|
||||
self.old_pineap = server.PINEAP_STATE_FILE
|
||||
server.PAGER_SNAPSHOT_FILE = self.snap
|
||||
server.PINEAP_STATE_FILE = self.snap + '.pineap'
|
||||
self.old_run = server.device_run
|
||||
server.device_run = self.fake_run
|
||||
self.old_ent = server._disable_enterprise_ap
|
||||
server._disable_enterprise_ap = lambda resume_hop=True: None
|
||||
|
||||
def tearDown(self):
|
||||
server.device_run = self.old_run
|
||||
server._disable_enterprise_ap = self.old_ent
|
||||
server.PAGER_SNAPSHOT_FILE = self.old_file
|
||||
server.PINEAP_STATE_FILE = self.old_pineap
|
||||
for path in (self.snap, self.snap + '.tmp', self.snap + '.pineap'):
|
||||
try:
|
||||
os.unlink(path)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
def fake_run(self, args, timeout=20, input_data=None):
|
||||
a = list(args)
|
||||
if a[:2] == ['uci', '-q'] and a[2] == 'get':
|
||||
key = a[3]
|
||||
if key.startswith('wireless.') and key.count('.') == 1:
|
||||
name = key.split('.', 1)[1]
|
||||
if name in self.ifaces:
|
||||
return 0, 'wifi-iface\n', ''
|
||||
return 1, '', ''
|
||||
val = self.uci.get(key)
|
||||
if val is None:
|
||||
return 1, '', ''
|
||||
return 0, val + '\n', ''
|
||||
if a[:2] == ['uci', 'show']:
|
||||
sec = a[2]
|
||||
name = sec.split('.', 1)[-1]
|
||||
cfg = self.ifaces.get(name) or {}
|
||||
body = ''.join("%s.%s='%s'\n" % (sec, k, v) for k, v in cfg.items())
|
||||
return (0, body, '') if cfg or name in self.ifaces else (1, '', '')
|
||||
if a[:2] == ['uci', 'set']:
|
||||
expr = a[2]
|
||||
if '=' not in expr:
|
||||
return 0, '', ''
|
||||
key, _, val = expr.partition('=')
|
||||
parts = key.split('.')
|
||||
if len(parts) == 2 and parts[0] == 'wireless' and val == 'wifi-iface':
|
||||
self.ifaces.setdefault(parts[1], {})
|
||||
return 0, '', ''
|
||||
if len(parts) == 3 and parts[0] == 'wireless' and (
|
||||
parts[1] in self.ifaces or parts[1] in ('wlan1open', 'wlan1wpa', 'wlan1ent')):
|
||||
self.ifaces.setdefault(parts[1], {})[parts[2]] = val
|
||||
return 0, '', ''
|
||||
self.uci[key] = val
|
||||
return 0, '', ''
|
||||
if a[:2] == ['uci', 'delete']:
|
||||
key = a[2]
|
||||
parts = key.split('.')
|
||||
if len(parts) == 2 and parts[0] == 'wireless':
|
||||
self.ifaces.pop(parts[1], None)
|
||||
self.uci.pop(key, None)
|
||||
return 0, '', ''
|
||||
if a[:2] == ['uci', 'commit']:
|
||||
return 0, '', ''
|
||||
if a[0] in ('wifi', '/etc/init.d/pineapd', 'kill', 'iw'):
|
||||
return 0, '', ''
|
||||
return 0, '', ''
|
||||
|
||||
def test_restore_reverts_radio1_ap_and_hop(self):
|
||||
server.capture_pager_snapshot()
|
||||
self.assertTrue(os.path.isfile(self.snap))
|
||||
self.uci['pineapd.wlan1mon.hop'] = '0'
|
||||
self.ifaces['wlan1open'] = {'ssid': 'EvilTwin', 'disabled': '0', 'device': 'radio1'}
|
||||
self.uci['wireless.dummy_radio0.disabled'] = '1'
|
||||
result = server.restore_pager_truth()
|
||||
self.assertTrue(result['ok'])
|
||||
self.assertTrue(result['restored'])
|
||||
self.assertEqual(self.uci['pineapd.wlan1mon.hop'], '1')
|
||||
self.assertNotIn('wlan1open', self.ifaces)
|
||||
self.assertEqual(self.uci['wireless.dummy_radio0.disabled'], '0')
|
||||
self.assertFalse(os.path.isfile(self.snap))
|
||||
|
||||
def test_restore_without_snapshot_is_safe(self):
|
||||
result = server.restore_pager_truth()
|
||||
self.assertTrue(result['ok'])
|
||||
self.assertFalse(result['restored'])
|
||||
self.assertEqual(result['reason'], 'no snapshot')
|
||||
|
||||
def test_mode_get_reports_snapshot(self):
|
||||
server.capture_pager_snapshot()
|
||||
status, payload = server.h_mode_get(None)
|
||||
self.assertEqual(status, 200)
|
||||
self.assertTrue(payload['snapshot'])
|
||||
self.assertTrue(payload['markviii'])
|
||||
self.assertEqual(payload['pager_port'], 1471)
|
||||
|
||||
def test_payload_refresh_falls_back_to_disk(self):
|
||||
tmp = tempfile.mkdtemp()
|
||||
self.addCleanup(lambda: shutil.rmtree(tmp, ignore_errors=True))
|
||||
payload_dir = tmp
|
||||
for part in ('user', 'games', 'snake'):
|
||||
payload_dir = os.path.join(payload_dir, part)
|
||||
if not os.path.isdir(payload_dir):
|
||||
os.mkdir(payload_dir)
|
||||
with open(os.path.join(payload_dir, 'payload.sh'), 'w') as handle:
|
||||
handle.write('#!/bin/sh\n')
|
||||
old_roots = server.PAYLOAD_ROOTS
|
||||
old_daemon = server._payload_daemon
|
||||
server.PAYLOAD_ROOTS = (tmp,)
|
||||
server._payload_daemon = lambda *a, **k: (500, {'error': 'portal down'})
|
||||
try:
|
||||
status, data = server.h_payloads_refresh(None)
|
||||
finally:
|
||||
server.PAYLOAD_ROOTS = old_roots
|
||||
server._payload_daemon = old_daemon
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(len(data['payloads']), 1)
|
||||
self.assertEqual(data['payloads'][0]['key'], 'user~games~snake')
|
||||
self.assertIn('warning', data)
|
||||
|
||||
def test_respawn_keeps_original_snapshot(self):
|
||||
server.capture_pager_snapshot()
|
||||
self.uci['pineapd.wlan1mon.hop'] = '0'
|
||||
server.capture_pager_snapshot()
|
||||
result = server.restore_pager_truth()
|
||||
self.assertTrue(result['restored'])
|
||||
self.assertEqual(self.uci['pineapd.wlan1mon.hop'], '1')
|
||||
|
||||
def test_uci_get_keeps_settings_default(self):
|
||||
self.assertEqual(server._uci_get('missing.key', 'UTC'), 'UTC')
|
||||
self.assertIsNone(server._uci_get('missing.key'))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user