diff --git a/payload/user/remote_access/pager-webui/server.py b/payload/user/remote_access/pager-webui/server.py index a5ae128..8500ef3 100644 --- a/payload/user/remote_access/pager-webui/server.py +++ b/payload/user/remote_access/pager-webui/server.py @@ -3341,6 +3341,39 @@ def _sigsegv_count(): return out.count('SIGSEGV') +def _stabilize_pineapd(): + """Idempotent crash-source pass. Field-verified SIGSEGV/terminate sources + on this firmware: + 1. SSID-pool broadcast (segfault, ~15s cadence) + 2. wlan2mon: a 6GHz monitor this hardware never creates; hopping the + missing iface segfaults pineapd + 3. wlan1mon fast-hopping the 6GHz channel list (ASIO io thread exit) + The pool list itself is also cleared: collect refills it, and a large + pool was observed crashing even with broadcast disabled. + Returns a human-readable description of what was changed.""" + actions = [] + wanted = { + 'pineapd.@ssidpool[0].disable': '1', + 'pineapd.wlan2mon.disable': '1', + 'pineapd.wlan2mon.hop': '0', + 'pineapd.wlan1mon.bands': '5', + 'pineapd.wlan0mon.bands': '2', + } + for key, value in wanted.items(): + rc, out, err = device_run(['uci', 'get', key]) + if rc != 0 or out.strip() != value: + device_run(['uci', 'set', '%s=%s' % (key, value)]) + actions.append(key.split('.')[-1]) + rc, out, err = device_run(['uci', 'get', 'pineapd.@ssidpool[0].ssid']) + if rc == 0 and out.strip(): + device_run(['uci', 'delete', 'pineapd.@ssidpool[0].ssid']) + actions.append('pool-list cleared') + if actions: + device_run(['uci', 'commit', 'pineapd']) + return 'stabilized: ' + ', '.join(actions) + return 'pineapd restart' + + def _iface_up(name): """True when the interface's admin flags contain UP. Monitor interfaces report operstate 'unknown' even when usable, so parse `ip link` flags.""" @@ -3383,17 +3416,7 @@ def health_check(): device_run(['ip', 'link', 'set', name, 'up'], timeout=10) h['last_action'] = 'monitor interfaces brought up' else: - # wlan2mon is a 6GHz monitor this hardware never materializes; with - # hop enabled pineapd crashes on the missing iface (second SIGSEGV - # source after the SSID pool). Keep it disabled. - wlan2 = _uci_section('pineapd.wlan2mon') - if wlan2.get('disable') != '1': - device_run(['uci', 'set', 'pineapd.wlan2mon.disable=1']) - device_run(['uci', 'set', 'pineapd.wlan2mon.hop=0']) - device_run(['uci', 'commit', 'pineapd']) - h['last_action'] = 'wlan2mon disabled (missing 6GHz iface crash source)' - else: - h['last_action'] = 'pineapd restart' + h['last_action'] = _stabilize_pineapd() device_run(['/etc/init.d/pineapd', 'restart'], timeout=30) h['sigsegv_last'] = _sigsegv_count() h['last_fix'] = now diff --git a/tests/test_health.py b/tests/test_health.py index 54578e6..080ef80 100644 --- a/tests/test_health.py +++ b/tests/test_health.py @@ -66,16 +66,28 @@ class HealthCheckTest(unittest.TestCase): self.ping_ok = False self.uci_state['pineapd.@ssidpool[0].disable'] = '1' self.uci_state['pineapd.wlan2mon.disable'] = '1' + self.uci_state['pineapd.wlan2mon.hop'] = '0' + self.uci_state['pineapd.wlan1mon.bands'] = '5' + self.uci_state['pineapd.wlan0mon.bands'] = '2' result = server.health_check() self.assertEqual(result['last_action'], 'pineapd restart') self.assertIn(['/etc/init.d/pineapd', 'restart'], [r[0] for r in self.runs]) - def test_down_disables_wlan2mon_crash_source(self): + def test_down_stabilizes_known_crash_sources(self): self.ping_ok = False self.uci_state['pineapd.@ssidpool[0].disable'] = '1' result = server.health_check() - self.assertEqual(result['last_action'], 'wlan2mon disabled (missing 6GHz iface crash source)') self.assertEqual(self.uci_state['pineapd.wlan2mon.disable'], '1') + self.assertEqual(self.uci_state['pineapd.wlan1mon.bands'], '5') + self.assertIn('stabilized', result['last_action']) + + def test_down_clears_refilled_pool_list(self): + self.ping_ok = False + self.uci_state['pineapd.@ssidpool[0].disable'] = '1' + self.uci_state['pineapd.@ssidpool[0].ssid'] = 'QmVlcg==' + result = server.health_check() + self.assertNotIn('pineapd.@ssidpool[0].ssid', self.uci_state) + self.assertIn('pool-list cleared', result['last_action']) def test_down_without_crash_brings_monitors_up(self): self.ping_ok = False