fix(enterprise,deauth,filters): PineAP validation round fixes, live-verified on Pager 24.10.1

- deauth/kick: use full PINEAPPLE_DEAUTH_CLIENT hak5cmd app name (BUG 1) - _allow_all_ssids: PINEAPPLE_NETWORK_FILTER_MODE deny so karma lets clients associate after deploy (BUG 2) - enterprise inner EAP (BUG 3), two root causes found by live experiment: - hostapd never wildcard-matches a bare `*` identity for phase-2 lookups; eap_users now uses quoted empty prefix `""* ... [2]`, which prefix-matches any inner identity - pineape_auth_pass=1 forwards inner EAP to pineapd, which has no standalone responder; deploy forces 0 and stop restores 1 Residual: plaintext MSCHAPv2 capture is firmware-blocked on 24.10.1 (MSG_DEBUG compiled out of the karma wpad; `-f` silently ignored); documented in code comments and the validation report. - ISSUE 4: point pineapd.@hostapd[0].mgmtiface at wlan1wpa during 5 GHz WPA deploys so handshakes/loot populate; cleared on stop - ISSUE 5: radio0 set_ap path polls 90 s across the wifi-reload window and retries set_ap once if the iface still has not appeared - ISSUE 6: skills docs use `llc && eth.type == 0x888e` (firmware tcpdump matches 0 frames on `eapol`) - ISSUE 7: capture state helper detects dead-pid / iface-down, cleans up, status reports {running:false, stale:true} instead of zombie running - tests: fix global os.path monkeypatch leaks between test modules that broke test_mk8_events/test_reliability_api under discovery; add regression coverage for all fixes above (463 tests green) Live validation evidence and newly discovered firmware quirks (DEVICE_FILTER_DELETE no-op, dropbear rate limiting) recorded in docs/validation/2026-08-23-pineap-validation-report.md.
2026-08-23 18:29:47 -06:00
parent a04319dfc8
commit ed50cd7b5a
6 changed files with 721 additions and 48 deletions
@@ -94,6 +94,13 @@ class AttacksDeployTest(unittest.TestCase):
        server.ENT_LOG = os.path.join(server.ENT_DIR, 'hostapd.log')
        server.ENT_CAPTURES = os.path.join(server.ENT_DIR, 'captures.json')
        server.ENT_DH_FILE = os.path.join(server.ENT_DIR, 'dh.pem')
+        # The fake device_run cannot execute real openssl; pre-seed the cert
+        # files so _ensure_ent_certs() short-circuits to True.
+        os.makedirs(server.ENT_DIR, exist_ok=True)
+        for p in (server.ENT_CA_CERT, server.ENT_SERVER_CERT,
+                  server.ENT_SERVER_KEY, server.ENT_DH_FILE):
+            with open(p, 'w') as f:
+                f.write('stub\n')
        self.old_ent_running = server._ent_running
        self.old_ent_state = server._ent_state_loaded
        server._ent_running = lambda: True
@@ -215,6 +222,7 @@ class AttacksDeployTest(unittest.TestCase):
        self.assertTrue(any(c[:4] == ['/usr/sbin/hostapd', '-B', '-P', server.ENT_PIDFILE]
                            for c in cmds))
        self.assertEqual(self.f.state['pineapd.@hostapd[0].mgmtiface'], 'wlan1ent')
+        self.assertEqual(self.f.state['pineapd.@hostapd[0].pineape_auth_pass'], '0')
        self.assertEqual(self.f.state['pineapd.wlan1mon.hop'], '0')
        with open(server.ENT_CONF) as f:
            conf = f.read()
@@ -231,6 +239,22 @@ class AttacksDeployTest(unittest.TestCase):
            users = f.read()
        self.assertIn('PEAP,TTLS', users)
        self.assertIn('[2]', users)
