feat: one-click attack orchestration backend (Evil WPA/Open/Enterprise)
Deploy/stop/status/capture/export-hc22000/deauth endpoints. Band-aware deauth inject (wlan0mon for 2.4GHz), enterprise AP via wlan0ent + PineAPE, verified writes polled from /sys, hop resumed when no radio1 AP active.
This commit is contained in:
@@ -0,0 +1,244 @@
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'payload', 'user', 'remote_access', 'pager-webui'))
|
||||
import server
|
||||
|
||||
|
||||
def setUpModule():
|
||||
__import__('importlib').reload(server)
|
||||
|
||||
|
||||
def ctx(body=None, query=None):
|
||||
return type('C', (), {'body': body, 'args': (), 'query': query or {}})()
|
||||
|
||||
|
||||
class FakeUciDevice:
|
||||
"""In-memory uci + device_run fake: 'uci set wireless.X=Y' state."""
|
||||
|
||||
def __init__(self):
|
||||
self.state = {}
|
||||
self.runs = []
|
||||
self.sock = []
|
||||
self._verify = True
|
||||
|
||||
def device_run(self, args, timeout=20, input_data=None):
|
||||
self.runs.append((list(args), input_data))
|
||||
a = list(args)
|
||||
if a[:2] == ['uci', 'set']:
|
||||
k, _, v = a[2].partition('=')
|
||||
self.state[k] = v
|
||||
elif a[:2] == ['uci', 'get']:
|
||||
return (0, self.state.get(a[2], '') + '\n', '')
|
||||
elif a[:2] == ['uci', 'delete']:
|
||||
for k in list(self.state):
|
||||
if k == a[2] or k.startswith(a[2] + '.'):
|
||||
del self.state[k]
|
||||
elif a[:2] == ['uci', 'commit']:
|
||||
pass
|
||||
elif a[0] == 'uci' and a[1] == 'show':
|
||||
sec = a[2]
|
||||
return (0, ''.join("%s=%s\n" % (k, v) for k, v in self.state.items()
|
||||
if k == sec or k.startswith(sec + '.')), '')
|
||||
return (0, '', '')
|
||||
|
||||
def uci_iface(self, name):
|
||||
cfg = {}
|
||||
prefix = 'wireless.%s.' % name
|
||||
for k, v in self.state.items():
|
||||
if k.startswith(prefix):
|
||||
cfg[k[len(prefix):]] = v
|
||||
if not cfg:
|
||||
return {}
|
||||
return cfg
|
||||
|
||||
def daemon_sock_call(self, method, path, body=None, timeout=10):
|
||||
self.sock.append((method, path, body))
|
||||
if path == '/api/pineap/hostapd/get_config':
|
||||
return 200, {'pineape_disabled': False, 'pineape_auth_pass': True}
|
||||
if path == '/api/pineap/get_config':
|
||||
return 200, {'autossidpool': True}
|
||||
return 200, {'success': True}
|
||||
|
||||
|
||||
class AttacksDeployTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.f = FakeUciDevice()
|
||||
server.device_run = self.f.device_run
|
||||
server.daemon_sock_call = self.f.daemon_sock_call
|
||||
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
|
||||
server._allow_all_ssids = lambda: True
|
||||
self.tmp = tempfile.mkdtemp(prefix='pager-attacks-')
|
||||
self.old_state = server.PINEAP_STATE_FILE
|
||||
server.PINEAP_STATE_FILE = os.path.join(self.tmp, 'state.json')
|
||||
|
||||
def tearDown(self):
|
||||
server.PINEAP_STATE_FILE = self.old_state
|
||||
shutil.rmtree(self.tmp)
|
||||
|
||||
def test_deploy_wpa_2g4_calls_daemon_and_enables_engine(self):
|
||||
status, payload = server.h_attacks_deploy(ctx({
|
||||
'kind': 'wpa', 'ssid': 'TargetNet', 'passphrase': 'secretpass1',
|
||||
'enctype': 'psk2', 'hidden': False, 'channel': 6}))
|
||||
self.assertEqual(status, 200)
|
||||
self.assertTrue(payload['ok'])
|
||||
self.assertTrue(payload['verified'])
|
||||
cfg = [s for s in self.f.sock if s[0] == 'PUT' and s[1] == '/api/settings/wifi/set_ap'][0][2]
|
||||
self.assertEqual(cfg['configs'][0]['interface'], 'wlan0wpa')
|
||||
self.assertEqual(cfg['configs'][0]['ssid'], 'TargetNet')
|
||||
self.assertEqual(cfg['configs'][0]['key'], 'secretpass1')
|
||||
self.assertEqual(cfg['configs'][0]['enctype'], 'psk2')
|
||||
engines = [s for s in self.f.sock
|
||||
if s[1] in ('/api/pineap/hostapd/enable_pineap',
|
||||
'/api/pineap/mimic/enable')]
|
||||
self.assertEqual(len(engines), 2)
|
||||
|
||||
def test_deploy_wpa_2g4_stops_enterprise_ap(self):
|
||||
self.f.state['wireless.wlan0ent.disabled'] = '0'
|
||||
server.h_attacks_deploy(ctx({
|
||||
'kind': 'wpa', 'ssid': 'TargetNet', 'passphrase': 'secretpass1',
|
||||
'enctype': 'psk2', 'hidden': False, 'channel': 6}))
|
||||
self.assertNotIn('wireless.wlan0ent.disabled', self.f.state)
|
||||
|
||||
def test_deploy_wpa_5g_writes_radio1(self):
|
||||
status, payload = server.h_attacks_deploy(ctx({
|
||||
'kind': 'wpa', 'ssid': 'Corp', 'passphrase': 'secretpass1',
|
||||
'enctype': 'psk2', 'hidden': False, 'channel': 36}))
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(self.f.state['wireless.wlan1wpa.ssid'], 'Corp')
|
||||
self.assertEqual(self.f.state['wireless.wlan1wpa.encryption'], 'psk2')
|
||||
self.assertEqual(self.f.state['wireless.radio1.channel'], '36')
|
||||
self.assertEqual(self.f.state['pineapd.wlan1mon.hop'], '0')
|
||||
self.assertEqual(payload['iface'], 'wlan1wpa')
|
||||
self.assertEqual(payload['band'], server.BAND_5G)
|
||||
|
||||
def test_deploy_open_2g4_includes_bssid_and_country(self):
|
||||
status, payload = server.h_attacks_deploy(ctx({
|
||||
'kind': 'open', 'ssid': 'Guest', 'hidden': False,
|
||||
'channel': 1, 'country': 'US',
|
||||
'bssid': 'DE:AD:BE:EF:00:01'}))
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(payload['iface'], 'wlan0open')
|
||||
cfg = [s for s in self.f.sock if s[1] == '/api/settings/wifi/set_ap'][0][2]
|
||||
self.assertEqual(cfg['configs'][0]['bssid'], 'DE:AD:BE:EF:00:01')
|
||||
|
||||
def test_deploy_enterprise_builds_wlan0ent(self):
|
||||
status, payload = server.h_attacks_deploy(ctx({
|
||||
'kind': 'enterprise', 'ssid': 'CorpAP', 'passphrase': 'anypass',
|
||||
'enctype': 'wpa2', 'hidden': False, 'channel': 1}))
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(payload['iface'], 'wlan0ent')
|
||||
self.assertEqual(self.f.state['wireless.wlan0ent.encryption'], 'wpa2')
|
||||
self.assertEqual(self.f.state['wireless.wlan0ent.ssid'], 'CorpAP')
|
||||
cfg = [s for s in self.f.sock if s[1] == '/api/settings/wifi/set_ap'][0][2]
|
||||
self.assertEqual(cfg['configs'][0]['interface'], 'wlan0ent')
|
||||
self.assertEqual(cfg['configs'][0]['enctype'], 'wpa2')
|
||||
pineape = [s for s in self.f.sock if s[1] == '/api/pineap/hostapd/set_config'][0]
|
||||
self.assertEqual(pineape[2], {'pineape_disabled': False, 'pineape_auth_pass': True})
|
||||
|
||||
def test_deploy_validation(self):
|
||||
status, _ = server.h_attacks_deploy(ctx({'kind': 'wpa', 'ssid': ''}))
|
||||
self.assertEqual(status, 400)
|
||||
status, _ = server.h_attacks_deploy(ctx({
|
||||
'kind': 'wpa', 'ssid': 'X', 'passphrase': 'short',
|
||||
'enctype': 'psk2', 'channel': 1}))
|
||||
self.assertEqual(status, 400)
|
||||
status, _ = server.h_attacks_deploy(ctx({
|
||||
'kind': 'wpa', 'ssid': 'X', 'passphrase': 'secretpass1',
|
||||
'enctype': 'psk2', 'channel': 200}))
|
||||
self.assertEqual(status, 400)
|
||||
status, _ = server.h_attacks_deploy(ctx({'kind': 'bogus'}))
|
||||
self.assertEqual(status, 400)
|
||||
|
||||
def test_stop_wpa_disables_both_bands(self):
|
||||
self.f.state['wireless.wlan0wpa.disabled'] = '0'
|
||||
self.f.state['wireless.wlan1wpa.disabled'] = '0'
|
||||
self.f.state['pineapd.wlan1mon.hop'] = '0'
|
||||
status, payload = server.h_attacks_stop(ctx({'kind': 'wpa'}))
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(self.f.state['wireless.wlan0wpa.disabled'], '1')
|
||||
self.assertEqual(self.f.state['wireless.wlan1wpa.disabled'], '1')
|
||||
self.assertIn('wlan0wpa', payload['stopped'])
|
||||
self.assertEqual(self.f.state['pineapd.wlan1mon.hop'], '1')
|
||||
|
||||
|
||||
class AttacksDeauthTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.f = FakeUciDevice()
|
||||
self.f._verify = True
|
||||
server.device_run = self.f.device_run
|
||||
server.daemon_sock_call = self.f.daemon_sock_call
|
||||
server._uci_wifi_iface = self.f.uci_iface
|
||||
|
||||
def test_deauth_2g4_uses_wlan0mon_inject(self):
|
||||
status, payload = server.h_attacks_deauth(ctx({
|
||||
'bssid': 'AA:BB:CC:DD:EE:FF', 'client': '11:22:33:44:55:66',
|
||||
'channel': 6}))
|
||||
self.assertEqual(status, 200)
|
||||
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',
|
||||
'11:22:33:44:55:66', '6'], calls)
|
||||
|
||||
def test_deauth_5g_keeps_wlan1mon_inject(self):
|
||||
status, payload = server.h_attacks_deauth(ctx({
|
||||
'bssid': 'AA:BB:CC:DD:EE:FF', 'client': '11:22:33:44:55:66',
|
||||
'channel': 36}))
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(payload['inject'], 'wlan1mon')
|
||||
|
||||
def test_deauth_bad_macs_rejected(self):
|
||||
status, _ = server.h_attacks_deauth(ctx({
|
||||
'bssid': 'nope', 'client': '11:22:33:44:55:66', 'channel': 6}))
|
||||
self.assertEqual(status, 400)
|
||||
|
||||
|
||||
class AttacksExportTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.f = FakeUciDevice()
|
||||
self.f._verify = True
|
||||
server.device_run = self.f.device_run
|
||||
|
||||
def fake_run(args, timeout=20, input_data=None):
|
||||
self.f.runs.append((list(args), input_data))
|
||||
a = list(args)
|
||||
if a[0] == 'ls':
|
||||
return (0, 'a.pcap\nb.cap\n', '')
|
||||
if a[0] == 'hcxpcapngtool':
|
||||
return (0, '', '')
|
||||
return (0, '', '')
|
||||
|
||||
server.device_run = fake_run
|
||||
server.daemon_sock_call = lambda method, path, body=None, timeout=10: (200, {})
|
||||
self._real_exists = os.path.exists
|
||||
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
|
||||
try:
|
||||
os.unlink('/tmp/mk8test.hc22000')
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
def test_export_converts_captures(self):
|
||||
status, payload = server.h_attacks_export_hc22000(ctx())
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(payload['size'], 12)
|
||||
self.assertIn('hashcat -m 22000', payload['hashcat'])
|
||||
hc = [r[0] for r in self.f.runs if r[0][0] == 'hcxpcapngtool'][0]
|
||||
self.assertEqual(hc[1], '-o')
|
||||
self.assertTrue(hc[2].startswith('/root/loot/hc22000/'))
|
||||
self.assertTrue(hc[2].endswith('.hc22000'))
|
||||
self.assertIn('/root/loot/handshakes/a.pcap', hc)
|
||||
self.assertIn('/root/loot/pcap/b.cap', hc)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user