"""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)