diff --git a/payload/user/remote_access/pager-webui/server.py b/payload/user/remote_access/pager-webui/server.py index ebce502..8e329c9 100644 --- a/payload/user/remote_access/pager-webui/server.py +++ b/payload/user/remote_access/pager-webui/server.py @@ -3445,11 +3445,16 @@ def _read_hop(): return out.strip() +_HOP_PAUSE_HELD = False # True while a Mark VIII role/attack holds hop off + + def _pause_hop(): + global _HOP_PAUSE_HELD if _read_hop() != '0': device_run(['uci', 'set', 'pineapd.wlan1mon.hop=0']) device_run(['uci', 'commit', 'pineapd']) device_run(['/etc/init.d/pineapd', 'reload']) + _HOP_PAUSE_HELD = True def _resume_hop(): @@ -3468,6 +3473,8 @@ def _resume_hop(): device_run(['uci', 'set', 'pineapd.wlan1mon.hop=%s' % wanted]) device_run(['uci', 'commit', 'pineapd']) device_run(['/etc/init.d/pineapd', 'reload']) + global _HOP_PAUSE_HELD + _HOP_PAUSE_HELD = False def _remove_radio1_ap(): @@ -4789,6 +4796,20 @@ def health_check(): msg='reconciler re-applied: %s' % ','.join(rep['changed'])) except Exception: pass + # Restore the hopping baseline unless a Mark VIII role/attack holds + # the pause intentionally (_resume_hop self-guards against fighting a + # live Pager hop enable). + if not _HOP_PAUSE_HELD: + try: + before = _read_hop() + if before == '0': + import mk8_events + _resume_hop() + mk8_events.log_event( + 'guard_fix', + msg='restored wlan1mon.hop baseline (was %s)' % before) + except Exception: + pass h['mem_percent'] = _mem_percent() if h['mem_percent'] >= MEM_WARN_PERCENT: h['mem_streak'] = h.get('mem_streak', 0) + 1 diff --git a/tests/test_health.py b/tests/test_health.py index 2e10866..1dd491b 100644 --- a/tests/test_health.py +++ b/tests/test_health.py @@ -244,3 +244,65 @@ class SupervisorExtrasTest(unittest.TestCase): if __name__ == '__main__': unittest.main() + + +class HopBaselineTest(unittest.TestCase): + def setUp(self): + self.runs = [] + self.old_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.runs.append(a) + if a[:2] == ['pidof', 'pineapd']: + return (0, '123\n', '') + if a[:3] == ['uci', 'get', 'pineapd.wlan1mon.hop']: + return (0, self.hop + '\n', '') + return (0, '', '') + + server.device_run = fake_run + self.hop = '0' + + def tearDown(self): + server.device_run = self.old_run + server._iface_up = self.old_iface + server._health['ticks'] = 0 + + def test_resume_called_when_not_held(self): + import unittest.mock as um + old_held = server._HOP_PAUSE_HELD + server._HOP_PAUSE_HELD = False + server._health['ticks'] = server.HEALTH_STA_PARK_INTERVAL - 1 + try: + with mock.patch.object(server, '_sta_uplink_enabled', + return_value=False), \ + mock.patch.object(server, '_raise_monitors', + return_value=[]), \ + mock.patch('mk8_guard.reconcile', + return_value={'changed': [], 'pool_cleared': False, + 'monitors_raised': []}), \ + mock.patch.object(server, '_resume_hop') as resume: + server.health_check() + resume.assert_called_once() + finally: + server._HOP_PAUSE_HELD = old_held + + def test_resume_skipped_while_pause_held(self): + old_held = server._HOP_PAUSE_HELD + server._HOP_PAUSE_HELD = True + server._health['ticks'] = server.HEALTH_STA_PARK_INTERVAL - 1 + try: + with mock.patch.object(server, '_sta_uplink_enabled', + return_value=False), \ + mock.patch.object(server, '_raise_monitors', + return_value=[]), \ + mock.patch('mk8_guard.reconcile', + return_value={'changed': [], 'pool_cleared': False, + 'monitors_raised': []}), \ + mock.patch.object(server, '_resume_hop') as resume: + server.health_check() + resume.assert_not_called() + finally: + server._HOP_PAUSE_HELD = old_held