diff --git a/payload/user/remote_access/pager-webui/mk8_guard.py b/payload/user/remote_access/pager-webui/mk8_guard.py index d84ed42..d8aef14 100644 --- a/payload/user/remote_access/pager-webui/mk8_guard.py +++ b/payload/user/remote_access/pager-webui/mk8_guard.py @@ -1,10 +1,26 @@ """Boot-time reconciliation of crash-prone PineAP settings.""" +import time + from server import (_apply_uci_wanted, _monitor_down, _raise_monitors, PINEAPD_SAFE_UCI, device_run) WANTED_EXTRA = {'pineapd.@pineapd[0].autossidpool': '0'} POOL_CLEAR_MAX = 20 MONITORS = ('wlan0mon', 'wlan1mon') +# pineapd.wlan1mon.hop is intentionally NOT part of the applied set: channel +# hopping is owned by the RF role manager (_pause_hop/_resume_hop). Hop=1 is +# the healthy recon baseline, so reconciling it here would silently disable a +# pager-enabled setting at every boot and latch GUARD PENDING after any +# attack-role switch. +GR_TTL_SECONDS = 30 +_GR_CACHE = {'t': 0.0, 'data': None} + + +def _wanted(): + wanted = {k: v for k, v in PINEAPD_SAFE_UCI.items() + if k != 'pineapd.wlan1mon.hop'} + wanted.update(WANTED_EXTRA) + return wanted def _pool_size(): @@ -16,7 +32,7 @@ def _pool_size(): def reconcile(clear_pool=True): - changed = _apply_uci_wanted(dict(PINEAPD_SAFE_UCI, **WANTED_EXTRA)) + changed = _apply_uci_wanted(_wanted()) pool_cleared = False if clear_pool and _pool_size() > POOL_CLEAR_MAX: device_run(['uci', 'delete', 'pineapd.@ssidpool[0].ssid']) @@ -24,12 +40,20 @@ def reconcile(clear_pool=True): if changed or pool_cleared: device_run(['uci', 'commit', 'pineapd']) raised = _raise_monitors() if any(_monitor_down(m) for m in MONITORS) else [] + _GR_CACHE['data'] = None return {'changed': changed, 'pool_cleared': pool_cleared, 'monitors_raised': raised} def guard_report(): + now = time.time() + cached = _GR_CACHE['data'] + if cached is not None and now - _GR_CACHE['t'] < GR_TTL_SECONDS: + return cached from server import _pending_uci - pending = _pending_uci(dict(PINEAPD_SAFE_UCI, **WANTED_EXTRA)) - return {'in_sync': not pending, 'pending': pending, - 'pool_size': _pool_size()} + pending = _pending_uci(_wanted()) + report = {'in_sync': not pending, 'pending': pending, + 'pool_size': _pool_size()} + _GR_CACHE['t'] = now + _GR_CACHE['data'] = report + return report diff --git a/scripts/smoke.sh b/scripts/smoke.sh index 88eb73f..8f7562c 100755 --- a/scripts/smoke.sh +++ b/scripts/smoke.sh @@ -166,7 +166,6 @@ pineapd.wlan2mon.disable=1 pineapd.wlan2mon.hop=0 pineapd.wlan1mon.bands=5 pineapd.wlan0mon.bands=2 -pineapd.wlan1mon.hop=0 pineapd.@pineapd[0].autossidpool=0 EOF pool="$(uci_get 'pineapd.@ssidpool[0].ssid')" diff --git a/tests/test_mk8_guard.py b/tests/test_mk8_guard.py index 569b1c9..93eb1bd 100644 --- a/tests/test_mk8_guard.py +++ b/tests/test_mk8_guard.py @@ -3,6 +3,8 @@ sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'payload', 'use import server import mk8_guard +HOP_KEY = 'pineapd.wlan1mon.hop' + class GuardTest(unittest.TestCase): def setUp(self): self.calls = [] @@ -30,19 +32,33 @@ class GuardTest(unittest.TestCase): 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), - len(server.PINEAPD_SAFE_UCI) + len(mk8_guard.WANTED_EXTRA)) + 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 @@ -55,5 +71,28 @@ class GuardTest(unittest.TestCase): 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() diff --git a/tests/test_reliability_api.py b/tests/test_reliability_api.py index 3eeb9eb..304e54d 100644 --- a/tests/test_reliability_api.py +++ b/tests/test_reliability_api.py @@ -32,6 +32,8 @@ class ReliabilityApiTest(unittest.TestCase): def tearDown(self): server.device_run = self.old_device_run + import mk8_guard + mk8_guard._GR_CACHE['data'] = None def test_h_health_exposes_reliability_feed(self): status, h = server.h_health(None)