"""Regression tests for round-2 validation findings (D1-D3). D1 portal download must build the zip with struct+zlib, not zipfile (python3-light has no zipfile; GET /api/portals//download 500s). D2 the deploy-time auto-capture must survive post-deploy radio settle: a revive pass re-arms the capture when it dies during bring-up. D3 GET /api/attacks/capture must be routed to the status handler instead of 404 (UI and scripts poll status). """ import io 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, args=()): return type('C', (), {'body': body if body is not None else {}, 'args': args, 'query': {}})() class BlockZipfile: """Import hook simulating python3-light: importing zipfile raises.""" def find_spec(self, fullname, path=None, target=None): if fullname == 'zipfile': raise ImportError("No module named 'zipfile'") return None def without_zipfile(fn): saved = sys.modules.pop('zipfile', None) blocker = BlockZipfile() sys.meta_path.insert(0, blocker) try: return fn() finally: sys.meta_path.remove(blocker) if saved is not None: sys.modules['zipfile'] = saved class ZipCreateTest(unittest.TestCase): """D1: _zip_create builds archives readable by _zip_entries.""" def test_roundtrip_stored_entries(self): files = {'index.php': b'portal', 'sub/a.ep': b'x' * 40} data = server._zip_create(files) self.assertEqual(server._zip_entries(data), files) def test_output_is_standard_zip(self): data = server._zip_create({'index.php': b'hello'}) with zipfile.ZipFile(io.BytesIO(data)) as zf: self.assertEqual(zf.namelist(), ['index.php']) self.assertEqual(zf.read('index.php'), b'hello') def test_no_zipfile_import_needed(self): def build(): return server._zip_create({'index.php': b'data'}) data = without_zipfile(build) self.assertIn(b'PK\x03\x04', data[:4]) class PortalDownloadWithoutZipfileTest(unittest.TestCase): """D1: h_portal_download must work where zipfile is absent.""" def setUp(self): self.tmp = tempfile.mkdtemp(prefix='mk8-d1-') portal = os.path.join(self.tmp, 'p1') os.makedirs(portal) with open(os.path.join(portal, 'index.php'), 'w') as f: f.write('
') with open(os.path.join(portal, '.logs'), 'w') as f: f.write('[log line]\n') self.old_dir = server.PORTALS_DIR server.PORTALS_DIR = self.tmp def tearDown(self): server.PORTALS_DIR = self.old_dir shutil.rmtree(self.tmp, ignore_errors=True) def test_download_returns_parseable_zip(self): def run(): status, payload = server.h_portal_download(ctx(args=('p1',))) return status, payload result = without_zipfile(run) status, payload = result self.assertEqual(status, 200) entries = server._zip_entries(payload.data) self.assertIn('p1/index.php', entries) self.assertIn(b'
', entries['p1/index.php']) def test_download_missing_portal_404(self): status, payload = server.h_portal_download(ctx(args=('nope',))) self.assertEqual(status, 404) class CaptureReviveTest(unittest.TestCase): """D2: revive pass re-arms a dead auto-capture.""" def setUp(self): self.started = [] self.old_state = server._capture_state self.old_ensure = server._ensure_attack_capture self.old_exists = os.path.exists # Pretend the monitor iface exists (tests run off-device). os.path.exists = lambda p: True def tearDown(self): os.path.exists = self.old_exists server._capture_state = self.old_state server._ensure_attack_capture = self.old_ensure def test_revive_restarts_dead_capture(self): states = iter([(False, 123, True)]) server._capture_state = lambda pf, iface: next(states) server._ensure_attack_capture = \ lambda iface: self.started.append(iface) or {'running': True} server._capture_revive_once('wlan1mon') self.assertEqual(self.started, ['wlan1mon']) def test_revive_skips_running_capture(self): server._capture_state = lambda pf, iface: (True, 5, False) def boom(iface): raise AssertionError('must not restart a running capture') server._ensure_attack_capture = boom server._capture_revive_once('wlan1mon') def test_revive_skips_when_iface_gone(self): server._capture_state = lambda pf, iface: (False, 7, True) os.path.exists = lambda p: 'net/wlan1mon' not in str(p) and \ self.old_exists(p) server._ensure_attack_capture = \ lambda iface: self.started.append(iface) or None server._capture_revive_once('wlan1mon') self.assertEqual(self.started, []) def test_deploy_schedules_revive_for_5g_wpa(self): scheduled = [] old_sched = server._schedule_capture_revive old_ensure = server._ensure_attack_capture server._schedule_capture_revive = lambda iface='wlan1mon': \ scheduled.append(iface) server._ensure_attack_capture = \ lambda iface: {'running': True, 'pid': 1, 'iface': iface} try: import tests.test_attacks as ta harness = ta.AttacksDeployTest('test_deploy_wpa_5g_writes_radio1') harness.setUp() try: status, payload = server.h_attacks_deploy(ctx({ 'kind': 'wpa', 'ssid': 'Corp', 'passphrase': 'secretpass1', 'enctype': 'psk2', 'hidden': False, 'channel': 36})) self.assertEqual(status, 200) self.assertTrue(payload['verified']) self.assertEqual(scheduled, ['wlan1mon']) finally: harness.tearDown() finally: server._schedule_capture_revive = old_sched server._ensure_attack_capture = old_ensure class CaptureGetRouteTest(unittest.TestCase): """D3: GET /api/attacks/capture routes to the capture handler.""" def test_get_route_registered(self): handler, groups = server.ROUTER.dispatch('GET', '/api/attacks/capture') self.assertIsNotNone(handler, 'GET /api/attacks/capture is not routed') self.assertEqual(handler, server.h_attacks_capture) def test_status_without_body_reports_stopped_not_error(self): status, payload = server.h_attacks_capture( ctx(body=None)) self.assertEqual(status, 200) self.assertFalse(payload.get('running')) self.assertNotIn('error', payload) if __name__ == '__main__': unittest.main()