bzuccaro/Mark-VIII
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

release: Mark VIII 1.0

Browse Source
This commit is contained in:
c4ch3c4d3
2026-08-11 20:24:24 -07:00
commit 3e1805dab8
86 changed files with 20678 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/__init__.py
View File
tests/test_auth.py
+124
View File
@@ -0,0 +1,124 @@
import json
import os
import stat
import sys
import tempfile
import unittest
from unittest import mock
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)
class SessionTest(unittest.TestCase):
def setUp(self):
fd, self.path = tempfile.mkstemp()
os.close(fd)
server.SESSION_FILE = self.path
def tearDown(self):
try:
os.unlink(self.path)
except OSError:
pass
def test_save_and_load_roundtrip(self):
server.save_session({'serverid': '001337AEE050', 'token': 'abc123'})
self.assertEqual(server.load_session(), {'serverid': '001337AEE050', 'token': 'abc123'})
if os.name != 'nt':
self.assertEqual(stat.S_IMODE(os.stat(self.path).st_mode), 0o600)
def test_load_missing_file_returns_empty(self):
os.unlink(self.path)
self.assertEqual(server.load_session(), {})
def test_check_auth_matches_cookie(self):
server.save_session({'serverid': '001337AEE050', 'token': 'abc123'})
self.assertTrue(server.check_auth('foo=bar; AUTH_001337AEE050=abc123'))
self.assertFalse(server.check_auth('AUTH_001337AEE050=wrong'))
self.assertFalse(server.check_auth(''))
def test_check_auth_rejects_prefix_value(self):
server.save_session({'serverid': '001337AEE050', 'token': 'abc123'})
self.assertFalse(server.check_auth('AUTH_001337AEE050=abc123xxxx'))
self.assertFalse(server.check_auth('prefix AUTH_001337AEE050=abc123'))
server.SESSION_FILE = os.path.join(tempfile.gettempdir(), 'pagerwebui_test_session.json')
if os.path.exists(server.SESSION_FILE):
os.unlink(server.SESSION_FILE)
self.assertFalse(server.check_auth('AUTH_=anything'))
def test_current_token(self):
server.save_session({'serverid': 'S', 'token': 'tok'})
self.assertEqual(server.current_token(), 'tok')
def test_logout_removes_session_and_expires_cookie(self):
server.save_session({'serverid': 'S', 'token': 'tok'})
class H:
def __init__(self):
self.headers = []
def add_extra_header(self, name, value):
self.headers.append((name, value))
ctx = type('C', (), {'h': H()})()
status, payload = server.h_logout(ctx)
self.assertEqual(status, 200)
self.assertTrue(payload['ok'])
self.assertEqual(server.load_session(), {})
self.assertIn(('Set-Cookie',
'AUTH_S=; Path=/; HttpOnly; SameSite=Strict; Max-Age=0'),
ctx.h.headers)
class LoginTest(unittest.TestCase):
def test_login_success_sets_cookie_payload(self):
def fake_daemon(method, path, body=None, token=None, timeout=15):
if path == '/api/login':
return 200, {'token': 'tok123'}
if path == '/api/api_ping':
return 200, {'serverid': '001337AEE050', 'version': '1.0'}
return 404, None
server.daemon_call = fake_daemon
server.SESSION_FILE = os.path.join(tempfile.gettempdir(), 'pagerwebui_test_session.json')
if os.path.exists(server.SESSION_FILE):
os.unlink(server.SESSION_FILE)
class H:
def __init__(self):
self.headers = []
def add_extra_header(self, name, value):
self.headers.append((name, value))
class Ctx:
body = {'username': 'root', 'password': 'pw'}
cookie = ''
h = H()
ctx = Ctx()
status, payload = server.h_login(ctx)
self.assertEqual(status, 200)
self.assertNotIn('token', payload)
self.assertTrue(payload['ok'])
self.assertEqual(payload['serverid'], '001337AEE050')
self.assertEqual(server.load_session()['token'], 'tok123')
self.assertTrue(any(n == 'Set-Cookie' and v.startswith('AUTH_001337AEE050=tok123') for n, v in ctx.h.headers))
def test_login_failure_returns_401(self):
def fake_daemon(method, path, body=None, token=None, timeout=15):
return 401, {'error': 'bad'}
server.daemon_call = fake_daemon
class Ctx:
body = {'username': 'root', 'password': 'nope'}
cookie = ''
status, payload = server.h_login(Ctx())
self.assertEqual(status, 401)
def test_api_ping(self):
server.daemon_call = lambda *a, **k: (200, {'serverid': 'S1', 'version': '9'})
server.SESSION_FILE = os.path.join(tempfile.gettempdir(), 'pagerwebui_test_session.json')
status, payload = server.h_api_ping(type('C', (), {'cookie': '', 'args': ()})())
self.assertEqual(status, 200)
self.assertEqual(payload['serverid'], 'S1')
if __name__ == '__main__':
unittest.main()
tests/test_core.py
+71
View File
@@ -0,0 +1,71 @@
import os
import sys
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)
class FakeHandler:
def __init__(self, path='/', headers=None, body=None):
self.path = path
self.headers = headers or {}
self._body = body
def read_body(self):
return self._body if self._body is not None else {}
class RouterTest(unittest.TestCase):
def test_dispatch_matches_and_captures(self):
server.ROUTER = server.Router()
server.ROUTER.add('GET', r'/api/recon/scans/(\d+)', lambda ctx: (200, {'id': ctx.args[0]}))
handler, groups = server.ROUTER.dispatch('GET', '/api/recon/scans/42')
self.assertIsNotNone(handler)
self.assertEqual(groups, ('42',))
def test_dispatch_returns_none_on_mismatch(self):
server.ROUTER = server.Router()
handler, groups = server.ROUTER.dispatch('POST', '/nope')
self.assertIsNone(handler)
class StaticTest(unittest.TestCase):
def test_safe_join_rejects_traversal(self):
self.assertFalse(server._safe_join(server.WWW_DIR, '../server.py'))
def test_safe_join_accepts_subpath(self):
p = server._safe_join(server.WWW_DIR, 'js/app.js')
self.assertTrue(p.startswith(server.WWW_DIR))
class OriginTest(unittest.TestCase):
def test_same_origin_accepts_matching_origin(self):
self.assertTrue(server.same_origin({
'host': '172.16.52.1:8080',
'origin': 'http://172.16.52.1:8080'}))
def test_same_origin_rejects_cross_origin(self):
self.assertFalse(server.same_origin({
'host': '172.16.52.1:8080',
'origin': 'http://attacker.example'}))
def test_same_origin_allows_non_browser_client(self):
self.assertTrue(server.same_origin({'host': '172.16.52.1:8080'}))
def test_same_origin_requires_host(self):
self.assertFalse(server.same_origin({}))
class DeviceRunTest(unittest.TestCase):
def test_device_run_returns_triple(self):
rc, out, err = server.device_run(['cmd', 'not', 'there'])
self.assertIsInstance(rc, int)
if __name__ == '__main__':
unittest.main()
tests/test_loot.py
+59
View File
@@ -0,0 +1,59 @@
import os
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)
class HandshakesTest(unittest.TestCase):
def setUp(self):
self.dir = tempfile.mkdtemp()
server.LOOT_HS_DIR = self.dir
with open(os.path.join(self.dir, 'hs1.cap'), 'wb') as f:
f.write(b'\x00\x01\x02')
with open(os.path.join(self.dir, 'hs2.pcap'), 'wb') as f:
f.write(b'\x00' * 5)
def test_list(self):
status, payload = server.h_handshakes_get(type('C', (), {'args': ()})())
self.assertEqual(status, 200)
names = [f['name'] for f in payload['files']]
self.assertEqual(names, ['hs1.cap', 'hs2.pcap'])
self.assertEqual(payload['files'][0]['size'], 3)
def test_delete_by_body(self):
server.h_handshakes_delete(type('C', (), {'args': (), 'body': {'name': 'hs1.cap'}})())
self.assertFalse(os.path.exists(os.path.join(self.dir, 'hs1.cap')))
def test_delete_rejects_traversal(self):
status, payload = server.h_handshakes_delete(type('C', (), {'args': (), 'body': {'name': '../server.py'}})())
self.assertEqual(status, 400)
class LootProxyTest(unittest.TestCase):
def test_loot_zip_returns_download(self):
server.daemon_call = lambda m, p, body=None, token=None, timeout=15: (200, b'PK\x03\x04zipdata')
server.current_token = lambda: 'tok'
status, payload = server.h_loot_zip(type('C', (), {'args': ()})())
self.assertEqual(status, 200)
self.assertEqual(payload.ctype, 'application/zip')
def test_loot_archive_posts_to_daemon(self):
calls = []
def fake(m, p, body=None, token=None, timeout=15):
calls.append((m, p))
return 200, {'ok': True}
server.daemon_call = fake
server.current_token = lambda: 'tok'
server.h_loot_archive(type('C', (), {'args': ()})())
self.assertTrue(any(m == 'POST' and p == '/api/loot/archive' for m, p in calls))
if __name__ == '__main__':
unittest.main()
tests/test_misc.py
+279
View File
@@ -0,0 +1,279 @@
import os
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)
class PayloadsProxyTest(unittest.TestCase):
def test_index_proxies(self):
server.daemon_call = lambda m, p, body=None, token=None, timeout=15: (200, {'payloads': []})
server.current_token = lambda: 'tok'
class Ctx:
args = ()
status, payload = server.h_payloads_index(Ctx())
self.assertEqual(status, 200)
self.assertEqual(payload, {'payloads': []})
def test_install_uses_key(self):
calls = []
def fake(m, p, body=None, token=None, timeout=15):
calls.append((m, p))
return 200, {'ok': True}
server.daemon_call = fake
server.current_token = lambda: 'tok'
server.h_payloads_install(type('C', (), {'args': (), 'body': {'key': 'nautilus'}})())
self.assertTrue(any(m == 'POST' and '/api/payloads/portal/nautilus/install' in p for m, p in calls))
def test_remove_uses_key(self):
calls = []
def fake(m, p, body=None, token=None, timeout=15):
calls.append((m, p))
return 200, {'ok': True}
server.daemon_call = fake
server.current_token = lambda: 'tok'
server.h_payloads_remove(type('C', (), {'args': (), 'body': {'key': 'nautilus'}})())
self.assertTrue(any(m == 'POST' and '/api/payloads/portal/nautilus/remove' in p for m, p in calls))
def test_installed_inventory_flattens_firmware_records(self):
old = server._payload_daemon
server._payload_daemon = lambda method, path, body=None: (200, [{
'installed': {'key': 'user~general~demo', 'title': 'Demo',
'path': '/root/payloads/user/general/demo',
'category': 'general', 'launchpoint': 'payload.sh'},
'missingmanifest': False, 'update': {'version': '2.0'}}])
try:
status, payload = server.h_payloads_installed(type('C', (), {'args': ()})())
finally:
server._payload_daemon = old
self.assertEqual(status, 200)
self.assertEqual(payload['payloads'][0]['title'], 'Demo')
self.assertEqual(payload['payloads'][0]['update']['version'], '2.0')
def test_install_rejects_unsafe_key(self):
status, payload = server.h_payloads_install(type('C', (), {
'args': (), 'body': {'key': '../payload'}})())
self.assertEqual(status, 400)
self.assertIn('key', payload['error'])
def test_run_launches_only_resolved_installed_payload(self):
old_record = server._payload_record
old_safe = server._payload_safe_launch
old_popen = server.subprocess.Popen
old_run_dir = server.PAYLOAD_RUN_DIR
class FakeProcess:
pid = 4321
def poll(self):
return None
with tempfile.TemporaryDirectory() as tmp:
launch = os.path.join(tmp, 'payload.sh')
with open(launch, 'w') as handle:
handle.write('#!/bin/sh\n')
server._payload_record = lambda key: (200, {
'key': key, 'title': 'Demo', 'path': tmp,
'launchpoint': 'payload.sh', 'interpreter': 'sh', 'disabled': False})
server._payload_safe_launch = lambda item: (tmp, launch)
calls = []
server.subprocess.Popen = lambda command, **kwargs: calls.append((command, kwargs)) or FakeProcess()
server.PAYLOAD_RUN_DIR = os.path.join(tmp, 'runs')
server._payload_runs.clear()
try:
status, payload = server.h_payloads_run(type('C', (), {
'body': {'key': 'user~general~demo'}})())
finally:
server._payload_record = old_record
server._payload_safe_launch = old_safe
server.subprocess.Popen = old_popen
server.PAYLOAD_RUN_DIR = old_run_dir
server._payload_runs.clear()
self.assertEqual(status, 200)
self.assertEqual(calls[0][0], ['sh', launch])
self.assertTrue(calls[0][1]['start_new_session'])
self.assertTrue(payload['run']['running'])
def test_stop_rejects_untracked_process(self):
server._payload_runs.clear()
status, payload = server.h_payloads_stop(type('C', (), {
'body': {'id': 'not-tracked'}})())
self.assertEqual(status, 404)
def test_remove_rejects_active_webui(self):
status, payload = server.h_payloads_remove(type('C', (), {
'args': (), 'body': {'key': server.SELF_PAYLOAD_KEY}})())
self.assertEqual(status, 409)
self.assertIn('cannot remove itself', payload['error'])
class LoggingTest(unittest.TestCase):
def test_system_tails(self):
server.device_run = lambda args, timeout=20: (0, '\n'.join('line%d' % i for i in range(20)), '')
status, payload = server.h_logging_system(type('C', (), {'args': (), 'query': {'lines': '5'}})())
self.assertEqual(status, 200)
self.assertEqual(len(payload['lines']), 5)
def test_invalid_line_count_uses_default(self):
server.device_run = lambda args, timeout=20: (
0, '\n'.join('line%d' % i for i in range(250)), '')
status, payload = server.h_logging_system(type('C', (), {
'args': (), 'query': {'lines': 'invalid'}})())
self.assertEqual(status, 200)
self.assertEqual(len(payload['lines']), 200)
def test_line_count_is_bounded(self):
self.assertEqual(server._line_count(
type('C', (), {'query': {'lines': '999999'}})(), 200), 2000)
self.assertEqual(server._line_count(
type('C', (), {'query': {'lines': '-10'}})(), 200), 0)
class SettingsTest(unittest.TestCase):
def test_hostname_get(self):
class H:
command = 'GET'
server.device_run = lambda args, timeout=20: (0, 'pager\n', '')
status, payload = server.h_settings_hostname(type('C', (), {'args': (), 'h': H()})())
self.assertEqual(status, 200)
self.assertEqual(payload['hostname'], 'pager')
def test_internet_status_uses_bounded_connectivity_check(self):
calls = []
old = server.device_run
server.device_run = lambda args, timeout=20: (
calls.append((args, timeout)) or (0, '', ''))
try:
status, payload = server.h_settings_internet(type('C', (), {})())
finally:
server.device_run = old
self.assertEqual(status, 200)
self.assertTrue(payload['online'])
self.assertEqual(calls[0][0][-1], '1.1.1.1')
self.assertEqual(calls[0][1], 5)
def test_reboot_is_scheduled_after_response(self):
scheduled = []
class FakeTimer:
def __init__(self, delay, fn):
scheduled.append((delay, fn))
self.daemon = False
def start(self):
scheduled.append('started')
old = server.threading.Timer
server.threading.Timer = FakeTimer
try:
status, payload = server.h_settings_reboot(type('C', (), {})())
finally:
server.threading.Timer = old
self.assertEqual(status, 200)
self.assertTrue(payload['ok'])
self.assertEqual(scheduled[0][0], 1.0)
self.assertEqual(scheduled[1], 'started')
def test_password_uses_passwd_stdin_and_invalidates_session(self):
calls = []
def fake(args, timeout=20, input_data=None):
calls.append((args, timeout, input_data))
return 0, '', ''
old_run = server.device_run
old_session = server.SESSION_FILE
fd, session = tempfile.mkstemp()
os.close(fd)
server.device_run = fake
server.SESSION_FILE = session
try:
status, payload = server.h_settings_password(
type('C', (), {'args': (), 'body': {'password': 'newpw'}})())
finally:
server.device_run = old_run
server.SESSION_FILE = old_session
self.assertEqual((status, payload), (200, {'ok': True}))
self.assertEqual(calls, [(['/bin/passwd', 'root'], 15, b'newpw\nnewpw\n')])
self.assertFalse(os.path.exists(session))
def test_password_reports_passwd_failure(self):
old_run = server.device_run
server.device_run = lambda args, timeout=20, input_data=None: (1, '', 'bad password')
try:
status, payload = server.h_settings_password(
type('C', (), {'args': (), 'body': {'password': 'newpw'}})())
finally:
server.device_run = old_run
self.assertEqual(status, 500)
self.assertEqual(payload['error'], 'bad password')
def test_ntp_get_parses(self):
class H:
command = 'GET'
server.device_run = lambda args, timeout=20: (0, "system.ntp.enabled='1'\nsystem.ntp.server='0.pool.ntp.org'\n", '')
status, payload = server.h_settings_ntp(type('C', (), {'args': (), 'h': H()})())
self.assertEqual(status, 200)
self.assertEqual(payload['enabled'], True)
self.assertIn('0.pool.ntp.org', payload['servers'])
def test_ntp_get_parses_multiple_values_on_one_uci_line(self):
class H:
command = 'GET'
server.device_run = lambda args, timeout=20: (
0,
"system.ntp.enabled='1'\n"
"system.ntp.server='0.openwrt.pool.ntp.org' '1.openwrt.pool.ntp.org'\n",
'')
status, payload = server.h_settings_ntp(type('C', (), {'args': (), 'h': H()})())
self.assertEqual(status, 200)
self.assertEqual(payload['servers'], [
'0.openwrt.pool.ntp.org', '1.openwrt.pool.ntp.org'])
def test_service_running_detection(self):
server.device_run = lambda args, timeout=20: (0, '', '')
status, payload = server.h_settings_service(type('C', (), {'args': ()})())
self.assertEqual(status, 200)
self.assertIn('running', payload)
self.assertIn('background', payload)
def test_resources_parses_mark_style_filesystem_rows(self):
sample = ("Filesystem Type 1024-blocks Used Available Capacity Mounted on\n"
"/dev/root squashfs 30720 30720 0 100% /rom\n"
"/dev/mmcblk0p1 ext4 3724208 21876 3492668 1% /mmc\n")
rows = server._parse_df_rows(sample)
self.assertEqual(len(rows), 2)
self.assertEqual(rows[1]['format'], 'ext4')
self.assertEqual(rows[1]['mount'], '/mmc')
self.assertEqual(rows[1]['size'], 3724208 * 1024)
def test_usb_parser_returns_structured_devices(self):
server.device_run = lambda args, timeout=20: (
0, 'Bus 001 Device 004: ID 0e8d:7961 MediaTek Inc. Wireless_Device\n', '')
status, payload = server.h_settings_usb(type('C', (), {'args': ()})())
self.assertEqual(status, 200)
self.assertEqual(payload['devices'][0]['id'], '0e8d:7961')
self.assertIn('MediaTek', payload['devices'][0]['name'])
def test_password_rejects_mismatched_confirmation(self):
status, payload = server.h_settings_password(type('C', (), {'body': {
'new_password': 'newpass', 'repeat_password': 'different'}})())
self.assertEqual(status, 400)
self.assertIn('match', payload['error'])
def test_management_wifi_get_hides_existing_password(self):
old = server._uci_wifi_iface
server._uci_wifi_iface = lambda name: {
'ssid': 'Pager Mgmt', 'key': 'secretpass', 'disabled': '0', 'hidden': '1'}
try:
status, payload = server.h_settings_management_wifi(
type('C', (), {'body': {}, 'args': ()})())
finally:
server._uci_wifi_iface = old
self.assertEqual(status, 200)
self.assertTrue(payload['has_password'])
self.assertNotIn('password', payload)
if __name__ == '__main__':
unittest.main()
tests/test_pineap_clients.py
+63
View File
@@ -0,0 +1,63 @@
import os
import sys
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)
_orig_hak5 = server.hak5
_orig_assoc_clients = server.assoc_clients
class NormalizeTest(unittest.TestCase):
def test_normalize(self):
self.assertEqual(server.normalize_mac(' 00:11:22:33:44:55 '), '00:11:22:33:44:55')
self.assertEqual(server.normalize_mac('aa:bb:cc:dd:ee:ff'), 'AA:BB:CC:DD:EE:FF')
def test_invalid_returns_none(self):
self.assertIsNone(server.normalize_mac('nope'))
class ClientsTest(unittest.TestCase):
def tearDown(self):
server.hak5 = _orig_hak5
server.assoc_clients = _orig_assoc_clients
def test_clients_handler(self):
server.assoc_clients = lambda: [{'mac': 'AA:BB:CC:DD:EE:FF', 'iface': 'wlan0open', 'rssi': -55}]
class Ctx:
args = ()
status, payload = server.h_clients(Ctx())
self.assertEqual(status, 200)
self.assertEqual(payload['count'], 1)
def test_kick_validates_and_deny_adds(self):
calls = []
def fake(*args):
calls.append(args)
return 'ok'
server.hak5 = fake
server.h_client_kick(type('C', (), {'args': (), 'body': {'mac': '00:11:22:33:44:55'}})())
self.assertIn(('PINEAPPLE_DEVICE_FILTER_ADD', 'deny', '00:11:22:33:44:55'), calls)
self.assertTrue(any(c[0] == 'PINEAPPLE_DEAUTH_CLIENT' for c in calls))
def test_kick_bad_mac_400(self):
status, payload = server.h_client_kick(type('C', (), {'args': (), 'body': {'mac': 'x'}})())
self.assertEqual(status, 400)
def test_deauth_client(self):
calls = []
def fake(*args):
calls.append(args)
return 'ok'
server.hak5 = fake
server.h_deauth_client(type('C', (), {'args': (), 'body': {'mac': '00:11:22:33:44:55'}})())
self.assertEqual(calls[0][0], 'PINEAPPLE_DEAUTH_CLIENT')
if __name__ == '__main__':
unittest.main()
tests/test_pineap_enterprise.py
+43
View File
@@ -0,0 +1,43 @@
import os
import sys
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)
class EnterpriseTest(unittest.TestCase):
def test_basic_rows(self):
server._db_rows = lambda db, sql: [{'time': 1, 'username': 'a', 'password': 'b'}]
status, payload = server.h_enterprise_data(type('C', (), {'args': ('basic',)})())
self.assertEqual(status, 200)
self.assertEqual(payload['rows'][0]['username'], 'a')
def test_challenge_rows(self):
server._db_rows = lambda db, sql: []
status, payload = server.h_enterprise_data(type('C', (), {'args': ('challenge',)})())
self.assertEqual(status, 200)
self.assertEqual(payload['rows'], [])
def test_unknown_table(self):
status, payload = server.h_enterprise_data(type('C', (), {'args': ('nope',)})())
self.assertEqual(status, 400)
def test_clear(self):
calls = []
server._db_write = lambda db, sql: calls.append(sql)
status, payload = server.h_enterprise_clear(type('C', (), {'body': {'table': 'challenge'}})())
self.assertEqual(status, 200)
self.assertTrue(any('hostap_challenge' in s for s in calls))
def test_clear_unknown_table(self):
status, payload = server.h_enterprise_clear(type('C', (), {'body': {'table': 'nope'}})())
self.assertEqual(status, 400)
if __name__ == '__main__':
unittest.main()
tests/test_pineap_modes.py
+126
View File
@@ -0,0 +1,126 @@
import os
import shutil
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):
return type('C', (), {'body': body, 'args': (), 'query': {}})()
class PineapModeTest(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.mkdtemp(prefix='pager-mode-')
self.old_state_file = server.PINEAP_STATE_FILE
self.old_proxy = server._daemon_proxy
self.old_sock = server.daemon_sock_call
server.PINEAP_STATE_FILE = os.path.join(self.tmp, 'state.json')
def tearDown(self):
server.PINEAP_STATE_FILE = self.old_state_file
server._daemon_proxy = self.old_proxy
server.daemon_sock_call = self.old_sock
shutil.rmtree(self.tmp)
def test_passive_applies_mark_vii_supported_preset(self):
calls = []
server._daemon_proxy = lambda method, path, body=None, timeout=15: (
calls.append((method, path, body)) or (200, {'success': True}))
status, payload = server.h_pineap_mode_post(ctx({'mode': 'passive'}))
self.assertEqual(status, 200)
self.assertEqual([c[1] for c in calls], [
'hostapd/enable_pineap', 'ssidpool/enable_collect',
'ssidpool/disable'])
self.assertEqual(calls[0][2], {'enable': False})
self.assertEqual(payload['mode'], 'passive')
self.assertTrue(payload['collect'])
self.assertFalse(payload['advertise'])
self.assertFalse(payload['karma'])
self.assertFalse(payload['enabled'])
def test_active_enables_response_engine_and_pool_broadcasting(self):
calls = []
server._daemon_proxy = lambda method, path, body=None, timeout=15: (
calls.append((method, path, body)) or (200, {'success': True}))
status, payload = server.h_pineap_mode_post(ctx({'mode': 'active'}))
self.assertEqual(status, 200)
self.assertEqual(calls[0], ('PUT', 'hostapd/enable_pineap', {'enable': True}))
self.assertNotIn('mimic/disable', [c[1] for c in calls])
self.assertEqual(calls[-1], ('POST', 'ssidpool/enable', {'enable': True}))
self.assertEqual(payload['mode'], 'active')
self.assertTrue(payload['advertise'])
self.assertTrue(payload['karma'])
self.assertTrue(payload['enabled'])
def test_advanced_preserves_device_settings(self):
calls = []
server._daemon_proxy = lambda *args, **kwargs: (calls.append(args) or (200, {}))
status, payload = server.h_pineap_mode_post(ctx({'mode': 'advanced'}))
self.assertEqual(status, 200)
self.assertEqual(calls, [])
self.assertEqual(payload['mode'], 'advanced')
def test_get_invalidates_preset_when_device_state_disagrees(self):
server.update_pineap_state(mode='passive', collect=True, advertise=False)
def fake(method, path, body=None, timeout=10):
if path.endswith('/get_config') and 'hostapd' not in path:
return 200, {'autossidpool': False}
return 200, {'pineap_disabled': False}
server.daemon_sock_call = fake
status, payload = server.h_pineap_mode_get(ctx())
self.assertEqual(status, 200)
self.assertEqual(payload['mode'], 'advanced')
self.assertFalse(payload['collect'])
def test_get_keeps_passive_when_response_engine_is_disabled(self):
server.update_pineap_state(mode='passive', enabled=False,
collect=True, advertise=False)
def fake(method, path, body=None, timeout=10):
if 'hostapd' in path:
return 200, {'pineap_disabled': True}
return 200, {'autossidpool': True}
server.daemon_sock_call = fake
status, payload = server.h_pineap_mode_get(ctx())
self.assertEqual(status, 200)
self.assertEqual(payload['mode'], 'passive')
self.assertFalse(payload['enabled'])
def test_get_invalidates_active_when_response_engine_stops(self):
server.update_pineap_state(mode='active', enabled=True,
collect=True, advertise=True)
def fake(method, path, body=None, timeout=10):
if 'hostapd' in path:
return 200, {'pineap_disabled': True}
return 200, {'autossidpool': True}
server.daemon_sock_call = fake
status, payload = server.h_pineap_mode_get(ctx())
self.assertEqual(status, 200)
self.assertEqual(payload['mode'], 'advanced')
self.assertFalse(payload['enabled'])
def test_manual_mimic_change_marks_advanced(self):
server.update_pineap_state(mode='passive', karma=False)
server._daemon_proxy = lambda *args, **kwargs: (200, {'success': True})
status, _ = server.h_pineap_mimic(ctx({'enable': True}))
self.assertEqual(status, 200)
state = server.load_pineap_state()
self.assertEqual(state['mode'], 'advanced')
self.assertTrue(state['karma'])
if __name__ == '__main__':
unittest.main()
tests/test_pineap_pool.py
+82
View File
@@ -0,0 +1,82 @@
import os
import sys
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)
class Hak5Test(unittest.TestCase):
def test_hak5_uses_full_path_arglist(self):
calls = []
def fake(args, timeout=20):
calls.append(args)
return 0, 'line1\n', ''
server.device_run = fake
out = server.hak5('PINEAPPLE_SSID_POOL_LIST')
self.assertEqual(calls[0][0], server.HAK5CMD)
self.assertIn('PINEAPPLE_SSID_POOL_LIST', calls[0])
self.assertEqual(out, 'line1\n')
class PoolParsingTest(unittest.TestCase):
def test_parse_json_pool(self):
text = '{"ssids": ["A", "B"]}'
self.assertEqual(server._parse_pool_list(text), ['A', 'B'])
def test_parse_line_pool(self):
text = '"SSID A"\nSSID B\n'
self.assertEqual(server._parse_pool_list(text), ['SSID A', 'SSID B'])
class SsidPoolHandlersTest(unittest.TestCase):
def test_ssids_get(self):
server.hak5 = lambda *a, **k: '{"ssids": ["FreeWiFi"]}\n'
class Ctx:
args = ()
status, payload = server.h_ssids_get(Ctx())
self.assertEqual(status, 200)
self.assertEqual(payload['ssids'], ['FreeWiFi'])
def test_ssids_post_add(self):
calls = []
def fake(*args):
calls.append(args)
return 'ok'
server.hak5 = fake
server.h_ssids_post(type('C', (), {'args': (), 'body': {'action': 'add', 'ssid': 'NewNet'}})())
self.assertTrue(any(c[0] == 'PINEAPPLE_SSID_POOL_ADD' for c in calls))
def test_advertise_routes(self):
calls = []
def fake(method, path, body=None, timeout=10):
calls.append((path, body))
return (200, {'success': True})
server.daemon_sock_call = fake
server.h_pineap_advertise(type('C', (), {'body': {'enable': True}})())
server.h_pineap_advertise(type('C', (), {'body': {'enable': False}})())
self.assertEqual(calls, [('/api/pineap/ssidpool/enable', {'enable': True}),
('/api/pineap/ssidpool/disable', {'enable': False})])
def test_collect_routes(self):
calls = []
def fake(method, path, body=None, timeout=10):
calls.append((path, body))
return (200, {'success': True})
server.daemon_sock_call = fake
server.h_pineap_collect(type('C', (), {'body': {'enable': True}})())
self.assertEqual(calls[0], ('/api/pineap/ssidpool/enable_collect', {'enable': True}))
if __name__ == '__main__':
unittest.main()
tests/test_pineap_proxy.py
+258
View File
@@ -0,0 +1,258 @@
import os
import sys
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, args=()):
return type('C', (), {'body': body, 'args': args, 'query': {}})()
class PineapProxyTest(unittest.TestCase):
def test_proxy_get_passthrough(self):
server.daemon_sock_call = lambda method, path, body=None, timeout=10: (200, {'loghandshake': False})
status, payload = server.h_pineap_get_config(ctx())
self.assertEqual(status, 200)
self.assertEqual(payload['loghandshake'], False)
def test_proxy_post_passthrough(self):
calls = []
def fake(method, path, body=None, timeout=10):
calls.append((method, path, body))
return (200, {'success': True})
server.daemon_sock_call = fake
server.h_pineap_enable(ctx({'enable': True}))
self.assertEqual(calls[0], ('PUT', '/api/pineap/hostapd/enable_pineap', {'enable': True}))
def test_proxy_502_on_socket_failure(self):
server.daemon_sock_call = lambda method, path, body=None, timeout=10: (0, None)
status, payload = server.h_pineap_get_config(ctx())
self.assertEqual(status, 502)
def test_mimic_routes_enable_and_disable(self):
calls = []
def fake(method, path, body=None, timeout=10):
calls.append(path)
return (200, {'success': True})
server.daemon_sock_call = fake
server.h_pineap_mimic(ctx({'enable': True}))
server.h_pineap_mimic(ctx({'enable': False}))
self.assertEqual(calls, ['/api/pineap/mimic/enable', '/api/pineap/mimic/disable'])
def test_examine_reset(self):
calls = []
def fake(method, path, body=None, timeout=10):
calls.append((path, body))
return (200, {'success': True})
server.daemon_sock_call = fake
server.h_pineap_examine(ctx({'reset': True}))
self.assertEqual(calls[0], ('/api/pineap/examine/reset', {'reset': True}))
def test_set_config_forwards_body(self):
calls = []
sock_calls = []
def fake_get(method, path, body=None, timeout=10):
if path == '/api/pineap/get_config':
return 200, {'reconpath': '/root/recon/', 'reconname': 'pager', 'autossidpool': False}
sock_calls.append((method, path))
return (200, {'success': True})
def fake(method, path, body=None, timeout=10):
calls.append((method, path, body))
return (200, {'success': True})
server.daemon_sock_call = lambda m, p, body=None, timeout=10: fake_get(m, p, body, timeout) if m == 'GET' else fake(m, p, body, timeout)
server.h_pineap_set_config(ctx({'loghandshake': True}))
self.assertEqual(calls[0][0], 'PUT')
self.assertEqual(calls[0][1], '/api/pineap/set_config')
self.assertEqual(calls[0][2]['reconpath'], '/root/recon/')
self.assertEqual(calls[0][2]['reconname'], 'pager')
self.assertEqual(calls[0][2]['loghandshake'], True)
self.assertEqual(calls[0][2]['autossidpool'], False)
def test_hostapd_set_merges_interfaces(self):
calls = []
def fake(method, path, body=None, timeout=10):
if method == 'GET':
return 200, {'mgmt_ifaces': ['wlan0mgmt'], 'wpa_ifaces': ['wlan0wpa'],
'pineap_disabled': False, 'pineape_disabled': False, 'pineape_auth_pass': True}
calls.append((method, path, body))
return (200, {'success': True})
server.daemon_sock_call = fake
server.h_pineap_hostapd_set(ctx({'pineape_disabled': True}))
method, path, body = calls[0]
self.assertEqual(method, 'PUT')
self.assertEqual(path, '/api/pineap/hostapd/set_config')
self.assertEqual(body['pineape_disabled'], True)
self.assertEqual(body['mgmt_ifaces'], ['wlan0mgmt'])
self.assertEqual(body['wpa_ifaces'], ['wlan0wpa'])
def test_wifi_get_ap_reads_uci_wireless(self):
def fake_run(args):
cmd = args[0]
if cmd == 'uci' and len(args) == 3:
sec = args[2]
if sec == 'wireless.wlan0wpa':
return 0, "wireless.wlan0wpa.ifname='wlan0wpa'\nwireless.wlan0wpa.ssid='Evil1'\nwireless.wlan0wpa.encryption='psk2'\nwireless.wlan0wpa.key='sekret'\nwireless.wlan0wpa.disabled='0'\nwireless.wlan0wpa.hidden='0'\n", ''
if sec == 'wireless.wlan0open':
return 0, "wireless.wlan0open.disabled='1'\nwireless.wlan0open.ssid='pager-open'\nwireless.wlan0open.macaddr='DE:AD:BE:EF:00:01'\nwireless.wlan0open.hidden='1'\n", ''
if sec == 'wireless.radio0':
return 0, "wireless.radio0.channel='6'\nwireless.radio0.country='US'\n", ''
if sec.startswith('pineapd.@ssidpool'):
return 0, "pineapd.@ssidpool[0].bssid='auto'\npineapd.@ssidpool[0].target='broadcast'\n", ''
return 0, '', ''
def fake_sock(method, path, body=None, timeout=10):
if path == '/api/pineap/hostapd/get_config':
return 200, {'pineape_disabled': False}
if path == '/api/pineap/get_config':
return 200, {'autossidpool': True}
return 200, {}
server.device_run = fake_run
server.daemon_sock_call = fake_sock
status, payload = server.h_pineap_wifi_get_ap(ctx())
self.assertEqual(status, 200)
self.assertEqual(payload['wpa'], {'ssid': 'Evil1', 'passphrase': 'sekret', 'enctype': 'psk2',
'hidden': False, 'enabled': True})
self.assertEqual(payload['open']['enabled'], False)
self.assertEqual(payload['open']['ssid'], 'pager-open')
self.assertEqual(payload['open']['bssid'], 'DE:AD:BE:EF:00:01')
self.assertEqual(payload['open']['hidden'], True)
self.assertEqual(payload['open']['channel'], 6)
self.assertEqual(payload['open']['country'], 'US')
self.assertEqual(payload['open']['target'], 'broadcast')
self.assertEqual(payload['enterprise']['enabled'], True)
self.assertEqual(payload['pool']['collecting'], True)
def test_wifi_set_ap_open_bssid_channel_and_country(self):
sock_calls = []
run_calls = []
def fake_sock(method, path, body=None, timeout=10):
sock_calls.append((method, path, body))
return (200, {'success': True})
def fake_run(args):
run_calls.append(args)
if args[0] == 'uci' and args[1] == 'show':
return 0, "wireless.radio0.channel='1'\n", ''
return 0, '', ''
server.daemon_sock_call = fake_sock
server.device_run = fake_run
status, _ = server.h_pineap_wifi_set_ap(ctx({'open': {
'ssid': 'Open', 'bssid': 'DE:AD:BE:EF:00:02', 'hidden': True,
'channel': 6, 'country': 'US', 'enabled': True}}))
self.assertEqual(status, 200)
method, path, body = sock_calls[0]
self.assertEqual(method, 'PUT')
self.assertEqual(path, '/api/settings/wifi/set_ap')
conf = body['configs'][0]
self.assertEqual(conf['interface'], 'wlan0open')
self.assertEqual(conf['ssid'], 'Open')
self.assertEqual(conf['bssid'], 'DE:AD:BE:EF:00:02')
self.assertEqual(conf['hidden'], True)
self.assertEqual(conf['channel'], 6)
self.assertEqual(conf['enabled'], True)
sets = [a for a in run_calls if a[:2] == ['uci', 'set']]
self.assertEqual(sets, [['uci', 'set', 'wireless.radio0.channel=6'],
['uci', 'set', 'wireless.radio0.country=US']])
self.assertIn(['uci', 'commit', 'wireless'], run_calls)
self.assertIn(['wifi', 'reload'], run_calls)
def test_wifi_set_ap_builds_configs(self):
calls = []
def fake_sock(method, path, body=None, timeout=10):
calls.append((method, path, body))
return (200, {'success': True})
server.daemon_sock_call = fake_sock
status, _ = server.h_pineap_wifi_set_ap(ctx({'wpa': {
'ssid': 'Evil2', 'passphrase': 'pass123', 'enctype': 'sae', 'hidden': True, 'enabled': True}}))
self.assertEqual(status, 200)
method, path, body = calls[0]
self.assertEqual(method, 'PUT')
self.assertEqual(path, '/api/settings/wifi/set_ap')
self.assertEqual(body['configs'][0]['interface'], 'wlan0wpa')
self.assertEqual(body['configs'][0]['ssid'], 'Evil2')
self.assertEqual(body['configs'][0]['enctype'], 'sae')
self.assertEqual(body['configs'][0]['key'], 'pass123')
self.assertEqual(body['configs'][0]['enabled'], True)
self.assertEqual(body['configs'][0]['channel'], 1)
class PineapFilterTest(unittest.TestCase):
def test_filter_get_reads_daemon_mode_and_active_list(self):
server.daemon_sock_call = lambda method, path, body=None, timeout=10: (
200, {'mode': 'deny', 'allowed_macs': ['00:11:22:33:44:55'], 'denied_macs': ['AA:BB:CC:DD:EE:FF']})
status, payload = server.h_filter_get(ctx(), 'client')
self.assertEqual(status, 200)
self.assertEqual(payload['mode'], 'deny')
self.assertEqual(payload['entries'], ['AA:BB:CC:DD:EE:FF'])
def test_ssid_filter_get_decodes_daemon_base64_values(self):
server.daemon_sock_call = lambda method, path, body=None, timeout=10: (
200, {'mode': 'allow', 'allowed_ssids': ['Q29kZXhGaWx0ZXJQcm9iZQ==', 'test'],
'denied_ssids': []})
status, payload = server.h_filter_get(ctx(), 'ssid')
self.assertEqual(status, 200)
self.assertEqual(payload['entries'], ['CodexFilterProbe', 'test'])
def test_filter_set_mode_uses_hak5_mode_command(self):
calls = []
server.daemon_sock_call = lambda method, path, body=None, timeout=10: (
200, {'mode': 'deny', 'allowed_macs': [], 'denied_macs': []})
server.device_run = lambda args, timeout=20: (calls.append(args) or (0, '', ''))
server.h_filter_post(ctx({'action': 'set_mode', 'mode': 'allow'}), 'client')
self.assertEqual(calls[0], [server.HAK5CMD, 'PINEAPPLE_DEVICE_FILTER_MODE', 'allow'])
def test_filter_add_includes_active_mode(self):
calls = []
server.daemon_sock_call = lambda method, path, body=None, timeout=10: (
200, {'mode': 'allow', 'allowed_macs': [], 'denied_macs': []})
server.device_run = lambda args, timeout=20: (calls.append(args) or (0, '', ''))
server.h_filter_post(ctx({'action': 'add', 'value': '00:11:22:33:44:55'}), 'client')
self.assertEqual(calls[0], [server.HAK5CMD, 'PINEAPPLE_DEVICE_FILTER_ADD',
'allow', '00:11:22:33:44:55'])
def test_filter_delete_and_clear_include_active_mode(self):
calls = []
server.daemon_sock_call = lambda method, path, body=None, timeout=10: (
200, {'mode': 'deny', 'allowed_ssids': [], 'denied_ssids': ['Guest']})
server.device_run = lambda args, timeout=20: (calls.append(args) or (0, '', ''))
server.h_filter_post(ctx({'action': 'delete', 'value': 'Guest'}), 'ssid')
server.h_filter_post(ctx({'action': 'clear'}), 'ssid')
self.assertEqual(calls[0], [server.HAK5CMD, 'PINEAPPLE_NETWORK_FILTER_DELETE', 'deny', 'Guest'])
self.assertEqual(calls[1], [server.HAK5CMD, 'PINEAPPLE_NETWORK_FILTER_CLEAR', 'deny'])
def test_filter_allow_all_clears_deny_list_and_sets_deny_mode(self):
calls = []
server.daemon_sock_call = lambda method, path, body=None, timeout=10: (
200, {'mode': 'allow', 'allowed_ssids': ['Guest'], 'denied_ssids': []})
server.device_run = lambda args, timeout=20: (calls.append(args) or (0, '', ''))
server.h_filter_post(ctx({'action': 'allow_all'}), 'ssid')
self.assertEqual(calls[:2], [
[server.HAK5CMD, 'PINEAPPLE_NETWORK_FILTER_CLEAR', 'deny'],
[server.HAK5CMD, 'PINEAPPLE_NETWORK_FILTER_MODE', 'deny']])
if __name__ == '__main__':
unittest.main()
tests/test_pineap_settings.py
+55
View File
@@ -0,0 +1,55 @@
import os
import sys
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):
return type('C', (), {'body': body, 'args': (), 'query': {}})()
class GetConfigProxyTest(unittest.TestCase):
def test_get_config_proxies_daemon(self):
server.daemon_sock_call = lambda method, path, body=None, timeout=10: (
200, {'loghandshake': True, 'logrecon': False, 'autossidpool': True})
status, payload = server.h_pineap_get_config(ctx())
self.assertEqual(status, 200)
self.assertTrue(payload['loghandshake'])
self.assertFalse(payload['logrecon'])
def test_set_config_forwards_flags(self):
calls = []
def fake(method, path, body=None, timeout=10):
if method == 'GET':
return 200, {'reconpath': '/root/recon/', 'loghandshake': False,
'logpcap': True, 'logrecon': True}
calls.append((path, body))
return (200, {'success': True})
server.daemon_sock_call = fake
server.h_pineap_set_config(ctx({'loghandshake': True, 'logpcap': False}))
path, body = calls[0]
self.assertEqual(path, '/api/pineap/set_config')
self.assertEqual(body['loghandshake'], True)
self.assertEqual(body['logpcap'], False)
self.assertEqual(body['logrecon'], True)
self.assertEqual(body['reconpath'], '/root/recon/')
self.assertTrue('autossidpool' in body)
def test_hostapd_get(self):
server.daemon_sock_call = lambda method, path, body=None, timeout=10: (
200, {'pineap_disabled': False, 'pineape_disabled': False})
status, payload = server.h_pineap_hostapd_get(ctx())
self.assertEqual(status, 200)
self.assertFalse(payload['pineap_disabled'])
if __name__ == '__main__':
unittest.main()
tests/test_recon.py
+599
View File
@@ -0,0 +1,599 @@
import os
import shutil
import sqlite3
import sys
import tempfile
import unittest
from unittest import mock
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)
SCHEMA = '''
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);
CREATE TABLE handshake(hash INT PRIMARY KEY, scan INT, stahash INT, aphash INT, time INT, beacon BLOB, hs1 BLOB, hs2 BLOB, hs3 BLOB, hs4 BLOB);
CREATE TABLE hostap_basic(id INT PRIMARY KEY, scan INT, time INT, type TEXT, identity TEXT, password TEXT, verified INT NOT NULL DEFAULT 0);
CREATE TABLE hostap_chalresp(id INT PRIMARY KEY, scan INT, time INT, type TEXT, username TEXT, challenge BLOB, response BLOB, verified INT NOT NULL DEFAULT 0);
CREATE TABLE hostap_client(id INT PRIMARY KEY, scan INT, hash INT, mac TEXT, ssid BLOB, connected_time INT, disconnected_time INT);
CREATE TABLE hostap_handshake(id INT PRIMARY KEY, scan INT, time INT, type INT, mic BLOB, apmac BLOB, stamac BLOB, ssid BLOB, nonce BLOB, eapol BLOB);
'''
def make_db():
fd, db = tempfile.mkstemp(suffix='.db')
os.close(fd)
conn = sqlite3.connect(db)
conn.executescript(SCHEMA)
conn.execute("INSERT INTO scan (uuid, time, name) VALUES ('u1', 1786466531, 'pager')")
conn.execute("INSERT INTO scan (uuid, time, name) VALUES ('u2', 1786466848, 'pager')")
conn.execute("INSERT INTO wifi_device (hash, scan, mac, time, signal, freq, packets) VALUES (1, 1, 'AE77C0EB3141', 1786466531, -71, 2412, 5)")
conn.execute("INSERT INTO wifi_device (hash, scan, mac, time, signal, freq, packets) VALUES (2, 1, 'C89E43648080', 1786466532, -76, 5745, 9)")
conn.execute("INSERT INTO ssid (hash, wifi_device, scan, type, bssid, ssid, hidden, time, signal, freq, channel, encryption) "
"VALUES (10, 2, 1, 8, 'C89E43648080', X'416E646572736F6E2D35', 0, 1786466532, -76, 5745, 149, 0x400400108)")
conn.execute("INSERT INTO ssid (hash, wifi_device, scan, type, bssid, ssid, hidden, time, signal, freq, channel, encryption) "
"VALUES (11, 2, 1, 8, '506F9A010000', X'', 1, 1786466532, -64, 5745, 149, 0)")
conn.execute("INSERT INTO ssid (hash, wifi_device, scan, type, bssid, ssid, hidden, time, signal, freq, channel, encryption) "
"VALUES (12, 1, 1, 4, NULL, X'5A6E6574', NULL, 1786466531, -40, 2412, NULL, NULL)")
conn.execute("INSERT INTO handshake (hash, scan, stahash, aphash, time) VALUES (20, 1, 1, 2, 1786466600)")
conn.execute("INSERT INTO hostap_basic (scan, time, type, identity, password, verified) VALUES (1, 1786466601, 'WPA', 'bob', '', 0)")
conn.commit()
conn.close()
return db
class DecodersTest(unittest.TestCase):
def test_fmt_mac_colon_form(self):
self.assertEqual(server.fmt_mac('AE77C0EB3141'), 'AE:77:C0:EB:31:41')
def test_fmt_mac_noop(self):
self.assertEqual(server.fmt_mac('AE:77:C0:EB:31:41'), 'AE:77:C0:EB:31:41')
self.assertEqual(server.fmt_mac(''), '--')
self.assertEqual(server.fmt_mac(None), '--')
def test_norm_mac_12hex(self):
self.assertEqual(server._norm_mac('C89E43648080'), 'C8:9E:43:64:80:80')
def test_decode_ssid_bytes_and_str(self):
self.assertEqual(server.decode_ssid(b'Znet'), 'Znet')
self.assertEqual(server.decode_ssid('Znet'), 'Znet')
self.assertEqual(server.decode_ssid(None), '')
self.assertEqual(server.decode_ssid(b'\xff\xfeZnet'), '\ufffd\ufffdZnet')
def test_decode_ssid_cli_escapes(self):
# Device sqlite3 CLI -json emits \xNN escapes for non-UTF8 BLOBs.
self.assertEqual(server.decode_ssid('casaalicia\\x00.\\xde_;'), 'casaalicia\x00.\ufffd_;')
def test_decode_encryption_cases(self):
self.assertEqual(server.decode_encryption(0), 'Open')
self.assertEqual(server.decode_encryption(None), 'Open')
self.assertEqual(server.decode_encryption(2), 'WEP')
self.assertEqual(server.decode_encryption(0x04), 'WPA')
self.assertEqual(server.decode_encryption(0x08), 'WPA2')
self.assertEqual(server.decode_encryption(0x04 | 0x08), 'WPA2 WPA')
self.assertEqual(server.decode_encryption(0x400400108), 'WPA3 WPA2')
self.assertEqual(server.decode_encryption(0x20050004C), 'WPA2 WPA')
class ReconDataTest(unittest.TestCase):
def setUp(self):
self.db = make_db()
server.RECON_DB = self.db
def tearDown(self):
os.unlink(self.db)
def test_scans_list_uses_time_and_counts(self):
data = server.recon_scans_data()
self.assertEqual(len(data['scans']), 2)
newest = data['scans'][0]
self.assertEqual(newest['id'], 2)
self.assertEqual(newest['time'], 1786466848)
self.assertEqual(newest['name'], 'pager')
old = data['scans'][1]
self.assertEqual(old['devices'], 2)
self.assertEqual(old['aps'], 2)
self.assertEqual(old['handshakes'], 1)
self.assertNotIn('uuid', old)
def test_scan_detail_decodes_aps(self):
data = server.recon_scan_data(1)
self.assertEqual(data['scan']['id'], 1)
self.assertEqual(data['scan']['time'], 1786466531)
self.assertEqual(len(data['aps']), 2)
aps = {a['bssid']: a for a in data['aps']}
a = aps['C8:9E:43:64:80:80']
self.assertEqual(a['ssid'], 'Anderson-5')
self.assertEqual(a['channel'], 149)
self.assertEqual(a['signal'], -76)
self.assertEqual(a['encryption'], 'WPA3 WPA2')
self.assertFalse(a['hidden'])
hidden = aps['50:6F:9A:01:00:00']
self.assertTrue(hidden['hidden'])
self.assertEqual(hidden['encryption'], 'Open')
def test_scan_detail_clients_exclude_ap_macs(self):
data = server.recon_scan_data(1)
macs = [c['mac'] for c in data['clients']]
self.assertEqual(macs, ['AE:77:C0:EB:31:41'])
def test_scan_detail_handshakes_resolve_macs(self):
data = server.recon_scan_data(1)
self.assertEqual(len(data['handshakes']), 1)
hs = data['handshakes'][0]
self.assertEqual(hs['ap'], 'C8:9E:43:64:80:80')
self.assertEqual(hs['client'], 'AE:77:C0:EB:31:41')
self.assertEqual(hs['time'], 1786466600)
def test_scan_detail_missing_returns_none(self):
self.assertIsNone(server.recon_scan_data(999))
class FakeSock:
def __init__(self, resp=b''):
self.resp = resp
self.sent = b''
self.connected = None
def settimeout(self, t):
pass
def connect(self, addr):
self.connected = addr
def sendall(self, b):
self.sent += b
def recv(self, n):
chunk, self.resp = self.resp[:n], self.resp[n:]
return chunk
def close(self):
pass
class DaemonSockTest(unittest.TestCase):
def test_socket_call_posts_json_to_sock(self):
server.DAEMON_SOCK = '/tmp/api.sock'
fake = FakeSock(b'HTTP/1.1 200 OK\r\nContent-Length: 17\r\n\r\n{"success":true}')
with mock.patch.object(server.socket, 'socket', lambda *a, **k: fake):
status, data = server.daemon_sock_call('POST', '/api/pineap/recon/new', body={'x': 1})
self.assertEqual(status, 200)
self.assertEqual(data, {'success': True})
self.assertEqual(fake.connected, '/tmp/api.sock')
self.assertTrue(fake.sent.startswith(b'POST /api/pineap/recon/new HTTP/1.1'))
self.assertTrue(b'Content-Length: 8' in fake.sent)
self.assertTrue(fake.sent.endswith(b'{"x": 1}'))
def test_socket_call_connect_error(self):
def boom(*a, **k):
raise OSError('nope')
with mock.patch.object(server.socket, 'socket', boom):
status, data = server.daemon_sock_call('POST', '/api/pineap/recon/new')
self.assertEqual(status, 0)
self.assertIsNone(data)
def test_start_stop_handlers_call_socket(self):
calls = []
server.daemon_sock_call = lambda m, p, body=None: calls.append((m, p, body)) or (200, {'success': True})
server.h_recon_start(type('C', (), {'args': ()})())
server.h_recon_stop(type('C', (), {'args': ()})())
self.assertEqual(calls[0], ('POST', '/api/pineap/log/recon/start', {}))
self.assertEqual(calls[1], ('POST', '/api/pineap/log/recon/stop', {}))
def test_start_forwards_scan_time(self):
calls = []
server.daemon_sock_call = lambda m, p, body=None: calls.append((m, p, body)) or (200, {'success': True})
ctx = type('C', (), {'args': (), 'body': {'scan_time': 60}})()
status, data = server.h_recon_start(ctx)
self.assertEqual(status, 200)
self.assertEqual(calls[0], ('POST', '/api/pineap/log/recon/start', {'scan_time': 60}))
def test_start_defaults_empty_body(self):
calls = []
server.daemon_sock_call = lambda m, p, body=None: calls.append((m, p, body)) or (200, {'success': True})
server.h_recon_start(type('C', (), {'args': ()})())
self.assertEqual(calls[0], ('POST', '/api/pineap/log/recon/start', {}))
class ReconScanStateTest(unittest.TestCase):
"""The daemon ignores scan_time and scans continuously until 'stop'. The webui
must track the requested duration itself so timed scans actually end and the
toggle can reflect real scan state."""
def setUp(self):
self.db = make_db()
server.RECON_DB = self.db
server._recon_scan_state = {'active': False, 'started': 0, 'duration': 0}
def tearDown(self):
os.unlink(self.db)
def _start(self, scan_time=None, fail=False):
body = {}
if scan_time is not None:
body['scan_time'] = scan_time
server.daemon_sock_call = (lambda m, p, body=None: (502, {})) if fail \
else (lambda m, p, body=None: (200, {'success': True}))
return server.h_recon_start(type('C', (), {'args': (), 'body': body})())
def _stop(self, fail=False):
server.daemon_sock_call = (lambda m, p, body=None: (502, {})) if fail \
else (lambda m, p, body=None: (200, {'success': True}))
return server.h_recon_stop(type('C', (), {'args': ()})())
def _status(self):
server.daemon_sock_call = lambda m, p, body=None: (200, {'success': True})
return server.h_recon_status(type('C', (), {'args': ()})())
def test_timed_start_marks_scanning_with_remaining(self):
server.time.time = lambda: 1000.0
status, data = self._start(scan_time=30)
self.assertEqual(status, 200)
status, data = self._status()
self.assertTrue(data['scanning'])
self.assertEqual(data['scan_remaining'], 30)
def test_timed_scan_expires_when_duration_elapses(self):
base = [1000.0]
server.time.time = lambda: base[0]
self._start(scan_time=30)
base[0] = 1031.0
status, data = self._status()
self.assertFalse(data['scanning'])
self.assertEqual(data['scan_remaining'], 0)
def test_continuous_scan_has_no_remaining(self):
server.time.time = lambda: 1000.0
self._start(scan_time=0)
status, data = self._status()
self.assertTrue(data['scanning'])
self.assertIsNone(data['scan_remaining'])
def test_default_start_is_continuous(self):
server.time.time = lambda: 1000.0
self._start()
status, data = self._status()
self.assertTrue(data['scanning'])
self.assertIsNone(data['scan_remaining'])
def test_stop_clears_scanning(self):
server.time.time = lambda: 1000.0
self._start(scan_time=30)
self._stop()
status, data = self._status()
self.assertFalse(data['scanning'])
def test_start_failure_does_not_mark_scanning(self):
server.time.time = lambda: 1000.0
status, data = self._start(scan_time=30, fail=True)
self.assertEqual(status, 502)
status, data = self._status()
self.assertFalse(data['scanning'])
def test_watchdog_stops_expired_timed_scan(self):
calls = []
server.time.time = lambda: 1000.0
self._start(scan_time=10)
server.time.time = lambda: 1012.0
server.daemon_sock_call = lambda m, p, body=None: calls.append((m, p)) or (200, {'success': True})
server._recon_watchdog_tick()
self.assertEqual(calls, [('POST', '/api/pineap/log/recon/stop')])
self.assertFalse(server._recon_scan_state['active'])
def test_watchdog_leaves_active_scan_alone(self):
server.time.time = lambda: 1000.0
self._start(scan_time=30)
server.time.time = lambda: 1010.0
calls = []
server.daemon_sock_call = lambda m, p, body=None: calls.append((m, p)) or (200, {'success': True})
server._recon_watchdog_tick()
self.assertEqual(calls, [])
self.assertTrue(server._recon_scan_state['active'])
class ReconExtrasTest(unittest.TestCase):
def setUp(self):
self.db = make_db()
server.RECON_DB = self.db
def tearDown(self):
os.unlink(self.db)
def test_status_reports_last_scan_and_active(self):
server.time.time = lambda: 1786466532 + 100
status, data = server.h_recon_status(type('C', (), {'args': ()})())
self.assertEqual(status, 200)
self.assertEqual(data['last_scan'], 1786466848)
self.assertEqual(data['last_activity'], 1786466532)
self.assertTrue(data['active'])
server.time.time = lambda: 1786466532 + 1000
status, data = server.h_recon_status(type('C', (), {'args': ()})())
self.assertFalse(data['active'])
def test_delete_cascades(self):
server.recon_delete_scan(1)
rows = server._db_rows(self.db, 'SELECT count(*) AS c FROM wifi_device')
self.assertEqual(rows[0]['c'], 0)
rows = server._db_rows(self.db, 'SELECT count(*) AS c FROM ssid')
self.assertEqual(rows[0]['c'], 0)
rows = server._db_rows(self.db, 'SELECT count(*) AS c FROM handshake')
self.assertEqual(rows[0]['c'], 0)
rows = server._db_rows(self.db, 'SELECT count(*) AS c FROM scan')
self.assertEqual(rows[0]['c'], 1)
def test_delete_handler_404_for_missing(self):
status, data = server.h_recon_delete(type('C', (), {'args': ('999',)})())
self.assertEqual(status, 404)
def test_events_lists_db_rows(self):
status, data = server.h_recon_events(type('C', (), {'args': ()})())
self.assertEqual(status, 200)
kinds = [e['type'] for e in data['events']]
self.assertIn('auth attempt', kinds)
self.assertEqual(data['events'][0]['time'], 1786466601)
class ReconExamineTest(unittest.TestCase):
def test_examine_bssid_calls_hak5(self):
calls = []
server.hak5 = lambda *args, **kw: calls.append(args) or ''
ctx = type('C', (), {'args': (), 'body': {'bssid': 'AA:BB:CC:DD:EE:FF'}})()
status, data = server.h_recon_examine(ctx)
self.assertEqual(status, 200)
self.assertEqual(calls[0], ('PINEAPPLE_EXAMINE_BSSID', 'AA:BB:CC:DD:EE:FF'))
def test_examine_channel_calls_hak5(self):
calls = []
server.hak5 = lambda *args, **kw: calls.append(args) or ''
ctx = type('C', (), {'args': (), 'body': {'channel': 6}})()
status, data = server.h_recon_examine(ctx)
self.assertEqual(status, 200)
self.assertEqual(calls[0], ('PINEAPPLE_EXAMINE_CHANNEL', '6'))
def test_examine_requires_target(self):
server.hak5 = lambda *args, **kw: ''
ctx = type('C', (), {'args': (), 'body': {}})()
status, data = server.h_recon_examine(ctx)
self.assertEqual(status, 400)
class HandshakeFileTest(unittest.TestCase):
def test_file_download_decodes_name(self):
import shutil
d = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, d)
server.LOOT_HS_DIR = d
name = '00 11_scan.cap'
with open(os.path.join(d, name), 'wb') as f:
f.write(b'PCAPDATA')
status, payload = server.h_handshake_file(type('C', (), {'args': ('00%2011_scan.cap',)}))
self.assertEqual(status, 200)
self.assertEqual(payload.data, b'PCAPDATA')
self.assertEqual(payload.filename, name)
@unittest.skipIf(os.name == 'nt', 'colons are not valid in filenames on Windows')
def test_file_download_decodes_mac_colon_name(self):
import shutil
d = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, d)
server.LOOT_HS_DIR = d
name = '00:11:22:33:44:55_scan.cap'
with open(os.path.join(d, name), 'wb') as f:
f.write(b'PCAPDATA')
encoded = '00%3A11%3A22%3A33%3A44%3A55_scan.cap'
status, payload = server.h_handshake_file(type('C', (), {'args': (encoded,)}))
self.assertEqual(status, 200)
self.assertEqual(payload.data, b'PCAPDATA')
self.assertEqual(payload.filename, name)
def test_file_download_404_missing(self):
import shutil
d = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, d)
server.LOOT_HS_DIR = d
status, payload = server.h_handshake_file(type('C', (), {'args': ('nope.cap',)}))
self.assertEqual(status, 404)
class CliFallbackTest(unittest.TestCase):
def setUp(self):
self.db = make_db()
server.RECON_DB = self.db
self._orig_sqlite3 = server.sqlite3
self._orig_device_run = server.device_run
server.sqlite3 = None
server.device_run = lambda args, timeout=20: (0, self._run(args), '')
def _run(self, args):
# emulate `sqlite3 -json <db> <sql>` over an in-memory copy
import sqlite3 as s3
import subprocess
conn = s3.connect(self.db)
try:
cur = conn.execute(args[-1])
conn.commit()
if cur.description:
import json
rows = cur.fetchall()
cols = [d[0] for d in cur.description]
objs = []
for row in rows:
obj = {}
for c, v in zip(cols, row):
if isinstance(v, bytes):
try:
v = v.decode('utf-8')
except Exception:
v = ''.join('\\x%02x' % b for b in v)
obj[c] = v
objs.append(obj)
return json.dumps(objs)
return ''
finally:
conn.close()
def tearDown(self):
server.sqlite3 = self._orig_sqlite3
server.device_run = self._orig_device_run
os.unlink(self.db)
def test_delete_cascade_via_cli(self):
server.recon_delete_scan(1)
rows = server._db_rows(self.db, 'SELECT count(*) AS c FROM scan')
self.assertEqual(rows[0]['c'], 1)
def test_cli_ssid_escapes_decoded(self):
# emulate the CLI emitting \xNN escapes for a non-UTF8 blob
server.device_run = lambda args, timeout=20: (0, '[{"ssid": "casaalicia\\\\x00.\\\\xde_"}]', '')
server.RECON_DB = '/nonexistent.db'
self.assertEqual(server.decode_ssid('casaalicia\\x00.\\xde_'), 'casaalicia\x00.\ufffd_')
def make_hs_db():
db = make_db()
conn = sqlite3.connect(db)
conn.execute(
"INSERT INTO handshake (hash, scan, stahash, aphash, time, beacon, hs1, hs2, hs3, hs4) "
"VALUES (21, 1, 1, 2, 1786466650, X'424541434F4E', X'01', X'02', X'03', X'04')")
conn.commit()
conn.close()
return db
class ParseHsFilenameTest(unittest.TestCase):
def test_parse_full_pcap(self):
p = server.parse_hs_filename('1786466650_C8:9E:43:64:80:80_AE:77:C0:EB:31:41_handshake.pcap')
self.assertEqual(p['ts'], 1786466650)
self.assertEqual(p['ap'], 'C8:9E:43:64:80:80')
self.assertEqual(p['client'], 'AE:77:C0:EB:31:41')
self.assertEqual(p['kind'], 'full')
self.assertEqual(p['ext'], 'pcap')
def test_parse_partial_and_incomplete(self):
p = server.parse_hs_filename('1_C8-9E-43-64-80-80_AE-77-C0-EB-31-41_handshake_partial.22000')
self.assertEqual(p['kind'], 'partial')
self.assertEqual(p['ext'], '22000')
p = server.parse_hs_filename('1_C8:9E:43:64:80:80_AE:77:C0:EB:31:41_handshake_incomplete.pcap')
self.assertEqual(p['kind'], 'incomplete')
self.assertEqual(p['ext'], 'pcap')
def test_parse_dash_macs_and_no_ts(self):
p = server.parse_hs_filename('C8-9E-43-64-80-80_AE-77-C0-EB-31-41_handshake.pcap')
self.assertIsNone(p['ts'])
self.assertEqual(p['ap'], 'C8:9E:43:64:80:80')
self.assertEqual(p['client'], 'AE:77:C0:EB:31:41')
def test_parse_unrecognized(self):
self.assertIsNone(server.parse_hs_filename('random.cap'))
self.assertIsNone(server.parse_hs_filename('notes.txt'))
self.assertIsNone(server.parse_hs_filename(''))
self.assertIsNone(server.parse_hs_filename('123_mac1_mac2_handshake'))
class HandshakesDataTest(unittest.TestCase):
def setUp(self):
self.db = make_hs_db()
self.dir = tempfile.mkdtemp()
self.addCleanup(server.__dict__.update, {'RECON_DB': server.RECON_DB, 'LOOT_HS_DIR': server.LOOT_HS_DIR})
server.RECON_DB = self.db
server.LOOT_HS_DIR = self.dir
def tearDown(self):
shutil.rmtree(self.dir)
os.unlink(self.db)
def _write(self, name, ts):
path = os.path.join(self.dir, name)
open(path, 'w').close()
os.utime(path, (ts, ts))
def test_empty_dir_skips_db(self):
with mock.patch.object(server, '_db_rows', side_effect=AssertionError('db should not be touched')):
data = server.handshakes_data()
self.assertEqual(data, {'files': [], 'handshakes': []})
def test_correlation_composes_full_record(self):
self._write('1786466650_C8-9E-43-64-80-80_AE-77-C0-EB-31-41_handshake.pcap', 1786466650)
data = server.handshakes_data()
self.assertEqual(len(data['files']), 1)
hs = data['handshakes'][0]
self.assertEqual(hs['mac'], 'C8:9E:43:64:80:80')
self.assertEqual(hs['client'], 'AE:77:C0:EB:31:41')
self.assertEqual(hs['source'], 'Recon')
self.assertEqual(hs['type'], 'full')
self.assertEqual(hs['extension'], 'pcap')
self.assertEqual(hs['timestamp'], 1786466650)
self.assertTrue(hs['in_db'])
self.assertEqual(hs['part_mask'], 15)
self.assertTrue(hs['beacon'])
self.assertEqual(hs['name'], '1786466650_C8-9E-43-64-80-80_AE-77-C0-EB-31-41_handshake.pcap')
self.assertTrue(hs['file_exists'])
self.assertEqual(hs['location'], os.path.join(server.LOOT_HS_DIR, hs['name']))
def test_file_not_in_db_has_question_mark_fields(self):
self._write('1786467000_AA-BB-CC-DD-EE-FF_00-11-22-33-44-55_handshake.pcap', 1786467000)
hs = server.handshakes_data()['handshakes'][0]
self.assertFalse(hs['in_db'])
self.assertEqual(hs['part_mask'], 0)
self.assertFalse(hs['beacon'])
self.assertEqual(hs['timestamp'], 1786467000)
def test_unparseable_file_still_listed_with_placeholders(self):
self._write('random.cap', 1786467005)
hs = server.handshakes_data()['handshakes'][0]
self.assertEqual(hs['mac'], '--')
self.assertEqual(hs['client'], '--')
self.assertFalse(hs['in_db'])
self.assertEqual(hs['extension'], 'cap')
class HandshakeRoutesTest(unittest.TestCase):
def setUp(self):
self.dir = tempfile.mkdtemp()
self.addCleanup(server.__dict__.update, {'LOOT_HS_DIR': server.LOOT_HS_DIR})
server.LOOT_HS_DIR = self.dir
def tearDown(self):
shutil.rmtree(self.dir)
def _write(self, name, data=b'data'):
with open(os.path.join(self.dir, name), 'wb') as f:
f.write(data)
def test_location_returns_loot_dir(self):
status, data = server.h_handshakes_location(type('C', (), {'args': ()})())
self.assertEqual(status, 200)
self.assertEqual(data['location'], self.dir)
def test_location_route_precedes_file_download(self):
h, args = server.ROUTER.dispatch('GET', '/api/pineap/handshakes/location')
self.assertIs(h, server.h_handshakes_location)
def test_delete_all_removes_files(self):
self._write('1_C8-9E-43-64-80-80_AE-77-C0-EB-31-41_handshake.pcap')
self._write('2_C8-9E-43-64-80-80_AE-77-C0-EB-31-41_handshake.22000')
status, data = server.h_handshakes_delete_all(type('C', (), {'args': ()})())
self.assertEqual(status, 200)
self.assertEqual(data['files'], [])
self.assertEqual(data['handshakes'], [])
self.assertEqual(os.listdir(self.dir), [])
def test_delete_all_empty_dir_is_ok(self):
status, data = server.h_handshakes_delete_all(type('C', (), {'args': ()})())
self.assertEqual(status, 200)
self.assertEqual(data['files'], [])
def test_delete_all_skips_dotfiles(self):
self._write('.hidden')
self._write('1_C8-9E-43-64-80-80_AE-77-C0-EB-31-41_handshake.pcap')
status, data = server.h_handshakes_delete_all(type('C', (), {'args': ()})())
self.assertEqual(status, 200)
self.assertEqual(data['files'], [])
self.assertEqual(data['handshakes'], [])
self.assertEqual(os.listdir(self.dir), ['.hidden'])
tests/test_status.py
+104
View File
@@ -0,0 +1,104 @@
import os
import sys
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 patch_device(fn):
def wrapper(self):
server.device_run = fn
self.addCleanup(lambda: setattr(server, 'device_run', server.device_run))
return wrapper
class StatusTest(unittest.TestCase):
def test_battery_from_capacity_file(self):
import tempfile
base = tempfile.mkdtemp()
os.makedirs(os.path.join(base, 'battery'))
with open(os.path.join(base, 'battery', 'type'), 'w') as f:
f.write('Battery')
with open(os.path.join(base, 'battery', 'capacity'), 'w') as f:
f.write('87')
with open(os.path.join(base, 'battery', 'status'), 'w') as f:
f.write('Charging')
self.assertEqual(server.battery_data(base), {'level': 87, 'charging': True})
def test_battery_missing_returns_none(self):
import tempfile
self.assertEqual(server.battery_data(tempfile.mkdtemp()), {'level': None, 'charging': False})
def test_uptime_parses(self):
server.device_run = lambda args, timeout=20: (0, '1234.56 4321.10\n', '')
self.assertEqual(server.uptime_data(), 1234)
def test_disk_parses_df(self):
server.device_run = lambda args, timeout=20: (0, 'Filesystem 1K-blocks Used Available Use% Mounted on\n/dev/mmcblk0p3 8388608 1024000 7364608 13% /root\n', '')
d = server.disk_data()
self.assertEqual(d['avail'], 7364608 * 1024)
def test_wifi_ifaces_extracts_names(self):
server.device_run = lambda args, timeout=20: (0, 'wlan0 ESSID: "Pineapple"\nwlan1 ESSID: "Pineapple"\n', '') if args == ['iwinfo'] else (0, '', '')
self.assertEqual(server.wifi_ifaces(), ['wlan0', 'wlan1'])
def test_assoc_clients_parses(self):
def fake(args, timeout=20):
if args == ['iwinfo']:
return 0, 'wlan0 ESSID: "Pineapple"\n', ''
if args == ['iwinfo', 'wlan0', 'assoclist']:
return 0, '00:11:22:33:44:55 -64 dBm Signal: -64 dBm Rate: 12 Mbit/s\nAA:BB:CC:DD:EE:FF -40 dBm Signal: -40 dBm Rate: 24 Mbit/s\n', ''
return 0, '', ''
server.device_run = fake
clients = server.assoc_clients()
self.assertEqual(len(clients), 2)
self.assertEqual(clients[0]['mac'], '00:11:22:33:44:55')
self.assertEqual(clients[0]['rssi'], -64)
self.assertEqual(clients[0]['iface'], 'wlan0')
def test_h_status_shape(self):
server.device_run = lambda args, timeout=20: (0, '', '')
server.current_token = lambda: 'tok'
server.daemon_call = lambda m, p, body=None, token=None, timeout=15: (200, {'version': '1'})
class Ctx:
cookie = 'AUTH_S=tok'
args = ()
status, payload = server.h_status(Ctx())
self.assertEqual(status, 200)
for k in ('battery', 'firmware', 'daemon', 'wifi', 'clients', 'disk', 'uptime', 'hostname'):
self.assertIn(k, payload)
def test_status_discovers_wifi_interfaces_once(self):
calls = []
def fake(args, timeout=20):
calls.append(args)
if args == ['iwinfo']:
return 0, 'wlan0 ESSID: "Pineapple"\n', ''
return 0, '', ''
server.device_run = fake
server.current_token = lambda: 'tok'
server.daemon_call = lambda m, p, body=None, token=None, timeout=15: (200, {})
server.status_data()
self.assertEqual(calls.count(['iwinfo']), 1)
class DeviceTest(unittest.TestCase):
def test_h_device(self):
server.device_run = lambda args, timeout=20: (0, 'pager\n', '')
class Ctx:
args = ()
status, payload = server.h_device(Ctx())
self.assertEqual(status, 200)
self.assertEqual(payload['hostname'], 'pager')
self.assertIn('macs', payload)
if __name__ == '__main__':
unittest.main()
tests/test_ws.py
+119
View File
@@ -0,0 +1,119 @@
import base64
import hashlib
import json
import os
import sys
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)
class FrameCodecTest(unittest.TestCase):
def test_encode_small_text_frame(self):
raw = server.ws_encode(b'hi', opcode=0x1)
self.assertEqual(raw[0] & 0x80, 0x80)
self.assertEqual(raw[0] & 0x0F, 0x1)
self.assertEqual(raw[1], 2)
def test_decode_unmasked_text(self):
frame = server.ws_encode(b'hello', opcode=0x1)
opcode, payload, consumed = server.ws_decode_frame(frame)
self.assertEqual(opcode, 0x1)
self.assertEqual(payload, b'hello')
self.assertEqual(consumed, len(frame))
def test_decode_masked_text(self):
import struct
mask = b'\x01\x02\x03\x04'
payload = b'abc'
masked = bytes(payload[i] ^ mask[i % 4] for i in range(len(payload)))
frame = bytes([0x81, 0x80 | len(payload)]) + mask + masked
opcode, out, consumed = server.ws_decode_frame(frame)
self.assertEqual(out, b'abc')
def test_handshake_reply_uses_sha1(self):
key = 'dGhlIHNhbXBsZSBub25jZQ=='
reply = server.ws_handshake_reply(key).decode()
self.assertIn('101 Switching Protocols', reply)
self.assertIn('s3pPLMBiTxaQ9kYGzzhZRbK+xOo=', reply)
def test_encode_masked_roundtrip(self):
raw = server.ws_encode(b'hello', opcode=0x1, mask=True)
self.assertTrue(raw[1] & 0x80)
self.assertEqual(raw[1] & 0x7F, 5)
opcode, payload, consumed = server.ws_decode_frame(raw)
self.assertEqual(opcode, 0x1)
self.assertEqual(payload, b'hello')
self.assertEqual(consumed, len(raw))
class RelayDrainTest(unittest.TestCase):
def test_complete_frame_plus_head_of_next(self):
frame1 = server.ws_encode(b'one')
head2 = server.ws_encode(b'two')[:3]
remaining, frames, closed = server._relay_drain(b'', frame1 + head2)
self.assertEqual(frames, [(0x1, b'one')])
self.assertFalse(closed)
self.assertEqual(remaining, head2)
def test_partial_frame_keeps_remainder(self):
head = server.ws_encode(b'hello')[:4]
remaining, frames, closed = server._relay_drain(b'', head)
self.assertEqual(frames, [])
self.assertFalse(closed)
self.assertEqual(remaining, head)
def test_split_frame_across_chunks(self):
frame = server.ws_encode(b'abcdef')
first, second = frame[:3], frame[3:]
remaining, frames, closed = server._relay_drain(b'', first)
self.assertEqual(frames, [])
self.assertEqual(remaining, first)
remaining, frames, closed = server._relay_drain(remaining, second)
self.assertEqual(frames, [(0x1, b'abcdef')])
self.assertFalse(closed)
self.assertEqual(remaining, b'')
def test_binary_frame_collected_with_opcode(self):
frame = server.ws_encode(b'\x00\x01\x02\x03', opcode=0x2)
remaining, frames, closed = server._relay_drain(b'', frame)
self.assertEqual(frames, [(0x2, b'\x00\x01\x02\x03')])
self.assertFalse(closed)
self.assertEqual(remaining, b'')
def test_close_frame_reported(self):
close = bytes([0x88, 0x02]) + b'\x03\xe8'
remaining, frames, closed = server._relay_drain(b'', close)
self.assertTrue(closed)
self.assertEqual(frames, [])
self.assertEqual(remaining, b'')
class WsPoolTest(unittest.TestCase):
def test_broadcast_skips_dead(self):
class FakeSock:
def __init__(self, fail=False):
self.fail = fail
self.sent = []
def sendall(self, b):
if self.fail:
raise OSError('closed')
self.sent.append(b)
pool = server.WSPool()
good = FakeSock()
dead = FakeSock(fail=True)
pool.add(good)
pool.add(dead)
pool.broadcast(b'x')
self.assertEqual(len(good.sent), 1)
self.assertNotIn(dead, pool.clients)
if __name__ == '__main__':
unittest.main()