iwd/autotests/testEAP/connection_test.py

#!/usr/bin/python3

from typing import Iterable
import unittest
import sys

sys.path.append('../util')
from iwd import IWD
from iwd import NetworkType
from iwd import PSKAgent
from hlrauc import AuthCenter
from ofono import Ofono
from config import ctx
from hostapd import HostapdCLI

import testutil
import traceback

class Test(unittest.TestCase):
    def copy_network(self, name):
        IWD.copy_to_storage(name, name='ssidEAP.8021x')
        self.wd.wait_for_object_condition(self.wd,
                                    '"ssidEAP" in [n.name for n in obj.list_known_networks()]')

    def remove_network(self):
        networks = self.wd.list_known_networks()
        [n.forget() for n in networks if n.name == 'ssidEAP']
        self.wd.wait_for_object_condition(self.wd,
                                    '"ssidEAP" not in [n.name for n in obj.list_known_networks()]')

    def validate_connection(self, wd, *secrets):
        if secrets:
            psk_agent = PSKAgent(*secrets)
            wd.register_psk_agent(psk_agent)

        devices = wd.list_devices(1)
        self.assertIsNotNone(devices)
        device = devices[0]

        ordered_network = device.get_ordered_network('ssidEAP')

        self.assertEqual(ordered_network.type, NetworkType.eap)

        condition = 'not obj.connected'
        wd.wait_for_object_condition(ordered_network.network_object, condition)

        ordered_network.network_object.connect()

        condition = 'obj.state == DeviceState.connected'
        wd.wait_for_object_condition(device, condition)

        testutil.test_iface_operstate()
        testutil.test_ifaces_connected()

        self.hostapd.eapol_reauth(device.address)

        if secrets:
            wd.unregister_psk_agent(psk_agent)

        device.disconnect()

        condition = 'not obj.connected'
        wd.wait_for_object_condition(ordered_network.network_object, condition)

    #
    # EAP-AKA
    #
    def test_eap_aka(self):
        if not ctx.is_process_running('ofonod'):
            self.skipTest("ofono not running")

        ofono = Ofono()
        ofono.enable_modem('/phonesim')
        ofono.wait_for_sim_auth()

        auth = AuthCenter('/tmp/hlrauc.sock', '/tmp/sim/aka.db')

        self.copy_network('sim/ssidEAP-AKA.8021x')

        try:
            self.validate_connection(self.wd)
        finally:
            auth.stop()

    #
    # EAP-AKA'
    #
    def test_eap_aka_prime(self):
        if not ctx.is_process_running('ofonod'):
            self.skipTest("ofono not running")

        ofono = Ofono()
        ofono.enable_modem('/phonesim')
        ofono.wait_for_sim_auth()

        auth = AuthCenter('/tmp/hlrauc.sock', '/tmp/sim/aka.db')

        self.copy_network('sim/ssidEAP-AKA-prime.8021x')

        try:
            self.validate_connection(self.wd)
        finally:
            auth.stop()

    #
    # EAP-SIM
    #
    def test_eap_sim(self):
        if not ctx.is_process_running('ofonod'):
            self.skipTest("ofono not running")

        ofono = Ofono()
        ofono.enable_modem('/phonesim')
        ofono.wait_for_sim_auth()

        auth = AuthCenter('/tmp/hlrauc.sock', '/tmp/sim/sim.db')

        self.copy_network('sim/ssidEAP-SIM.8021x')

        try:
            self.validate_connection(self.wd)
        finally:
            auth.stop()

    #
    # EAP-MSCHAPv2
    #
    # * Credentials in 8021x file
    # * Password-Hash in 8021x file
    # * Agent request for password
    # * Agent request for user + password
    #
    def test_eap_mschapv2(self):
        self.copy_network('mschapv2/ssidEAP-MSCHAPV2.8021x')
        self.validate_connection(self.wd)

        self.copy_network('mschapv2/ssidEAP-MSCHAPV2-hash.8021x')
        self.validate_connection(self.wd)

        self.copy_network('mschapv2/ssidEAP-MSCHAPV2-nopass.8021x')
        self.validate_connection(self.wd, [], ('mschapv2@example.com', 'Password'))

        self.copy_network('mschapv2/ssidEAP-MSCHAPV2-nouserpass.8021x')
        self.validate_connection(self.wd, [], ('mschapv2@example.com', 'Password'))

    #
    # EAP-PEAP
    #
    # * Test all combinations of PEAP, PEAPv0, PEAPv1 with MD5, GTC, MSCHAPv2
    #
    def test_eap_peap(self):
        for ver in ['PEAP', 'PEAPv0', 'PEAPv1']:
            for inner in ['MD5', 'GTC', 'MSCHAPv2']:
                self.copy_network('peap/ssidEAP-%s-%s.8021x' % (ver, inner))

                try:
                    self.validate_connection(self.wd)
                except Exception as e:
                    # Catch an error here and print the actual PEAP combo that failed
                    traceback.print_exc()
                    raise Exception("%s-%s test failed" % (ver, inner))

                self.remove_network()

    #
    # EAP-PEAP + SIM
    #
    # * Tests EAP-PEAP + SIM separately to allow skipping if ofono is not found
    #
    def test_eap_peap_sim(self):
        if not ctx.is_process_running('ofonod'):
            self.skipTest("ofono not running")

        ofono = Ofono()
        ofono.enable_modem('/phonesim')
        ofono.wait_for_sim_auth()

        auth = AuthCenter('/tmp/hlrauc.sock', '/tmp/sim/sim.db')

        for ver in ['PEAP', 'PEAPv0', 'PEAPv1']:
                self.copy_network('peap/ssidEAP-%s-SIM.8021x' % ver)

                try:
                    self.validate_connection(self.wd)
                except Exception as e:
                    # Catch an error here and print the actual PEAP combo that failed
                    traceback.print_exc()
                    auth.stop()
                    raise Exception("%s-SIM test failed" % ver)

                self.remove_network()

        auth.stop()

    #
    # EAP-PWD
    #
    def test_eap_pwd(self):
        self.copy_network('ssidEAP-PWD.8021x')

        self.validate_connection(self.wd)

    #
    # EAP-TLS
    #
    # * Encrypted private key, passphrase in 8021x file
    # * Unencrypted private key
    # * Encrypted private key, passphrase provided by agent
    # * Embedded PEM inside 8021x file
    # * KeyBundle
    #
    def test_eap_tls(self):
        for name, secrets in [('keypass', None), ('nokeypass', None),
                        ('des-ede3', 'abc'), ('embedded', None), ('keybundle', None)]:
            self.copy_network('tls/ssidEAP-TLS-%s.8021x' % name)
            try:
                self.validate_connection(self.wd, secrets)
            except Exception as e:
                traceback.print_exc()
                raise Exception('EAP-TLS (%s) failed' % name)

            self.remove_network()

    #
    # EAP-TTLS
    #
    # * CHAP, MD5, MSCHAPV2 as phase 2
    # * Tunneled-MSCHAP, Tunneled-MSCHAPV2, Tunneled-PAP as phase 2
    #
    def test_eap_ttls(self):
        for name, secrets in [('CHAP', ('ttls@example.com', ('ttls-chap-phase2@example.com', 'Password'))),
                              ('MD5', None),
                              ('MSCHAPV2', ('ttls@example.com', ('mschapv2-phase2@example.com', 'Password'))),
                              ('Tunneled-MSCHAP', ('ttls@example.com', ('ttls-mschap-phase2@example.com', 'Password'))),
                              ('Tunneled-MSCHAPV2', ('ttls@example.com', ('ttls-mschapv2-phase2@example.com', 'Password'))),
                              ('Tunneled-PAP', ('ttls@example.com', ('ttls-pap-phase2@example.com', 'Password')))]:
            self.copy_network('ttls/ssidEAP-TTLS-%s.8021x' % name)
            try:
                if isinstance(secrets, Iterable):
                    self.validate_connection(self.wd, *secrets)
                else:
                    self.validate_connection(self.wd, None)
            except Exception as e:
                    traceback.print_exc()
                    raise Exception('EAP-TTLS (%s) failed' % name)

            self.remove_network()

    def setUp(self):
        IWD.clear_storage()

    def tearDown(self):
        self.remove_network()

    @classmethod
    def setUpClass(cls):
        cls.wd = IWD()
        cls.hostapd = HostapdCLI()

    @classmethod
    def tearDownClass(cls):
        cls.wd = None
        IWD.clear_storage()

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