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, args=(), query=None): return type('C', (), {'body': body or {}, 'args': args, 'query': query or {}})() 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() INDEX_PHP = (b"\n" b'') META_EP = json.dumps({'name': 'facebook-login', 'type': 'basic'}).encode() class PortalsTest(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp(prefix='mk8-portals-test-') 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') server._portal_set_active(None) self.hijacks = [] server._portal_dns_hijack = ( lambda enable: self.hijacks.append(enable)) def tearDown(self): server.PORTALS_DIR, server.PORTAL_ACTIVE_FILE, \ server.PORTAL_CAPTURES_FILE = self.old shutil.rmtree(self.tmp, ignore_errors=True) # ---- import ---- def test_import_flat_zip(self): status, payload = server.h_portals_import(ctx({ 'name': 'my-portal', 'data': base64.b64encode(make_zip({ 'index.php': INDEX_PHP, 'assets/style.css': b'body{}'})).decode()})) self.assertEqual(status, 200) self.assertEqual(payload['name'], 'my-portal') root = os.path.join(self.tmp, 'my-portal') self.assertTrue(os.path.isfile(os.path.join(root, 'index.php'))) self.assertTrue(os.path.isfile(os.path.join(root, 'assets', 'style.css'))) def test_import_nested_top_dir_flattens_and_uses_ep_name(self): status, payload = server.h_portals_import(ctx({ 'data': base64.b64encode(make_zip({ 'index.php': INDEX_PHP, 'MyPortal.php': b'', 'facebook-login.ep': META_EP}, top_dir='facebook-login')).decode()})) self.assertEqual(status, 200) self.assertEqual(payload['name'], 'facebook-login') root = os.path.join(self.tmp, 'facebook-login') self.assertTrue(os.path.isfile(os.path.join(root, 'index.php'))) self.assertFalse(os.path.isdir(os.path.join(root, 'facebook-login'))) def test_import_rejects_missing_index_php(self): status, payload = server.h_portals_import(ctx({ 'name': 'bad', 'data': base64.b64encode(make_zip( {'only.css': b'body{}'})).decode()})) self.assertEqual(status, 400) def test_import_rejects_zip_slip(self): evil = make_zip({'index.php': INDEX_PHP}) # Hand-build a zip with an unsafe entry. buf = io.BytesIO() with zipfile.ZipFile(buf, 'w') as zf: zf.writestr('index.php', INDEX_PHP) zf.writestr('../../evil.sh', b'rm -rf /') status, _ = server.h_portals_import(ctx({ 'name': 'evil', 'data': base64.b64encode(buf.getvalue()).decode()})) self.assertEqual(status, 400) self.assertFalse(os.path.exists('/tmp/evil.sh')) self.assertFalse(os.path.exists(evil and '/etc/passwd.mk8test')) def test_import_rejects_garbage(self): status, _ = server.h_portals_import(ctx({ 'name': 'junk', 'data': base64.b64encode(b'not a zip').decode()})) self.assertEqual(status, 400) status, _ = server.h_portals_import(ctx({})) self.assertEqual(status, 400) def test_import_overwrites_same_name(self): data = base64.b64encode(make_zip({ 'index.php': INDEX_PHP})).decode() s1, _ = server.h_portals_import(ctx({'name': 'dup', 'data': data})) data2 = base64.b64encode(make_zip({ 'index.php': INDEX_PHP, 'extra.txt': b'x'})).decode() s2, _ = server.h_portals_import(ctx({'name': 'dup', 'data': data2})) self.assertEqual((s1, s2), (200, 200)) self.assertTrue(os.path.isfile( os.path.join(self.tmp, 'dup', 'extra.txt'))) # ---- list / activate / delete ---- def _import_one(self, name='p1'): status, payload = server.h_portals_import(ctx({ 'name': name, 'data': base64.b64encode(make_zip( {'index.php': INDEX_PHP})).decode()})) assert status == 200, payload return name def test_list_reports_portals_and_active(self): self._import_one('alpha') status, payload = server.h_portals_list(ctx()) self.assertEqual(status, 200) names = [p['name'] for p in payload['portals']] self.assertIn('alpha', names) self.assertIsNone(payload['active']) def test_activate_starts_dns_hijack_and_persists(self): name = self._import_one() status, payload = server.h_portals_activate(ctx(args=(name,))) self.assertEqual(status, 200) self.assertEqual(self.hijacks, [True]) with open(server.PORTAL_ACTIVE_FILE) as f: self.assertEqual(f.read().strip(), name) status, payload = server.h_portals_list(ctx()) self.assertEqual(payload['active'], name) def test_activate_unknown_portal_404(self): status, _ = server.h_portals_activate(ctx(args=('ghost',))) self.assertEqual(status, 404) def test_deactivate_stops_hijack(self): name = self._import_one() server.h_portals_activate(ctx(args=(name,))) status, payload = server.h_portals_deactivate(ctx()) self.assertEqual(status, 200) self.assertEqual(self.hijacks, [True, False]) _, payload = server.h_portals_list(ctx()) self.assertIsNone(payload['active']) def test_delete_active_portal_deactivates_first(self): name = self._import_one() server.h_portals_activate(ctx(args=(name,))) status, _ = server.h_portals_delete(ctx(args=(name,))) self.assertEqual(status, 200) self.assertFalse(os.path.exists(os.path.join(self.tmp, name))) self.assertEqual(self.hijacks, [True, False]) def test_restore_on_boot_reapplies_hijack(self): name = self._import_one() with open(server.PORTAL_ACTIVE_FILE, 'w') as f: f.write(name + '\n') server._portal_restore_on_boot() self.assertEqual(self.hijacks, [True]) self.assertEqual(server._portal_active['name'], name) # ---- php shim ---- def test_php_shim_substitutes_client_values(self): old_leases = server._dhcp_leases server._dhcp_leases = lambda: {'10.0.0.5': ('AA:BB:CC:DD:EE:FF', 'victim-pc')} try: out = server._php_shim(INDEX_PHP.decode(), '10.0.0.5', 'http://login.example.com/') finally: server._dhcp_leases = old_leases self.assertNotIn('