c4ch3c4d3
88d7141d45
- Recon AP focus sidebar: 'Deauth All Clients' with engagement-scope confirm - Deauth Targeting card: 'Deauth All' behind the same scope confirmation - New POST /api/attacks/deauth/bulk (max 32 targets, per-target results) - Evil Portal tab: import EvilPortalNano-format portal zips into /mmc/mk8/portals, serve active portal on port 80 to unauthenticated clients via a minimal PHP shim, capture all form POSTs (.logs in stock MyPortal.php format + captures.jsonl), dnsmasq address=/#/ DNS hijack - OpenAP: Evil Portal template dropdown (greyed when none), activated with the attack and stopped with it - Monitor Capture fix: iface-less status now reports whichever monitor is actually capturing; pcap dir mkdir'd; tcpdump stderr surfaced instead of discarded
254 lines
10 KiB
Python
254 lines
10 KiB
Python
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"<?php\n$destination = 'x';\nrequire_once('helper.php');\n?>\n"
|
|
b'<html><form method="post" action="/captiveportal/index.php">'
|
|
b'<input type="hidden" name="hostname" value="<?=getClientHostName($_SERVER[\'REMOTE_ADDR\']);?>">'
|
|
b'<input type="hidden" name="mac" value="<?=getClientMac($_SERVER[\'REMOTE_ADDR\']);?>">'
|
|
b'<input type="hidden" name="ip" value="<?=$_SERVER[\'REMOTE_ADDR\'];?>">'
|
|
b'<input type="hidden" name="target" value="<?=$destination?>">'
|
|
b'<input name="email"></form></html>')
|
|
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'<?php ?>',
|
|
'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('<?php', out)
|
|
self.assertNotIn('<%=', out)
|
|
self.assertNotIn('<?=', out)
|
|
self.assertIn('value="victim-pc"', out)
|
|
self.assertIn('value="AA:BB:CC:DD:EE:FF"', out)
|
|
self.assertIn('value="10.0.0.5"', out)
|
|
self.assertIn('value="http://login.example.com/"', out)
|
|
|
|
def test_php_shim_escapes_quotes_in_lease_values(self):
|
|
old_leases = server._dhcp_leases
|
|
server._dhcp_leases = lambda: {'10.0.0.5': ('AA:BB:CC:DD:EE:FF',
|
|
'vic"tim')}
|
|
try:
|
|
out = server._php_shim(INDEX_PHP.decode(), '10.0.0.5', 'http://x/')
|
|
finally:
|
|
server._dhcp_leases = old_leases
|
|
self.assertIn('value="vic"tim"', out)
|
|
|
|
# ---- capture ----
|
|
|
|
def test_capture_writes_logs_and_jsonl(self):
|
|
name = self._import_one('credtrap')
|
|
server._portal_capture(b'email=a@b.c&password=hunter2&submit=Log+In',
|
|
'10.0.0.9', name)
|
|
logs_path = os.path.join(self.tmp, 'credtrap', '.logs')
|
|
with open(logs_path) as f:
|
|
text = f.read()
|
|
self.assertIn('email: a@b.c', text)
|
|
self.assertIn('password: hunter2', text)
|
|
self.assertIn('[', text)
|
|
with open(server.PORTAL_CAPTURES_FILE) as f:
|
|
entries = [json.loads(line) for line in f if line.strip()]
|
|
self.assertEqual(len(entries), 1)
|
|
self.assertEqual(entries[0]['fields']['password'], 'hunter2')
|
|
self.assertEqual(entries[0]['ip'], '10.0.0.9')
|
|
self.assertEqual(entries[0]['portal'], 'credtrap')
|
|
|
|
def test_captures_endpoint_lists_newest_first_and_clears(self):
|
|
name = self._import_one()
|
|
server._portal_capture(b'a=1', '10.0.0.1', name)
|
|
server._portal_capture(b'a=2', '10.0.0.2', name)
|
|
status, payload = server.h_portals_captures(ctx(query={'limit': 200}))
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(payload['total'], 2)
|
|
self.assertEqual(payload['captures'][0]['fields']['a'], '2')
|
|
status, _ = server.h_portals_captures_clear(ctx())
|
|
self.assertEqual(status, 200)
|
|
_, payload = server.h_portals_captures(ctx(query={}))
|
|
self.assertEqual(payload['total'], 0)
|
|
|
|
def test_logs_download_returns_file(self):
|
|
name = self._import_one()
|
|
server._portal_capture(b'a=1', '10.0.0.1', name)
|
|
status, payload = server.h_portal_logs(ctx(args=(name,)))
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(payload.filename, '%s.logs.txt' % name)
|
|
self.assertIn(b'a: 1', payload.data)
|
|
|
|
def test_logs_download_404_when_empty(self):
|
|
name = self._import_one()
|
|
status, _ = server.h_portal_logs(ctx(args=(name,)))
|
|
self.assertEqual(status, 404)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|