Mark-VIII/tests/test_pineap_enterprise.py

import json
import os
import sqlite3
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)


SCHEMA = '''
CREATE TABLE hostap_basic(id INTEGER PRIMARY KEY, scan INT, time INT, type TEXT,
                          identity TEXT, password TEXT, verified INT NOT NULL DEFAULT 0);
CREATE TABLE hostap_chalresp(id INTEGER 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 INTEGER PRIMARY KEY, scan INT, hash INT, mac TEXT, ssid BLOB,
                           connected_time INT, disconnected_time INT);
'''


class EnterpriseApiTest(unittest.TestCase):
    def setUp(self):
        self._orig_rows = server._db_rows
        self._orig_write = server._db_write

    def tearDown(self):
        server._db_rows = self._orig_rows
        server._db_write = self._orig_write

    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['table'], 'hostap_basic')
        self.assertEqual(payload['rows'][0]['username'], 'a')
        self.assertEqual(payload['rows'][0]['identity'], 'a')

    def test_challenge_rows_empty(self):
        server._db_rows = lambda db, sql: []
        status, payload = server.h_enterprise_data(type('C', (), {'args': ('challenge',)})())
        self.assertEqual(status, 200)
        self.assertEqual(payload['table'], 'hostap_chalresp')
        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_uses_chalresp_table(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_chalresp' in s for s in calls))
        self.assertFalse(any('hostap_challenge' in s for s in calls))

    def test_clear_all(self):
        calls = []
        server._db_write = lambda db, sql: calls.append(sql)
        status, payload = server.h_enterprise_clear(type('C', (), {'body': {'table': 'all'}})())
        self.assertEqual(status, 200)
        joined = ' '.join(calls)
        self.assertIn('hostap_basic', joined)
        self.assertIn('hostap_chalresp', joined)

    def test_clear_unknown_table(self):
        status, payload = server.h_enterprise_clear(type('C', (), {'body': {'table': 'nope'}})())
        self.assertEqual(status, 400)


class EnterpriseHashFormatTest(unittest.TestCase):
    def test_blob_to_hex_bytes_and_hex_string(self):
        self.assertEqual(server._blob_to_hex(b'\x11\x22\x33\x44'), '11223344')
        self.assertEqual(server._blob_to_hex('AABBCCDD'), 'aabbccdd')
        self.assertEqual(server._blob_to_hex("X'AABB'"), 'aabb')
        self.assertEqual(server._blob_to_hex('\\xde\\xad'), 'dead')

    def test_hashcat_5500_and_john(self):
        chal = '1122334455667788'
        resp = '00112233445566778899aabbccddeeff0011223344556677'
        self.assertEqual(
            server._mschap_hashcat_5500('bob', chal, resp),
            'bob::::00112233445566778899aabbccddeeff0011223344556677:1122334455667788')
        self.assertEqual(
            server._mschap_john('bob', chal, resp),
            'bob:$NETNTLM$1122334455667788$00112233445566778899aabbccddeeff0011223344556677')

    def test_format_chalresp_row_hexes_blobs_and_is_json_safe(self):
        row = server._format_chalresp_row({
            'time': 1700000000,
            'username': 'alice',
            'type': 'MSCHAPV2',
            'challenge': bytes.fromhex('1122334455667788'),
            'response': bytes.fromhex('00112233445566778899aabbccddeeff0011223344556677'),
            'verified': 0,
        })
        self.assertEqual(row['challenge'], '1122334455667788')
        self.assertEqual(row['response'], '00112233445566778899aabbccddeeff0011223344556677')
        self.assertIn('alice::::', row['hashcat'])
        self.assertIn(':$NETNTLM$', row['john'])
        json.dumps(row)


class EnterpriseLogParseTest(unittest.TestCase):
    def test_parse_wpe_mschapv2_and_identity(self):
        log = (
            "mschapv2: Wed Aug 19 21:00:00 2026\n"
            "        username: bob\n"
            "        challenge: 11:22:33:44:55:66:77:88\n"
            "        response: 00112233445566778899aabbccddeeff0011223344556677\n"
            "hashcat NETNTLM: bob::::00112233445566778899aabbccddeeff0011223344556677:1122334455667788\n"
            "EAP-Identity 'alice@corp.local'\n"
            "GTC: username: carol password: hunter2\n"
        )
        items = server._parse_ent_log(log)
        kinds = [i['kind'] for i in items]
        self.assertIn('mschapv2', kinds)
        self.assertIn('eap-identity', kinds)
        self.assertIn('gtc', kinds)
        mschap = [i for i in items if i['kind'] == 'mschapv2'][0]
        self.assertEqual(mschap['username'], 'bob')
        self.assertIn('bob::::', mschap['hashcat'])
        gtc = [i for i in items if i['kind'] == 'gtc'][0]
        self.assertEqual(gtc['password'], 'hunter2')


class EnterpriseCaptureDbTest(unittest.TestCase):
    def setUp(self):
        fd, self.db = tempfile.mkstemp(suffix='.db')
        os.close(fd)
        conn = sqlite3.connect(self.db)
        conn.executescript(SCHEMA)
        chal = bytes.fromhex('1122334455667788')
        resp = bytes.fromhex('00112233445566778899aabbccddeeff0011223344556677')
        conn.execute(
            "INSERT INTO hostap_basic (id, scan, time, type, identity, password, verified) "
            "VALUES (1, 1, 1700000001, 'PEAP', 'bob', '', 0)")
        conn.execute(
            "INSERT INTO hostap_chalresp (id, scan, time, type, username, challenge, response, verified) "
            "VALUES (1, 1, 1700000002, 'MSCHAPV2', 'bob', ?, ?, 0)", (chal, resp))
        conn.execute(
            "INSERT INTO hostap_client (id, scan, hash, mac, ssid, connected_time, disconnected_time) "
            "VALUES (1, 1, 1, 'AABBCCDDEEFF', X'436F7270', 1700000003, NULL)")
        conn.commit()
        conn.close()
        self.old_db = server.RECON_DB
        server.RECON_DB = self.db

    def tearDown(self):
        server.RECON_DB = self.old_db
        try:
            os.unlink(self.db)
        except OSError:
            pass

    def test_challenge_endpoint_returns_hashcat(self):
        status, payload = server.h_enterprise_data(type('C', (), {'args': ('challenge',)})())
        self.assertEqual(status, 200)
        self.assertEqual(payload['table'], 'hostap_chalresp')
        row = payload['rows'][0]
        self.assertEqual(row['username'], 'bob')
        self.assertEqual(row['challenge'], '1122334455667788')
        self.assertEqual(
            row['hashcat'],
            'bob::::00112233445566778899aabbccddeeff0011223344556677:1122334455667788')
        json.dumps(payload, default=server._json_default)

    def test_radius_payload_unifies_captures(self):
        status, payload = server.h_enterprise_radius(type('C', (), {'args': ()})())
        self.assertEqual(status, 200)
        self.assertTrue(payload['note'])
        kinds = [c['kind'] for c in payload['captures']]
        self.assertIn('eap-identity', kinds)
        self.assertIn('mschapv2', kinds)
        self.assertEqual(payload['hashcat']['mode'], 5500)
        self.assertEqual(len(payload['hashcat']['lines']), 1)
        self.assertEqual(payload['clients'][0]['ssid'], 'Corp')

    def test_export_hashcat_download(self):
        status, payload = server.h_enterprise_export(type('C', (), {'args': ('hashcat',)})())
        self.assertEqual(status, 200)
        self.assertIsInstance(payload, server.Download)
        self.assertIn(b'bob::::', payload.data)
        self.assertTrue(payload.filename.endswith('.5500'))

    def test_export_john_and_json(self):
        status, payload = server.h_enterprise_export(type('C', (), {'args': ('john',)})())
        self.assertEqual(status, 200)
        self.assertIn(b'$NETNTLM$', payload.data)
        status, payload = server.h_enterprise_export(type('C', (), {'args': ('json',)})())
        self.assertEqual(status, 200)
        body = json.loads(payload.data.decode('utf-8'))
        self.assertEqual(body['hashcat']['mode'], 5500)


class EnterpriseHarvestTest(unittest.TestCase):
    def test_harvest_reads_hostapd_file_and_skips_logread(self):
        calls = []
        fd, log_path = tempfile.mkstemp()
        os.write(fd, b"EAP-Identity 'fromfile'\n")
        os.close(fd)
        cap_path = log_path + '.json'
        old_log, old_cap = server.ENT_LOG, server.ENT_CAPTURES
        old_run = server.device_run
        server.ENT_LOG = log_path
        server.ENT_CAPTURES = cap_path
        server.device_run = lambda args, timeout=20, input_data=None: (
            calls.append(list(args)) or (0, "identity: 'syslog-user'\n", ''))
        try:
            items = server._harvest_ent_log()
            self.assertTrue(any(item.get('username') == 'fromfile' for item in items))
            self.assertFalse(any(args and args[0] == 'logread' for args in calls))
        finally:
            server.ENT_LOG = old_log
            server.ENT_CAPTURES = old_cap
            server.device_run = old_run
            os.unlink(log_path)
            if os.path.exists(cap_path):
                os.unlink(cap_path)


if __name__ == '__main__':
    unittest.main()