c4ch3c4d3
d23ea56364
- portals: replace zipfile with struct+zlib ZIP reader (python3-light has no urllib; import endpoint was dead on device) - dns hijack: uci add_list/del_list for dhcp.@dnsmasq[0].address (list option; plain set was silently dropped from generated dnsmasq config) - radio1: bridge attack APs into br-lan via network.brlan.ports so victims get DHCP/portal reach; wlan1ent runtime-bridged after hostapd verify (retry loop may recreate the iface) - capture: auto-start pinned wlan1mon pcap on 5GHz WPA deploy, teardown on stop; loot flows via hc22000 export (crack-verified end-to-end) - enterprise: pineapd restart after ctrl link + re-assert PineAPE toggles Documented residual: pineapd refuses forwarding from foreign hostapd instances (broken pipe), and daemon set_ap rejects radio1 names - so hostap_handshake rows for radio1 twins and enterprise cred tables cannot populate without a Hak5 firmware change. New tests/test_validation_fixes.py covers each fix (TDD); full suite (29 modules) green.
360 lines
15 KiB
Python
360 lines
15 KiB
Python
"""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'<html><form method="post"><input name="email"></form></html>'
|
|
|
|
|
|
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)
|