+        # Phase-2 entries must use a quoted empty prefix: hostapd never
+        # wildcard-matches a bare `*` identity for phase2 lookups.
+        self.assertIn('""*', users)
+        self.assertNotRegex(users, r'(?m)^\*\t.*\[2\]$')
+
+    def test_eap_users_text_grammar_matches_pineapple_wpad(self):
+        text = server._eap_users_text('secret123', 'any')
+        lines = text.splitlines()
+        self.assertEqual(lines[0], '*\tPEAP,TTLS')
+        self.assertTrue(lines[1].startswith('""*\t'))
+        self.assertIn('"secret123"', lines[1])
+        self.assertTrue(lines[1].endswith('[2]'))
+        # A bare-wildcard phase-2 line is the firmware-broken grammar.
+        for line in lines:
+            if line.endswith('[2]'):
+                self.assertFalse(line.startswith('*\t'))

    def test_deploy_enterprise_rejects_non_5g_channel(self):
        status, _ = server.h_attacks_deploy(ctx({
@@ -241,11 +265,15 @@ class AttacksDeployTest(unittest.TestCase):
    def test_stop_enterprise_tears_down_engine(self):
        server._ent_running = lambda: True
        server._ent_state_loaded = lambda: {'ssid': 'CorpAP', 'channel': 36}
+        self.f.state['pineapd.@hostapd[0].pineape_auth_pass'] = '0'
        status, payload = server.h_attacks_stop(ctx({'kind': 'enterprise'}))
        self.assertEqual(status, 200)
        self.assertIn('wlan1ent', payload['stopped'])
        cmds = [r[0] for r in self.f.runs]
        self.assertIn(['iw', 'dev', 'wlan1ent', 'del'], cmds)
+        # Deploy disabled PineAPE auth passthrough; stop must restore it.
+        self.assertEqual(self.f.state['pineapd.@hostapd[0].pineape_auth_pass'],
+                         '1')

    def test_deploy_enterprise_writes_passphrase_and_hidden(self):
        status, payload = server.h_attacks_deploy(ctx({
@@ -360,7 +388,7 @@ class AttacksDeauthTest(unittest.TestCase):
        self.assertEqual(payload['inject'], 'wlan0mon')
        calls = [r[0] for r in self.f.runs]
        self.assertIn(['_pineap', 'INTERFACE', 'INJECT', 'wlan0mon'], calls)
-        self.assertIn(['/usr/bin/hak5cmd', 'DEAUTH_CLIENT', 'AA:BB:CC:DD:EE:FF',
+        self.assertIn(['/usr/bin/hak5cmd', 'PINEAPPLE_DEAUTH_CLIENT', 'AA:BB:CC:DD:EE:FF',
                       '11:22:33:44:55:66', '6'], calls)

    def test_deauth_5g_keeps_wlan1mon_inject(self):
@@ -376,6 +404,20 @@ class AttacksDeauthTest(unittest.TestCase):
        self.assertEqual(status, 400)


+class AttackFilterModeTest(unittest.TestCase):
+    def test_allow_all_ssids_uses_full_network_filter_app_name(self):
+        calls = []
+        old_run = server.device_run
+        server.device_run = lambda args, timeout=20, input_data=None: (
+            calls.append(list(args)) or (0, '', ''))
+        try:
+            self.assertTrue(server._allow_all_ssids())
+        finally:
+            server.device_run = old_run
+        self.assertEqual(calls, [[server.HAK5CMD,
+                                  'PINEAPPLE_NETWORK_FILTER_MODE', 'deny']])
+
+
 class AttacksExportTest(unittest.TestCase):
    def setUp(self):
        self.f = FakeUciDevice()
@@ -394,11 +436,13 @@ class AttacksExportTest(unittest.TestCase):
        server.device_run = fake_run
        server.daemon_sock_call = lambda method, path, body=None, timeout=10: (200, {})
        self._real_exists = os.path.exists
+        self._real_getsize = os.path.getsize
        server.os.path.exists = lambda p: p.endswith('.hc22000') or p.startswith('/sys')
        server.os.path.getsize = lambda p: 12

    def tearDown(self):
        server.os.path.exists = self._real_exists
+        server.os.path.getsize = self._real_getsize
        try:
            os.unlink('/tmp/mk8test.hc22000')
        except OSError:
@@ -455,5 +499,108 @@ class AttacksStatusTest(unittest.TestCase):
        self.assertTrue(payload['enterprise']['pineape']['enabled'])


+class AttacksCaptureTest(unittest.TestCase):
+    def setUp(self):
+        self.f = FakeUciDevice()
+        server.device_run = self.f.device_run
+        server.daemon_sock_call = lambda *a, **k: (200, {})
+        server._uci_wifi_iface = self.f.uci_iface
+        server._uci_section = self.f.uci_iface
+        server._verify_iface = lambda name, timeout=20: self.f._verify
+        self.pidfile = tempfile.mktemp(prefix='mk8-cap-test-')
+        self.real_exists = os.path.exists
+
+    def tearDown(self):
+        os.path.exists = self.real_exists
+        try:
+            os.unlink(self.pidfile)
+        except OSError:
+            pass
+
+    def test_status_reports_dead_pid_as_stopped_stale(self):
+        with open(self.pidfile, 'w') as f:
+            f.write('999999\n')
+        running, pid, stale = server._capture_state(self.pidfile, 'wlan0mon')
+        self.assertFalse(running)
+        self.assertTrue(stale)
+        self.assertFalse(os.path.exists(self.pidfile),
+                         'stale pidfile must be cleaned up')
+
+    def test_status_kills_capture_when_iface_dropped(self):
+        with open(self.pidfile, 'w') as f:
+            f.write('4242\n')
+        real_exists = os.path.exists
+        killed = []
+        old_run = server.device_run
+
+        def fake_run(args, timeout=20, input_data=None):
+            if args[0] == 'kill':
+                killed.append(args[1])
+            return (0, '', '')
+
+        server.device_run = fake_run
+        try:
+            os.path.exists = lambda p: p.startswith('/proc/4242')
+            running, pid, stale = server._capture_state(
+                self.pidfile, 'wlan0mon')
+        finally:
+            os.path.exists = real_exists
+            server.device_run = old_run
+        self.assertFalse(running)
+        self.assertTrue(stale)
+        self.assertEqual(killed, ['4242'])
+
+    def test_status_live_capture_running(self):
+        with open(self.pidfile, 'w') as f:
+            f.write(str(os.getpid()))
+        real_exists = os.path.exists
+        try:
+            # /proc/<pid> exists for our own process; iface path faked up.
+            os.path.exists = lambda p: (
+                not p.startswith('/sys/class/net') or p.endswith('wlan0mon'))
+            running, pid, stale = server._capture_state(
+                self.pidfile, 'wlan0mon')
+        finally:
+            os.path.exists = real_exists
+        self.assertTrue(running)
+        self.assertFalse(stale)
+
+    def test_deploy_retries_radio0_set_ap_when_iface_never_verifies(self):
+        calls = {'set_ap': 0}
+
+        def sock(method, path, body=None, timeout=10):
+            if path == '/api/settings/wifi/set_ap':
+                calls['set_ap'] += 1
+            return (200, {})
+
+        server.daemon_sock_call = sock
+        self.f._verify = False
+        status, payload = server.h_attacks_deploy(ctx({
+            'kind': 'open', 'ssid': 'Guest', 'channel': 1}))
+        self.assertEqual(status, 200)
+        self.assertFalse(payload['verified'])
+        self.assertEqual(calls['set_ap'], 2,
+                         'radio0 deploy must retry set_ap once')
+
+    def test_deploy_5g_does_not_retry_set_ap(self):
+        calls = {'set_ap': 0}
+
+        def sock(method, path, body=None, timeout=10):
+            if path == '/api/settings/wifi/set_ap':
+                calls['set_ap'] += 1
+            return (200, {})
+
+        server.daemon_sock_call = sock
+        self.f._verify = False
+        status, payload = server.h_attacks_deploy(ctx({
+            'kind': 'wpa', 'ssid': 'Corp', 'passphrase': 'secretpass1',
+            'enctype': 'psk2', 'channel': 36}))
+        self.assertEqual(status, 200)
+        self.assertEqual(calls['set_ap'], 0,
+                         '5 GHz path writes UCI directly, no set_ap')
+        self.assertEqual(self.f.state['pineapd.@hostapd[0].mgmtiface'],
+                         'wlan1wpa')
+
+
 if __name__ == '__main__':
    unittest.main()
@@ -345,20 +345,24 @@ class GetApReconcileTest(unittest.TestCase):
            404, {'error': 'not found'})
        server.device_run = lambda args, timeout=20, input_data=None: (
            self.runs.append(list(args)) or (0, '', ''))
+        self._real_exists = os.path.exists
+
+    def tearDown(self):
+        os.path.exists = self._real_exists

    def test_missing_netdev_triggers_wifi_reload(self):
-        server.os.path.exists = lambda p: False
+        os.path.exists = lambda p: False
        server.h_pineap_wifi_get_ap(ctx())
        self.assertIn(['wifi', 'reload'], self.runs)

    def test_present_netdev_skips_reload(self):
-        server.os.path.exists = lambda p: True
+        os.path.exists = lambda p: True
        server.h_pineap_wifi_get_ap(ctx())
        self.assertNotIn(['wifi', 'reload'], self.runs)

    def test_disabled_section_skips_reload(self):
        self.uci['wlan1wpa']['disabled'] = '1'
-        server.os.path.exists = lambda p: False
+        os.path.exists = lambda p: False
        server.h_pineap_wifi_get_ap(ctx())
        self.assertNotIn(['wifi', 'reload'], self.runs)

@@ -55,7 +55,7 @@ class ClientsTest(unittest.TestCase):
        server.h_client_kick(type('C', (), {'args': (), 'body': {'mac': '00:11:22:33:44:55'}})())
        self.assertIn([server.HAK5CMD, 'PINEAPPLE_DEVICE_FILTER_ADD', 'deny', '00:11:22:33:44:55'], calls)
        # The immediate deauth must use the full bssid/target/channel form.
-        self.assertTrue(any(c[:4] == [server.HAK5CMD, 'DEAUTH_CLIENT', 'AA:BB:CC:DD:EE:FF',
+        self.assertTrue(any(c[:4] == [server.HAK5CMD, 'PINEAPPLE_DEAUTH_CLIENT', 'AA:BB:CC:DD:EE:FF',
                                      '00:11:22:33:44:55'] and c[4] == '6' for c in calls))

    def test_kick_not_associated_still_filters(self):
@@ -86,7 +86,7 @@ class ClientsTest(unittest.TestCase):
        status, payload = server.h_deauth_client(type('C', (), {'args': (), 'body': {'mac': '00:11:22:33:44:55'}})())
        self.assertEqual(status, 200)
        # 5 GHz client -> wlan1mon inject, no _pineap pin needed.
-        self.assertTrue(any(c[:4] == [server.HAK5CMD, 'DEAUTH_CLIENT', 'AA:BB:CC:DD:EE:FF',
+        self.assertTrue(any(c[:4] == [server.HAK5CMD, 'PINEAPPLE_DEAUTH_CLIENT', 'AA:BB:CC:DD:EE:FF',
                                      '00:11:22:33:44:55'] and c[4] == '149' for c in calls))

    def test_deauth_client_not_associated_502(self):