diff --git a/payload/user/remote_access/pager-webui/server.py b/payload/user/remote_access/pager-webui/server.py index 3cc1529..d8fec4b 100644 --- a/payload/user/remote_access/pager-webui/server.py +++ b/payload/user/remote_access/pager-webui/server.py @@ -860,17 +860,25 @@ def h_client_kick(ctx): mac = normalize_mac((ctx.body or {}).get('mac')) if not mac: return 400, {'error': 'invalid mac'} - hak5('PINEAPPLE_DEVICE_FILTER_MODE', 'deny') - hak5('PINEAPPLE_DEVICE_FILTER_ADD', 'deny', mac) - hak5('PINEAPPLE_DEAUTH_CLIENT', mac) - return 200, {'ok': True} + # Persistent kick: deny-filter the client so pineapd deauths every probe + # and connect, then deauth it once immediately with the full + # (bssid, target, channel) form hak5cmd requires. + for argv in ([HAK5CMD, 'PINEAPPLE_DEVICE_FILTER_MODE', 'deny'], + [HAK5CMD, 'PINEAPPLE_DEVICE_FILTER_ADD', 'deny', mac]): + rc, out, err = device_run(argv, timeout=20) + if rc != 0: + return 502, {'error': 'kick filter failed', 'detail': (err or out)[-300:]} + ok, detail = _deauth_client_via_iface(mac) + return 200, {'ok': True, 'deauth': ok, 'detail': detail or None} def h_deauth_client(ctx): mac = normalize_mac((ctx.body or {}).get('mac')) if not mac: return 400, {'error': 'invalid mac'} - hak5('PINEAPPLE_DEAUTH_CLIENT', mac) + ok, detail = _deauth_client_via_iface(mac) + if not ok: + return 502, {'error': 'deauth failed', 'detail': detail} return 200, {'ok': True} @@ -2298,6 +2306,53 @@ def assoc_clients(ifaces=None): return clients +def _iface_ap_info(iface): + """(bssid, channel) of the AP running on `iface`, via iwinfo.""" + rc, out, err = device_run(['iwinfo', iface, 'info']) + bssid = None + channel = None + for line in out.splitlines(): + m = re.search(r'Access Point:\s*([0-9A-Fa-f]{2}(?::[0-9A-Fa-f]{2}){5})', line) + if m: + bssid = m.group(1).upper() + m = re.search(r'Channel:\s*(\d+)', line) + if m: + channel = int(m.group(1)) + return bssid, channel + + +def _deauth_target(mac): + """(iface, bssid, channel) for an associated client, else None.""" + for c in assoc_clients(): + if c['mac'] != mac: + continue + bssid, channel = _iface_ap_info(c['iface']) + if not bssid or not channel: + return None + return c['iface'], bssid, channel + return None + + +def _deauth_client_via_iface(mac): + """Deauth a client associated to one of our own APs. + + hak5cmd's deauth needs the full (bssid, target, channel) triple; the + client's AP and channel are resolved from its association interface. + Returns (ok, detail). + """ + target = _deauth_target(mac) + if not target: + return False, 'client not associated' + iface, bssid, channel = target + band = _band_of_channel(channel) + inject = 'wlan1mon' if band == BAND_5G or band == BAND_6G else 'wlan0mon' + if inject != 'wlan1mon': + _pineap('INTERFACE', 'INJECT', inject) + rc, out, err = device_run([HAK5CMD, 'DEAUTH_CLIENT', bssid, mac, + str(channel)], timeout=30) + return rc == 0, (err or out) + + def disk_data(): rc, out, err = device_run(['df', '-k', '/root']) lines = out.splitlines() @@ -3698,11 +3753,27 @@ def _mcp_tools(): return {'devices': _sql_table('wifi_device', args.get('limit', 50))} def kick(args): - mac = (args.get('mac') or '').strip() + mac = normalize_mac(args.get('mac')) if not mac: return {'error': 'mac required'} - rc, out, err = device_run([HAK5CMD, 'CLIENT_KICK', mac], timeout=20) - return {'ok': rc == 0, 'detail': (err or out)[-300:]} + # This firmware's hak5cmd has no CLIENT_KICK command; mirror the web + # UI's kick: deny-filter the client (deauths every probe/connect) then + # deauth it once with the full (bssid, target, channel) form. The + # client must be associated first so a failed kick has no side effects. + if not _deauth_target(mac): + return {'ok': False, 'detail': 'client not associated', 'mac': mac} + ok = True + detail = '' + for argv in ([HAK5CMD, 'PINEAPPLE_DEVICE_FILTER_MODE', 'deny'], + [HAK5CMD, 'PINEAPPLE_DEVICE_FILTER_ADD', 'deny', mac]): + rc, out, err = device_run(argv, timeout=20) + if rc != 0: + ok = False + detail = (err or out)[-300:] + break + if ok: + ok, detail = _deauth_client_via_iface(mac) + return {'ok': ok, 'detail': detail, 'mac': mac} def set_filter(args): kind = (args.get('kind') or 'ssid').strip() diff --git a/tests/test_pineap_clients.py b/tests/test_pineap_clients.py index 6780d93..d9579ed 100644 --- a/tests/test_pineap_clients.py +++ b/tests/test_pineap_clients.py @@ -9,8 +9,10 @@ import server def setUpModule(): __import__('importlib').reload(server) -_orig_hak5 = server.hak5 +_orig_device_run = server.device_run _orig_assoc_clients = server.assoc_clients +_orig_iface_ap_info = server._iface_ap_info +_orig_pineap = server._pineap class NormalizeTest(unittest.TestCase): @@ -24,8 +26,10 @@ class NormalizeTest(unittest.TestCase): class ClientsTest(unittest.TestCase): def tearDown(self): - server.hak5 = _orig_hak5 + server.device_run = _orig_device_run server.assoc_clients = _orig_assoc_clients + server._iface_ap_info = _orig_iface_ap_info + server._pineap = _orig_pineap def test_clients_handler(self): server.assoc_clients = lambda: [{'mac': 'AA:BB:CC:DD:EE:FF', 'iface': 'wlan0open', 'rssi': -55}] @@ -35,15 +39,37 @@ class ClientsTest(unittest.TestCase): self.assertEqual(status, 200) self.assertEqual(payload['count'], 1) - def test_kick_validates_and_deny_adds(self): + def _kick_env(self): calls = [] - def fake(*args): - calls.append(args) - return 'ok' - server.hak5 = fake + def fake_run(argv, timeout=30): + calls.append(argv) + return 0, '', '' + server.device_run = fake_run + server.assoc_clients = lambda: [{'mac': '00:11:22:33:44:55', 'iface': 'wlan0open', 'rssi': -55}] + server._iface_ap_info = lambda iface: ('AA:BB:CC:DD:EE:FF', 6) + server._pineap = lambda *a, **k: (0, '', '') + return calls + + def test_kick_validates_and_deny_adds(self): + calls = self._kick_env() server.h_client_kick(type('C', (), {'args': (), 'body': {'mac': '00:11:22:33:44:55'}})()) - self.assertIn(('PINEAPPLE_DEVICE_FILTER_ADD', 'deny', '00:11:22:33:44:55'), calls) - self.assertTrue(any(c[0] == 'PINEAPPLE_DEAUTH_CLIENT' for c in calls)) + 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', + '00:11:22:33:44:55'] and c[4] == '6' for c in calls)) + + def test_kick_not_associated_still_filters(self): + calls = [] + def fake_run(argv, timeout=30): + calls.append(argv) + return 0, '', '' + server.device_run = fake_run + server.assoc_clients = lambda: [] + status, payload = server.h_client_kick(type('C', (), {'args': (), 'body': {'mac': '00:11:22:33:44:55'}})()) + self.assertEqual(status, 200) + self.assertIs(payload['deauth'], False) + self.assertTrue(any(c == [server.HAK5CMD, 'PINEAPPLE_DEVICE_FILTER_ADD', 'deny', + '00:11:22:33:44:55'] for c in calls)) def test_kick_bad_mac_400(self): status, payload = server.h_client_kick(type('C', (), {'args': (), 'body': {'mac': 'x'}})()) @@ -51,12 +77,22 @@ class ClientsTest(unittest.TestCase): def test_deauth_client(self): calls = [] - def fake(*args): - calls.append(args) - return 'ok' - server.hak5 = fake - server.h_deauth_client(type('C', (), {'args': (), 'body': {'mac': '00:11:22:33:44:55'}})()) - self.assertEqual(calls[0][0], 'PINEAPPLE_DEAUTH_CLIENT') + def fake_run(argv, timeout=30): + calls.append(argv) + return 0, '', '' + server.device_run = fake_run + server.assoc_clients = lambda: [{'mac': '00:11:22:33:44:55', 'iface': 'wlan1wpa', 'rssi': -60}] + server._iface_ap_info = lambda iface: ('AA:BB:CC:DD:EE:FF', 149) + 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', + '00:11:22:33:44:55'] and c[4] == '149' for c in calls)) + + def test_deauth_client_not_associated_502(self): + server.assoc_clients = lambda: [] + status, payload = server.h_deauth_client(type('C', (), {'args': (), 'body': {'mac': '00:11:22:33:44:55'}})()) + self.assertEqual(status, 502) if __name__ == '__main__':