259 lines
12 KiB
Python
259 lines
12 KiB
Python
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, 'channel': 6})
|
|
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' and args[2] == 'wireless.radio0':
|
|
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()
|