Files
Mark-VIII/tests/test_attacks.py
T
bzuccaroandCursor 7d48b7ad06 fix: harden UI actions and daemon calls for reliable control (v1.3.2)
Retry and serialize pineapd/hak5 calls, queue virtual-pager keys, and grey out buttons until the pager finishes. Deploy now installs python3-light after factory firmware. Bump version to 1.3.2.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-20 15:49:10 -05:00

460 lines
20 KiB
Python

import os
import shutil
import sqlite3
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 + '.')), '')
elif a[0] == 'hostapd_cli' and a[-1] == 'status':
return (0, 'state=ENABLED\nssid[0]=test\n', '')
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')
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, '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')
self.old_ent_running = server._ent_running
self.old_ent_state = server._ent_state_loaded
server._ent_running = lambda: True
server._ent_state_loaded = lambda: {'ssid': 'CorpAP', 'channel': 36}
def tearDown(self):
server.PINEAP_STATE_FILE = self.old_state
server._ent_running = self.old_ent_running
server._ent_state_loaded = self.old_ent_state
for k, v in self.old_ent.items():
setattr(server, k, v)
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):
server.h_attacks_deploy(ctx({
'kind': 'wpa', 'ssid': 'TargetNet', 'passphrase': 'secretpass1',
'enctype': 'psk2', 'hidden': False, 'channel': 6}))
cmds = [r[0] for r in self.f.runs]
self.assertIn(['iw', 'dev', 'wlan1ent', 'del'], cmds)
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_wpa_auto_channel_defaults_to_1_without_recon(self):
status, payload = server.h_attacks_deploy(ctx({
'kind': 'wpa', 'ssid': 'UnknownNet', 'passphrase': 'secretpass1',
'enctype': 'psk2', 'hidden': False}))
self.assertEqual(status, 200)
self.assertTrue(payload['auto'])
self.assertEqual(payload['channel'], 1)
self.assertEqual(payload['band'], server.BAND_2G)
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]['channel'], 1)
def test_deploy_wpa_auto_channel_uses_recon_target_channel(self):
db = self._make_recon_db()
old = server.RECON_DB
server.RECON_DB = db
try:
status, payload = server.h_attacks_deploy(ctx({
'kind': 'wpa', 'ssid': 'Anderson-5', 'passphrase': 'secretpass1',
'enctype': 'psk2', 'hidden': False}))
finally:
server.RECON_DB = old
os.unlink(db)
self.assertEqual(status, 200)
self.assertEqual(payload['channel'], 149)
self.assertEqual(payload['band'], server.BAND_5G)
self.assertEqual(self.f.state['wireless.radio1.channel'], '149')
def _make_recon_db(self):
fd, db = tempfile.mkstemp(suffix='.db')
os.close(fd)
conn = sqlite3.connect(db)
conn.executescript(
'CREATE TABLE scan(id INTEGER PRIMARY KEY AUTOINCREMENT, uuid TEXT,'
' time INT, name TEXT);'
'CREATE TABLE wifi_device(hash INT PRIMARY KEY, scan INT, mac TEXT,'
' time INT, signal INT, freq INT, packets INT);'
'CREATE TABLE ssid(hash INT PRIMARY KEY, wifi_device INT, scan INT,'
' type INT, bssid TEXT, ssid BLOB, hidden INT, time INT, signal INT,'
' freq INT, channel INT, encryption INT);')
conn.execute("INSERT INTO scan (uuid, time, name) VALUES ('u1', 1, 'pager')")
conn.execute("INSERT INTO ssid (hash, wifi_device, scan, type, bssid, ssid, hidden,"
" time, signal, freq, channel, encryption) VALUES"
" (10, 1, 1, 8, 'C89E43648080', X'416E646572736F6E2D35', 0,"
" 1786466532, -76, 5745, 149, 0x400400108)")
conn.commit()
conn.close()
return db
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_uses_standalone_phy1_engine(self):
status, payload = server.h_attacks_deploy(ctx({
'kind': 'enterprise', 'ssid': 'CorpAP', 'passphrase': 'anypass',
'enctype': 'wpa2', 'hidden': False, 'channel': 36}))
self.assertEqual(status, 200)
self.assertEqual(payload['iface'], 'wlan1ent')
self.assertEqual(payload['band'], server.BAND_5G)
cmds = [r[0] for r in self.f.runs]
self.assertIn(['iw', 'phy', 'phy1', 'interface', 'add', 'wlan1ent',
'type', 'managed'], cmds)
self.assertIn(['iw', 'dev', 'wlan1ent', 'set', 'type', 'ap'], cmds)
self.assertTrue(any(c[:4] == ['/usr/sbin/hostapd', '-B', '-P', server.ENT_PIDFILE]
for c in cmds))
self.assertEqual(self.f.state['pineapd.@hostapd[0].mgmtiface'], 'wlan1ent')
self.assertEqual(self.f.state['pineapd.wlan1mon.hop'], '0')
with open(server.ENT_CONF) as f:
conf = f.read()
self.assertIn('ca_cert=', conf)
self.assertIn('server_cert=', conf)
self.assertIn('private_key=', conf)
self.assertIn('ieee8021x=1', conf)
self.assertIn('eap_server=1', conf)
self.assertNotIn('eap_server_identity', conf)
self.assertNotIn('eap_server_erp', conf)
self.assertNotIn('dh_file=', conf)
self.assertIn('ieee80211w=0', conf)
with open(server.ENT_EAP_USERS) as f:
users = f.read()
self.assertIn('PEAP,TTLS', users)
self.assertIn('[2]', users)
def test_deploy_enterprise_rejects_non_5g_channel(self):
status, _ = server.h_attacks_deploy(ctx({
'kind': 'enterprise', 'ssid': 'CorpAP', 'enctype': 'wpa2',
'channel': 6}))
self.assertEqual(status, 400)
def test_stop_enterprise_tears_down_engine(self):
server._ent_running = lambda: True
server._ent_state_loaded = lambda: {'ssid': 'CorpAP', 'channel': 36}
status, payload = server.h_attacks_stop(ctx({'kind': 'enterprise'}))
self.assertEqual(status, 200)
self.assertIn('wlan1ent', payload['stopped'])
cmds = [r[0] for r in self.f.runs]
self.assertIn(['iw', 'dev', 'wlan1ent', 'del'], cmds)
def test_deploy_enterprise_writes_passphrase_and_hidden(self):
status, payload = server.h_attacks_deploy(ctx({
'kind': 'enterprise', 'ssid': 'CorpAP', 'passphrase': 'labsecret',
'enctype': 'wpa2', 'hidden': True, 'channel': 36}))
self.assertEqual(status, 200)
with open(server.ENT_EAP_USERS) as f:
users = f.read()
self.assertIn('labsecret', users)
self.assertIn('PEAP,TTLS', users)
self.assertIn('[2]', users)
with open(server.ENT_CONF) as f:
conf = f.read()
self.assertIn('ignore_broadcast_ssid=1', conf)
self.assertIn('ca_cert=', conf)
def test_eap_secret_sanitizes_quotes(self):
self.assertEqual(server._eap_secret(''), 'dummy')
self.assertEqual(server._eap_secret('ab"c\ndef'), 'abcdef')
def test_hostapd_unknown_items_parse_pager_error(self):
err = ("Line 14: unknown configuration item 'eap_server_identity'\n"
"1 errors found in configuration file '/root/loot/enterprise.conf'\n"
"Failed to set up interface with /root/loot/enterprise.conf\n"
"Failed to initialize interface\n")
self.assertEqual(server._hostapd_unknown_items(err), ['eap_server_identity'])
def test_drop_hostapd_keys_removes_only_named_lines(self):
conf = ('interface=wlan1ent\n'
'eap_server=1\n'
'eap_server_identity=hostapd\n'
'dh_file=/tmp/dh.pem\n')
new, changed = server._drop_hostapd_keys(
conf, ['eap_server_identity', 'dh_file'])
self.assertTrue(changed)
self.assertIn('eap_server=1', new)
self.assertIn('interface=wlan1ent', new)
self.assertNotIn('eap_server_identity', new)
self.assertNotIn('dh_file=', new)
def test_start_ent_hostapd_strips_unknown_keys_and_retries(self):
with open(server.ENT_CONF, 'w') as f:
f.write('interface=wlan1ent\neap_server=1\neap_server_identity=hostapd\n')
orig = server.device_run
seen = []
def wrapped(args, timeout=20, input_data=None):
a = list(args)
if a and a[0] == '/usr/sbin/hostapd':
with open(server.ENT_CONF) as fh:
text = fh.read()
seen.append(text)
if 'eap_server_identity' in text:
return (1, '',
"Line 3: unknown configuration item 'eap_server_identity'\n"
"1 errors found in configuration file '%s'\n"
"Failed to initialize interface\n" % server.ENT_CONF)
return (0, '', '')
return orig(args, timeout=timeout, input_data=input_data)
server.device_run = wrapped
try:
rc, _out, _err = server._start_ent_hostapd()
finally:
server.device_run = orig
self.assertEqual(rc, 0)
self.assertGreaterEqual(len(seen), 2)
with open(server.ENT_CONF) as f:
conf = f.read()
self.assertNotIn('eap_server_identity', conf)
self.assertIn('eap_server=1', conf)
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)
class AttacksStatusTest(unittest.TestCase):
def setUp(self):
self.saved = {
'_count_table': server._count_table,
'_ent_summary': server._ent_summary,
'daemon_sock_call': server.daemon_sock_call,
'_uci_ap_summary': server._uci_ap_summary,
'_read_hop': server._read_hop,
}
def tearDown(self):
for name, fn in self.saved.items():
setattr(server, name, fn)
def test_status_exposes_enterprise_ap_for_ui(self):
server._count_table = lambda t: {
'hostap_handshake': 2, 'hostap_basic': 3, 'hostap_chalresp': 1
}.get(t, 0)
server._ent_summary = lambda detail=True: {
'enabled': True, 'live': True, 'ssid': 'CorpLab',
'iface': 'wlan1ent', 'stations': ['AA:BB:CC:DD:EE:FF'],
'auth_method': 'mschapv2', 'certs': True, 'captures': 4,
'ctrl_linked': True,
}
server.daemon_sock_call = lambda *a, **k: (200, {'pineape_disabled': False})
server._uci_ap_summary = lambda *a, **k: None
server._read_hop = lambda: '1'
status, payload = server.h_attacks_status(ctx())
self.assertEqual(status, 200)
self.assertTrue(payload['enterprise']['ap']['live'])
self.assertEqual(payload['enterprise']['ap']['ssid'], 'CorpLab')
self.assertEqual(payload['enterprise']['identities'], 3)
self.assertEqual(payload['enterprise']['mschapv2'], 1)
self.assertEqual(payload['enterprise']['creds'], 4)
self.assertEqual(payload['handshakes'], 2)
self.assertTrue(payload['enterprise']['pineape']['enabled'])
if __name__ == '__main__':
unittest.main()