diff options
Diffstat (limited to 'debian/tests')
| -rw-r--r-- | debian/tests/Makefile | 8 | ||||
| -rw-r--r-- | debian/tests/control | 15 | ||||
| -rw-r--r-- | debian/tests/fake-rfkill.c | 51 | ||||
| -rwxr-xr-x | debian/tests/killswitches-no-urfkill | 47 | ||||
| -rw-r--r-- | debian/tests/network_test_base.py | 360 | ||||
| -rwxr-xr-x | debian/tests/nm | 848 | ||||
| -rw-r--r-- | debian/tests/urfkill-integration | 103 | ||||
| -rwxr-xr-x | debian/tests/wpa-dhclient | 239 |
8 files changed, 1671 insertions, 0 deletions
diff --git a/debian/tests/Makefile b/debian/tests/Makefile new file mode 100644 index 00000000..77b9baab --- /dev/null +++ b/debian/tests/Makefile @@ -0,0 +1,8 @@ + +obj-m += fake-rfkill.o + +fake-rfkill: + make -C /lib/modules/$(shell uname -r)/build KBUILD_SRC=/lib/modules/$(shell uname -r)/build M=$(shell pwd)/debian/tests + +clean-rfkill: + make -C /lib/modules/$(shell uname -r)/build KBUILD_SRC=/lib/modules/$(shell uname -r)/build M=$(shell pwd)/debian/tests clean diff --git a/debian/tests/control b/debian/tests/control new file mode 100644 index 00000000..5bce4cf9 --- /dev/null +++ b/debian/tests/control @@ -0,0 +1,15 @@ +Tests: wpa-dhclient +Depends: python3, hostapd, dnsmasq-base, wpasupplicant, isc-dhcp-client, iw +Restrictions: needs-root allow-stderr isolation-machine + +Tests: nm +Depends: python3, gir1.2-networkmanager-1.0, network-manager, hostapd, iw, python3-dbusmock +Restrictions: needs-root isolation-machine + +Tests: killswitches-no-urfkill +Depends: network-manager, build-essential, linux-headers-generic, rfkill +Restrictions: needs-root allow-stderr isolation-machine + +Tests: urfkill-integration +Depends: network-manager, build-essential, linux-headers-generic, rfkill, urfkill +Restrictions: needs-root allow-stderr isolation-machine diff --git a/debian/tests/fake-rfkill.c b/debian/tests/fake-rfkill.c new file mode 100644 index 00000000..93aa2859 --- /dev/null +++ b/debian/tests/fake-rfkill.c @@ -0,0 +1,51 @@ +#include <linux/rfkill.h> +#include <linux/module.h> + +MODULE_LICENSE("GPL"); +MODULE_AUTHOR("Johannes Berg <johannes@sipsolutions.net>"); + +static struct rfkill *rfk; + +static void test_poll(struct rfkill *rfkill, void *data) +{ + printk(KERN_DEBUG "poll test rfkill\n"); +} + +static void test_query(struct rfkill *rfkill, void *data) +{ + printk(KERN_DEBUG "query test rfkill\n"); +} + +static int test_set_block(void *data, bool blocked) +{ + printk(KERN_DEBUG "set test rfkill (%s)\n", + blocked ? "blocked" : "active"); + return 0; +} + +static struct rfkill_ops ops = { + .poll = test_poll, + .query = test_query, + .set_block = test_set_block, +}; + +int mod_init(void) +{ + int err; + + rfk = rfkill_alloc("fake", NULL, RFKILL_TYPE_WLAN, &ops, NULL); + if (!rfk) + return -ENOMEM; + err = rfkill_register(rfk); + if (err) + rfkill_destroy(rfk); + return err; +} +module_init(mod_init); + +void mod_exit(void) +{ + rfkill_unregister(rfk); + rfkill_destroy(rfk); +} +module_exit(mod_exit); diff --git a/debian/tests/killswitches-no-urfkill b/debian/tests/killswitches-no-urfkill new file mode 100755 index 00000000..ea6f6e90 --- /dev/null +++ b/debian/tests/killswitches-no-urfkill @@ -0,0 +1,47 @@ +#!/bin/sh +set -e + +make -f debian/tests/Makefile fake-rfkill +# poor man's dependency resolver +DEPS=$(modinfo debian/tests/fake-rfkill.ko | sed -n '/depends:/ {s/^.*://; s/[[:space:]]*$//; p}') +[ -z "$DEPS" ] || modprobe "$DEPS" +insmod debian/tests/fake-rfkill.ko + +fake_id=$(rfkill list | grep fake | awk -F: '{ print $1; }') + +service network-manager start +sleep 30 + +# test blocking the device +rfkill block $fake_id +if ! rfkill list $fake_id | grep 'Soft' | awk '{ print $3; }' | grep -qc yes; then + echo "ERROR: could not block fake device" + rmmod fake-rfkill || true + exit 1 +fi + +if ! LC_MESSAGES=C nmcli radio wifi | grep -qc disabled; then + echo "ERROR: NM could not track device state." + rmmod fake-rfkill || true + exit 1 +fi + + +# test unblocking the device +rfkill unblock $fake_id +if ! rfkill list $fake_id | grep 'Soft' | awk '{ print $3; }' | grep -qc no; then + echo "ERROR: could not unblock fake device" + rmmod fake-rfkill || true + exit 1 +fi + +if ! LC_MESSAGES=C nmcli radio wifi | grep -qc enabled; then + echo "ERROR: NM could not track device state to enabled." + rmmod fake-rfkill || true + exit 1 +fi + +rmmod fake-rfkill +make -f debian/tests/Makefile clean-rfkill + +echo OK diff --git a/debian/tests/network_test_base.py b/debian/tests/network_test_base.py new file mode 100644 index 00000000..9ef8dec3 --- /dev/null +++ b/debian/tests/network_test_base.py @@ -0,0 +1,360 @@ +''' +Base class for network related tests. + +This provides fake wifi devices with mac80211_hwsim and hostapd, test ethernet +devices with veth, utility functions to start wpasupplicant, dnsmasq, get/set +rfkill status, and some utility functions. +''' + +__author__ = 'Martin Pitt <martin.pitt@ubuntu.com>' +__copyright__ = '(C) 2013 Canonical Ltd.' +__license__ = 'GPL v2 or later' + +import sys +import os +import os.path +import time +import tempfile +import subprocess +import re +import unittest +import traceback +import functools +from glob import glob + +# check availability of programs, and cleanly skip test if they are not +# available +for program in ['wpa_supplicant', 'hostapd', 'dnsmasq', 'dhclient']: + if subprocess.call(['which', program], stdout=subprocess.PIPE) != 0: + sys.stderr.write('%s is required for this test suite, but not available. Skipping\n' % program) + sys.exit(0) + + +class NetworkTestBase(unittest.TestCase): + '''Common functionality for network test cases + + setUp() creates two test wlan devices, one for a simulated access point + (self.dev_w_ap), the other for a simulated client device + (self.dev_w_client), and two test ethernet devices (self.dev_e_ap and + self.dev_e_client). + + Each test should call self.setup_ap() or self.setup_eth() with the desired + configuration. + ''' + @classmethod + def setUpClass(klass): + # ensure we have this so that iw works + subprocess.check_call(['modprobe', 'cfg80211']) + + # set regulatory domain "EU", so that we can use 80211.a 5 GHz channels + out = subprocess.check_output(['iw', 'reg', 'get'], universal_newlines=True) + m = re.match('^(?:global\n)?country (\S+):', out) + assert m + klass.orig_country = m.group(1) + subprocess.check_call(['iw', 'reg', 'set', 'EU']) + + @classmethod + def tearDownClass(klass): + subprocess.check_call(['iw', 'reg', 'set', klass.orig_country]) + os.remove('/run/udev/rules.d/99-nm-veth-test.rules') + + @classmethod + def create_devices(klass): + '''Create Access Point and Client devices with mac80211_hwsim and veth''' + + klass.dev_e_ap = 'veth42' + klass.dev_e_client = 'eth42' + + if os.path.exists('/sys/module/mac80211_hwsim'): + raise SystemError('mac80211_hwsim module already loaded') + if os.path.exists('/sys/class/net/' + klass.dev_e_client): + raise SystemError('%s interface already exists' % klass.dev_e_client) + + # ensure NM can manage our fake eths + os.makedirs('/run/udev/rules.d', exist_ok=True) + with open('/run/udev/rules.d/99-nm-veth-test.rules', 'w') as f: + f.write('ENV{ID_NET_DRIVER}=="veth", ENV{INTERFACE}=="%s", ENV{NM_UNMANAGED}="0"\n' % klass.dev_e_client) + subprocess.check_call(['udevadm', 'control', '--reload']) + + # create virtual ethernet devs + subprocess.check_call(['ip', 'link', 'add', 'name', klass.dev_e_client, 'type', + 'veth', 'peer', 'name', klass.dev_e_ap]) + + # create virtual wlan devs + before_wlan = set([c for c in os.listdir('/sys/class/net') if c.startswith('wlan')]) + subprocess.check_call(['modprobe', 'mac80211_hwsim']) + # wait 5 seconds for fake devices to appear + timeout = 50 + while timeout > 0: + after_wlan = set([c for c in os.listdir('/sys/class/net') if c.startswith('wlan')]) + if len(after_wlan) - len(before_wlan) >= 2: + break + timeout -= 1 + time.sleep(0.1) + else: + raise SystemError('timed out waiting for fake devices to appear') + + devs = list(after_wlan - before_wlan) + klass.dev_w_ap = devs[0] + klass.dev_w_client = devs[1] + + # determine and store MAC addresses + with open('/sys/class/net/%s/address' % klass.dev_w_ap) as f: + klass.mac_w_ap = f.read().strip().upper() + with open('/sys/class/net/%s/address' % klass.dev_w_client) as f: + klass.mac_w_client = f.read().strip().upper() + with open('/sys/class/net/%s/address' % klass.dev_e_ap) as f: + klass.mac_e_ap = f.read().strip().upper() + with open('/sys/class/net/%s/address' % klass.dev_e_client) as f: + klass.mac_e_client = f.read().strip().upper() + #print('Created fake devices: AP: %s, client: %s' % (klass.dev_w_ap, klass.dev_w_client)) + + @classmethod + def shutdown_devices(klass): + '''Remove test wlan devices''' + + subprocess.check_call(['rmmod', 'mac80211_hwsim']) + subprocess.check_call(['ip', 'link', 'del', 'dev', klass.dev_e_ap]) + klass.dev_w_ap = None + klass.dev_w_client = None + klass.dev_e_ap = None + klass.dev_e_client = None + + @classmethod + def get_rfkill(klass, interface): + '''Get rfkill status of an interface. + + Returns whether the interface is blocked, i. e. "True" for blocked, + "False" for enabled. + ''' + with open(klass._rfkill_attribute(interface)) as f: + val = f.read() + return val == '1' + + @classmethod + def set_rfkill(klass, interface, block): + '''Set rfkill status of an interface + + Use block==True for disabling ("killswitching") an interface, + block==False to re-enable. + ''' + with open(klass._rfkill_attribute(interface), 'w') as f: + f.write(block and '1' or '0') + + def run(self, result=None): + '''Show log files on failed tests''' + + if result: + orig_err_fail = len(result.errors) + len(result.failures) + super().run(result) + if hasattr(self, 'workdir'): + logs = glob(os.path.join(self.workdir, '*.log')) + if result and len(result.errors) + len(result.failures) > orig_err_fail: + for log_file in logs: + with open(log_file) as f: + print('\n----- %s -----\n%s\n------\n' + % (os.path.basename(log_file), f.read())) + + # clean up log files, so that we don't see ones from previous tests + for log_file in logs: + os.unlink(log_file) + + def setUp(self): + '''Create test devices and workdir''' + + self.create_devices() + self.addCleanup(self.shutdown_devices) + self.workdir_obj = tempfile.TemporaryDirectory() + self.workdir = self.workdir_obj.name + + # create static entropy file to avoid draining/blocking on /dev/random + self.entropy_file = os.path.join(self.workdir, 'entropy') + with open(self.entropy_file, 'wb') as f: + f.write(b'012345678901234567890') + + def setup_ap(self, hostapd_conf, ipv6_mode): + '''Set up simulated access point + + On self.dev_w_ap, run hostapd with given configuration. Setup dnsmasq + according to ipv6_mode, see start_dnsmasq(). + + This is torn down automatically at the end of the test. + ''' + # give our AP an IP + subprocess.check_call(['ip', 'a', 'flush', 'dev', self.dev_w_ap]) + if ipv6_mode is not None: + subprocess.check_call(['ip', 'a', 'add', '2600::1/64', 'dev', self.dev_w_ap]) + else: + subprocess.check_call(['ip', 'a', 'add', '192.168.5.1/24', 'dev', self.dev_w_ap]) + + self.start_hostapd(hostapd_conf) + self.start_dnsmasq(ipv6_mode, self.dev_w_ap) + + def setup_eth(self, ipv6_mode, start_dnsmasq=True): + '''Set up simulated ethernet router + + On self.dev_e_ap, run dnsmasq according to ipv6_mode, see + start_dnsmasq(). + + This is torn down automatically at the end of the test. + ''' + # give our router an IP + subprocess.check_call(['ip', 'a', 'flush', 'dev', self.dev_e_ap]) + if ipv6_mode is not None: + subprocess.check_call(['ip', 'a', 'add', '2600::1/64', 'dev', self.dev_e_ap]) + else: + subprocess.check_call(['ip', 'a', 'add', '192.168.5.1/24', 'dev', self.dev_e_ap]) + subprocess.check_call(['ip', 'link', 'set', self.dev_e_ap, 'up']) + # we don't really want to up the client iface already, but veth doesn't + # work otherwise (no link detected) + subprocess.check_call(['ip', 'link', 'set', self.dev_e_client, 'up']) + + if start_dnsmasq: + self.start_dnsmasq(ipv6_mode, self.dev_e_ap) + + def start_wpasupp(self, conf): + '''Start wpa_supplicant on client interface''' + + w_conf = os.path.join(self.workdir, 'wpasupplicant.conf') + with open(w_conf, 'w') as f: + f.write('ctrl_interface=%s\nnetwork={\n%s\n}\n' % (self.workdir, conf)) + log = os.path.join(self.workdir, 'wpasupp.log') + p = subprocess.Popen(['wpa_supplicant', '-Dwext', '-i', self.dev_w_client, + '-e', self.entropy_file, '-c', w_conf, '-f', log], + stderr=subprocess.PIPE) + self.addCleanup(p.wait) + self.addCleanup(p.terminate) + # TODO: why does this sometimes take so long? + self.poll_text(log, 'CTRL-EVENT-CONNECTED', timeout=200) + + def wrap_process(self, fn, *args, **kwargs): + '''Run a test method in a separate process. + + Run test method fn(*args, **kwargs) in a child process. If that raises + any exception, it gets propagated to the main process and + wrap_process() fails with that exception. + ''' + # exception from subprocess is propagated through this file + exc_path = os.path.join(self.workdir, 'exc') + try: + os.unlink(exc_path) + except OSError: + pass + + pid = os.fork() + + # run the actual test in the child + if pid == 0: + # short-circuit tearDownClass(), as this will be done by the parent + # process + self.addCleanup(os._exit, 0) + try: + fn(*args, **kwargs) + except: + with open(exc_path, 'w') as f: + f.write(traceback.format_exc()) + raise + else: + # get success/failure state from child + os.waitpid(pid, 0) + # propagate exception + if os.path.exists(exc_path): + with open(exc_path) as f: + self.fail(f.read()) + + # + # Internal implementation details + # + + @classmethod + def poll_text(klass, logpath, string, timeout=50): + '''Poll log file for a given string with a timeout. + + Timeout is given in deciseconds. + ''' + log = '' + while timeout > 0: + if os.path.exists(logpath): + break + timeout -= 1 + time.sleep(0.1) + assert timeout > 0, 'Timed out waiting for file %s to appear' % logpath + + with open(logpath) as f: + while timeout > 0: + line = f.readline() + if line: + log += line + if string in line: + break + continue + timeout -= 1 + time.sleep(0.1) + + assert timeout > 0, 'Timed out waiting for "%s":\n------------\n%s\n-------\n' % (string, log) + + def start_hostapd(self, conf): + hostapd_conf = os.path.join(self.workdir, 'hostapd.conf') + with open(hostapd_conf, 'w') as f: + f.write('interface=%s\ndriver=nl80211\n' % self.dev_w_ap) + f.write(conf) + + log = os.path.join(self.workdir, 'hostapd.log') + p = subprocess.Popen(['hostapd', '-e', self.entropy_file, '-f', log, hostapd_conf], + stdout=subprocess.PIPE) + self.addCleanup(p.wait) + self.addCleanup(p.terminate) + self.poll_text(log, '' + self.dev_w_ap + ': AP-ENABLED') + + def start_dnsmasq(self, ipv6_mode, iface): + '''Start dnsmasq. + + If ipv6_mode is None, IPv4 is set up with DHCP. If it is not None, it + must be a valid dnsmasq mode, i. e. a combination of "ra-only", + "slaac", "ra-stateless", and "ra-names". See dnsmasq(8). + ''' + if ipv6_mode is None: + dhcp_range = '192.168.5.10,192.168.5.200' + else: + dhcp_range = '2600::10,2600::20' + if ipv6_mode: + dhcp_range += ',' + ipv6_mode + + self.dnsmasq_log = os.path.join(self.workdir, 'dnsmasq.log') + lease_file = os.path.join(self.workdir, 'dnsmasq.leases') + + p = subprocess.Popen(['dnsmasq', '--keep-in-foreground', '--log-queries', + '--log-facility=' + self.dnsmasq_log, + '--conf-file=/dev/null', + '--dhcp-leasefile=' + lease_file, + '--bind-interfaces', + '--interface=' + iface, + '--except-interface=lo', + '--enable-ra', + '--dhcp-range=' + dhcp_range]) + self.addCleanup(p.wait) + self.addCleanup(p.terminate) + + if ipv6_mode is not None: + self.poll_text(self.dnsmasq_log, 'IPv6 router advertisement enabled') + else: + self.poll_text(self.dnsmasq_log, 'DHCP, IP range') + + @classmethod + def _rfkill_attribute(klass, interface): + '''Return the path to interface's rfkill soft toggle in sysfs.''' + + g = glob('/sys/class/net/%s/phy80211/rfkill*/soft' % interface) + assert len(g) == 1, 'Did not find exactly one "soft" rfkill attribute for %s: %s' % ( + interface, str(g)) + return g[0] + + +def run_in_subprocess(fn): + '''Decorator for running fn in a child process''' + + @functools.wraps(fn) + def wrapped(*args, **kwargs): + # args[0] is self + args[0].wrap_process(fn, *args, **kwargs) + return wrapped diff --git a/debian/tests/nm b/debian/tests/nm new file mode 100755 index 00000000..e44728f2 --- /dev/null +++ b/debian/tests/nm @@ -0,0 +1,848 @@ +#!/usr/bin/python3 +# Test NetworkManager on simulated network devices +# For an interactive shell test, run "nm ColdplugWifi.shell", see below + +__author__ = 'Martin Pitt <martin.pitt@ubuntu.com>' +__copyright__ = '(C) 2013 Canonical Ltd.' +__license__ = 'GPL v2 or later' + +import sys +import os +import os.path +import time +import subprocess +import socket +import unittest +import ctypes + +try: + from dbusmock import DBusTestCase +except ImportError: + DBusTestCase = object # dummy so that the class declaration works + +from gi.repository import NetworkManager, NMClient, GLib + +sys.path.append(os.path.dirname(__file__)) +import network_test_base + +SSID = 'fake net' + +# If True, NetworkManager logs directly to stdout, to watch logs in real time +NM_LOG_STDOUT = os.getenv('NM_LOG_STDOUT', False) + +# avoid accidentally destroying any real config +os.environ['GSETTINGS_BACKEND'] = 'memory' + +# we currently get a lot of WARNINGs/CRITICALs from GI (leaked objects from +# previous test runs/main loops?) Redirect them to stdout, to avoid failing +# autopkgtests +os.dup2(sys.stdout.fileno(), sys.stderr.fileno()) + + +class NetworkManagerTest(network_test_base.NetworkTestBase): + '''Provide common functionality for NM tests''' + + def start_nm(self, wait_iface=None, auto_connect=True): + '''Start NetworkManager and initialize client object + + If wait_iface is given, wait until NM recognizes that interface. + Otherwise, just wait until NM has initialized (for coldplug mode). + + If auto_connect is False, set the "no-auto-default=*" option to avoid + auto-connecting to wired devices. + ''' + # mount tmpfses over system directories, to avoid destroying the + # production configuration, and isolating tests from each other + if not os.path.exists('/run/NetworkManager'): + os.mkdir('/run/NetworkManager') + for d in ['/etc/NetworkManager', '/var/lib/NetworkManager', + '/run/NetworkManager']: + subprocess.check_call(['mount', '-n', '-t', 'tmpfs', 'none', d]) + self.addCleanup(subprocess.call, ['umount', d]) + os.mkdir('/etc/NetworkManager/system-connections') + + # create local configuration; this allows us to have full control, and + # we also need to blacklist the AP device so that NM does not tear it + # down; we also blacklist any existing real interface to avoid + # interfering with it, and for getting predictable results + blacklist = '' + for iface in os.listdir('/sys/class/net'): + if iface == "bonding_masters": + continue + if iface != self.dev_w_client and iface != self.dev_e_client: + with open('/sys/class/net/%s/address' % iface) as f: + if blacklist: + blacklist += ';' + blacklist += 'mac:%s' % f.read().strip() + + conf = os.path.join(self.workdir, 'NetworkManager.conf') + extra_main = '' + if not auto_connect: + extra_main += 'no-auto-default=*\n' + + with open(conf, 'w') as f: + f.write('[main]\nplugins=keyfile\n%s\n[keyfile]\nunmanaged-devices=%s\n' % + (extra_main, blacklist)) + + if NM_LOG_STDOUT: + f_log = None + else: + log = os.path.join(self.workdir, 'NetworkManager.log') + f_log = os.open(log, os.O_CREAT | os.O_WRONLY | os.O_SYNC) + + # build NM command line + argv = ['NetworkManager', '--log-level=debug', '--debug', '--config=' + conf] + # allow specifying extra arguments + argv += os.environ.get('NM_TEST_DAEMON_ARGS', '').strip().split() + + p = subprocess.Popen(argv, stdout=f_log, stderr=subprocess.STDOUT) + # automatically terminate process at end of test case + self.addCleanup(p.wait) + self.addCleanup(p.terminate) + self.addCleanup(self.shutdown_connections) + + if NM_LOG_STDOUT: + # let it initialize, then print a marker + time.sleep(1) + print('******* NM initialized *********\n\n') + else: + self.addCleanup(os.close, f_log) + + # this should be fast, give it 2 s to initialize + if wait_iface: + self.poll_text(log, 'manager: (%s): new' % wait_iface, timeout=100) + + self.nmclient = NMClient.Client.new() + self.assertTrue(self.nmclient.networking_get_enabled()) + + # FIXME: This certainly ought to be true, but isn't + #self.assertTrue(self.nmclient.get_manager_running()) + + # determine device objects + for d in self.nmclient.get_devices(): + if d.props.interface == self.dev_w_ap: + self.assertEqual(d.get_device_type(), NetworkManager.DeviceType.WIFI) + self.assertEqual(d.get_driver(), 'mac80211_hwsim') + self.assertEqual(d.get_hw_address(), self.mac_w_ap) + self.nmdev_w_ap = d + elif d.props.interface == self.dev_w_client: + self.assertEqual(d.get_device_type(), NetworkManager.DeviceType.WIFI) + self.assertEqual(d.get_driver(), 'mac80211_hwsim') + # NM ≥ 1.4 randomizes MAC addresses by default, so we can't + # test for equality, just make sure it's not our AP + self.assertNotEqual(d.get_hw_address(), self.mac_w_ap) + self.nmdev_w = d + elif d.props.interface == self.dev_e_client: + self.assertEqual(d.get_device_type(), NetworkManager.DeviceType.ETHERNET) + self.assertEqual(d.get_driver(), 'veth') + self.assertEqual(d.get_hw_address(), self.mac_e_client) + self.nmdev_e = d + + self.assertTrue(hasattr(self, 'nmdev_w_ap'), 'Could not determine wifi AP NM device') + self.assertTrue(hasattr(self, 'nmdev_w'), 'Could not determine wifi client NM device') + self.assertTrue(hasattr(self, 'nmdev_e'), 'Could not determine eth client NM device') + + self.process_glib_events() + + def shutdown_connections(self): + '''Shut down all active NM connections.''' + + if NM_LOG_STDOUT: + print('\n\n******* Shutting down NM connections *********') + + # remove all created connections + for active_conn in self.nmclient.get_active_connections(): + self.nmclient.deactivate_connection(active_conn) + self.assertEventually(lambda: self.nmclient.get_active_connections() == [], + timeout=20) + + # verify that NM properly deconfigures the devices + self.assert_iface_down(self.dev_w_client) + self.assert_iface_down(self.dev_e_client) + + @classmethod + def process_glib_events(klass): + '''Process pending GLib main loop events''' + + context = GLib.MainContext.default() + while context.iteration(False): + pass + + def assertEventually(self, condition, message=None, timeout=50): + '''Assert that condition function eventually returns True. + + timeout is in deciseconds, defaulting to 50 (5 seconds). message is + printed on failure. + ''' + while timeout >= 0: + self.process_glib_events() + if condition(): + break + timeout -= 1 + time.sleep(0.1) + else: + self.fail(message or 'timed out waiting for ' + str(condition)) + + def assert_iface_down(self, iface): + '''Assert that client interface is down''' + + out = subprocess.check_output(['ip', 'a', 'show', 'dev', iface], + universal_newlines=True) + self.assertIn('state DOWN', out) + self.assertNotIn('inet 192', out) + self.assertNotIn('inet6 2600', out) + + if iface == self.dev_w_client: + out = subprocess.check_output(['iw', 'dev', iface, 'link'], + universal_newlines=True) + self.assertIn('Not connected', out) + + # but AP device should never be touched by NM + out = subprocess.check_output(['ip', 'a', 'show', 'dev', self.dev_w_ap], + universal_newlines=True) + self.assertIn('state UP', out) + + def assert_iface_up(self, iface, expected_ip_a=None, unexpected_ip_a=None): + '''Assert that client interface is up''' + + out = subprocess.check_output(['ip', 'a', 'show', 'dev', iface], + universal_newlines=True) + self.assertIn('state UP', out) + if expected_ip_a: + for r in expected_ip_a: + self.assertRegex(out, r) + if unexpected_ip_a: + for r in unexpected_ip_a: + self.assertNotRegex(out, r) + + if iface == self.dev_w_client: + out = subprocess.check_output(['iw', 'dev', iface, 'link'], + universal_newlines=True) + self.assertIn('Connected to ' + self.mac_w_ap, out) + self.assertIn('SSID: ' + SSID, out) + + def wait_ap(self, timeout): + '''Wait for AccessPoint NM object to appear, and return it''' + + self.assertEventually(lambda: len(self.nmdev_w.get_access_points()) > 0, + 'timed out waiting for AP to be detected', + timeout=timeout) + + return self.nmdev_w.get_access_points()[0] + + def connect_to_ap(self, ap, secret, ipv6_mode, ip6_privacy): + '''Connect to an NMAccessPoint. + + secret should be None for open networks, and a string with the password + for WEP/WPA. + + ip6_privacy is a NetworkManager.SettingIP6ConfigPrivacy flag. + + Return (NMConnection, NMActiveConnection) objects. + ''' + + ip4_method = NetworkManager.SETTING_IP4_CONFIG_METHOD_DISABLED + ip6_method = NetworkManager.SETTING_IP6_CONFIG_METHOD_IGNORE + if ipv6_mode is None: + ip4_method = NetworkManager.SETTING_IP4_CONFIG_METHOD_AUTO + else: + ip6_method = NetworkManager.SETTING_IP6_CONFIG_METHOD_AUTO + + # If we have a secret, supply it to the new connection right away; + # adding it afterwards with update_secrets() does not work, and we + # can't implement a SecretAgent as get_secrets() would need to build a + # map of a map of gpointers to gpointers which is too much for PyGI + partial_conn = NetworkManager.Connection.new() + partial_conn.add_setting(NetworkManager.SettingIP4Config(method=ip4_method)) + if secret: + partial_conn.add_setting(NetworkManager.SettingWirelessSecurity.new()) + # FIXME: needs update for other auth types + partial_conn.update_secrets(NetworkManager.SETTING_WIRELESS_SECURITY_SETTING_NAME, + {'psk': secret}) + if ip6_privacy is not None: + partial_conn.add_setting(NetworkManager.SettingIP6Config(ip6_privacy=ip6_privacy, + method=ip6_method)) + + ml = GLib.MainLoop() + self.cb_conn = None + + def add_activate_cb(client, conn, conn_path, error, data): + self.cb_conn = conn + self.cb_error = error + ml.quit() + self.nmclient.add_and_activate_connection(partial_conn, self.nmdev_w, ap.get_path(), add_activate_cb, None) + ml.run() + self.assertEqual(self.cb_error, None) + active_conn = self.cb_conn + self.cb_conn = None + + conn = self.conn_from_active_conn(active_conn) + self.assertTrue(conn.verify()) + + # verify need_secrets() + needed_secrets = conn.need_secrets() + if secret is None: + self.assertEqual(needed_secrets, (None, [])) + else: + self.assertEqual(needed_secrets[0], NetworkManager.SETTING_WIRELESS_SECURITY_SETTING_NAME) + self.assertEqual(type(needed_secrets[1]), list) + self.assertGreaterEqual(len(needed_secrets[1]), 1) + # FIXME: needs update for other auth types + self.assertIn(needed_secrets[1][0], [NetworkManager.SETTING_WIRELESS_SECURITY_PSK]) + + # we are usually ACTIVATING at this point; wait for completion + # TODO: 5s is not enough, argh slow DHCP client + self.assertEventually(lambda: active_conn.get_state() == NetworkManager.ActiveConnectionState.ACTIVATED, + 'timed out waiting for %s to get activated' % active_conn.get_connection(), + timeout=600) + self.assertEqual(self.nmdev_w.get_state(), NetworkManager.DeviceState.ACTIVATED) + return (conn, active_conn) + + def conn_from_active_conn(self, active_conn): + '''Get NMConnection object for an NMActiveConnection object''' + + # this sometimes takes a second try, when the corresponding + # NMConnection object is not yet available + tries = 3 + while tries > 0: + self.process_glib_events() + path = active_conn.get_connection() + for dev in active_conn.get_devices(): + for c in dev.get_available_connections(): + if c.get_path() == path: + return c + time.sleep(0.1) + tries -= 1 + + self.fail('Could not find NMConnection object for %s' % path) + + def check_connected_device_config(self, ipv6_mode, nmdev): + '''Check NMDevice configuration state after being connected''' + + time.sleep(10) + if ipv6_mode is not None: + # FIXME: why do we need to wait here, if state is already ACTIVATED? + self.assertEventually(lambda: nmdev.get_ip6_config() is not None, timeout=50) + #self.assertEqual(nmdev.get_ip4_config(), None) + conf = nmdev.get_ip6_config() + self.assertNotEqual(conf, None) + # we expect at least a link-local and a RA prefix or DHCP assigned + # address + addrs = conf.get_addresses() + self.assertGreaterEqual(len(addrs), 2, [a.get_address() for a in addrs]) + # note, we cannot call IP6Address.get_address(), as that returns a + # raw gpointer; check address with low-level tools only + else: + # FIXME: why do we need to wait here, if state is already ACTIVATED? + self.assertEventually(lambda: nmdev.get_ip4_config() is not None, timeout=50) + conf = nmdev.get_ip4_config() + self.assertNotEqual(conf, None) + self.assertEqual(len(conf.get_addresses()), 1) + self.assertEqual(socket.ntohl(conf.get_addresses()[0].get_address()) & 0xFFFFFF00, + 0xC0A80500) # 192.168.5.x + + def check_low_level_config(self, iface, ipv6_mode, ip6_privacy): + '''Check actual hardware state with ip/iw after being connected''' + + # list of expected regexps in "ip a" output + expected_ip_a = [] + unexpected_ip_a = [] + + if ipv6_mode is not None: + if ipv6_mode in ('', 'slaac'): + # has global address from our DHCP server + expected_ip_a.append('inet6 2600::[0-9a-f]+/') + else: + # has address with our prefix and MAC + expected_ip_a.append('inet6 2600::[0-9a-f:]+/64 scope global (?:tentative )?(?:mngtmpaddr )?(?:noprefixroute )?dynamic') + # has address with our prefix and random IP (Privacy + # Extension), if requested + priv_re = 'inet6 2600:[0-9a-f:]+/64 scope global temporary (?:tentative )?(?:mngtmpaddr )?dynamic' + if ip6_privacy in (NetworkManager.SettingIP6ConfigPrivacy.PREFER_TEMP_ADDR, + NetworkManager.SettingIP6ConfigPrivacy.PREFER_PUBLIC_ADDR): + expected_ip_a.append(priv_re) + else: + # FIXME: add a negative test here + pass + #unexpected_ip_a.append(priv_re) + + # has a link-local address + expected_ip_a.append('inet6 fe80::[0-9a-f:]+/64 scope link') + else: + expected_ip_a.append('inet 192.168.5.\d+/24') + + self.assert_iface_up(iface, expected_ip_a, unexpected_ip_a) + + +class ColdplugWifi(NetworkManagerTest): + '''Wifi: In these tests NM starts after setting up the AP''' + + # not run by default; run "nm-wifi ColdplugWifi.shell" to get this + @network_test_base.run_in_subprocess + def shell(self): + '''Start AP and NM, then run a shell (for debugging)''' + + self.setup_ap('hw_mode=b\nchannel=1\nssid=' + SSID, None) + self.start_nm(self.dev_w_client) + print(''' + +client interface: %s, access point interface: %s, AP SSID: "%s" + +You can now run commands like "nmcli dev" or "nmcli dev wifi connect '%s'". +Logs are in '%s'. When done, exit the shell. + +''' % (self.dev_w_client, self.dev_w_ap, SSID, SSID, self.workdir)) + subprocess.call(['bash', '-i']) + + @network_test_base.run_in_subprocess + def test_no_ap(self): + '''no available access point''' + + self.start_nm(self.dev_w_client) + self.assertEventually(self.nmclient.networking_get_enabled, timeout=20) + + # state independent properties + self.assertEqual(self.nmdev_w.props.device_type, NetworkManager.DeviceType.WIFI) + self.assertTrue(self.nmdev_w.props.managed) + self.assertFalse(self.nmdev_w.props.firmware_missing) + self.assertTrue(self.nmdev_w.props.udi.startswith('/sys/devices/'), self.nmdev_w.props.udi) + + # get_version() plausibility check + out = subprocess.check_output(['nmcli', '--version'], universal_newlines=True) + cli_version = out.split()[-1] + self.assertTrue(cli_version[0].isdigit()) + self.assertEqual(self.nmclient.get_version(), cli_version) + + # state dependent properties (disconnected) + self.assertIn(self.nmdev_w.get_state(), + [NetworkManager.DeviceState.DISCONNECTED, NetworkManager.DeviceState.UNAVAILABLE]) + self.assertEqual(self.nmdev_w.get_access_points(), []) + self.assertEqual(self.nmdev_w.get_available_connections(), []) + + def test_open_b_ip4(self): + '''Open network, 802.11b, IPv4''' + + self.do_test('hw_mode=b\nchannel=1\nssid=' + SSID, None, 11000) + + def test_open_b_ip6_raonly_tmpaddr(self): + '''Open network, 802.11b, IPv6 with only RA, preferring temp address''' + + self.do_test('hw_mode=b\nchannel=1\nssid=' + SSID, 'ra-only', 11000, + ip6_privacy=NetworkManager.SettingIP6ConfigPrivacy.PREFER_TEMP_ADDR) + + def test_open_b_ip6_raonly_pubaddr(self): + '''Open network, 802.11b, IPv6 with only RA, preferring public address''' + + self.do_test('hw_mode=b\nchannel=1\nssid=' + SSID, 'ra-only', 11000, + ip6_privacy=NetworkManager.SettingIP6ConfigPrivacy.PREFER_PUBLIC_ADDR) + + def test_open_b_ip6_raonly_no_pe(self): + '''Open network, 802.11b, IPv6 with only RA, PE disabled''' + + self.do_test('hw_mode=b\nchannel=1\nssid=' + SSID, 'ra-only', 11000, + ip6_privacy=NetworkManager.SettingIP6ConfigPrivacy.DISABLED) + + def test_open_b_ip6_dhcp(self): + '''Open network, 802.11b, IPv6 with DHCP, preferring temp address''' + + self.do_test('hw_mode=b\nchannel=1\nssid=' + SSID, '', 11000, + ip6_privacy=NetworkManager.SettingIP6ConfigPrivacy.UNKNOWN) + + def test_open_g_ip4(self): + '''Open network, 802.11g, IPv4''' + + self.do_test('hw_mode=g\nchannel=1\nssid=' + SSID, None, 54000) + + def test_wpa1_ip4(self): + '''WPA1, 802.11g, IPv4''' + + self.do_test('''hw_mode=g +channel=1 +ssid=%s +wpa=1 +wpa_key_mgmt=WPA-PSK +wpa_pairwise=TKIP +wpa_passphrase=12345678 +''' % SSID, None, 54000, '12345678') + + def test_wpa2_ip4(self): + '''WPA2, 802.11g, IPv4''' + + self.do_test('''hw_mode=g +channel=1 +ssid=%s +wpa=2 +wpa_key_mgmt=WPA-PSK +wpa_pairwise=CCMP +wpa_passphrase=12345678 +''' % SSID, None, 54000, '12345678') + + def test_wpa2_ip6(self): + '''WPA2, 802.11g, IPv6 with only RA''' + + self.do_test('''hw_mode=g +channel=1 +ssid=%s +wpa=2 +wpa_key_mgmt=WPA-PSK +wpa_pairwise=CCMP +wpa_passphrase=12345678 +''' % SSID, 'ra-only', 54000, '12345678', + ip6_privacy=NetworkManager.SettingIP6ConfigPrivacy.PREFER_TEMP_ADDR) + + @network_test_base.run_in_subprocess + def test_rfkill(self): + '''shut down connection on killswitch, restore it on unblock''' + + self.setup_ap('hw_mode=b\nchannel=1\nssid=' + SSID, None) + self.start_nm(self.dev_w_client) + ap = self.wait_ap(timeout=1800) + (conn, active_conn) = self.connect_to_ap(ap, None, None, None) + + self.assertFalse(self.get_rfkill(self.dev_w_client)) + self.assertFalse(self.get_rfkill(self.dev_w_ap)) + + # now block the client interface + self.set_rfkill(self.dev_w_client, True) + # disabling should be fast, give it ten seconds + self.assertEventually(lambda: self.nmdev_w.get_state() == NetworkManager.DeviceState.UNAVAILABLE, + timeout=100) + + # dev_w_client should be down now + self.assert_iface_down(self.dev_w_client) + + # turn it back on + self.set_rfkill(self.dev_w_client, False) + # this involves DHCP, use same timeout as for regular connection + self.assertEventually(lambda: self.nmdev_w.get_state() == NetworkManager.DeviceState.ACTIVATED, + timeout=200) + + # dev_w_client should be back up + self.assert_iface_up(self.dev_w_client, ['inet 192.168.5.\d+/24']) + + # + # Common test code + # + + # libnm-glib has a lot of internal persistent state (private D-BUS + # connections and such); as it is very brittle and hard to track down + # all remaining references to any NM* object after a test, we rather + # run each test in a separate subprocess + @network_test_base.run_in_subprocess + def do_test(self, hostapd_conf, ipv6_mode, expected_max_bitrate, + secret=None, ip6_privacy=None): + '''Actual test code, parameterized for the particular test case''' + + self.setup_ap(hostapd_conf, ipv6_mode) + self.start_nm(self.dev_w_client) + + # on coldplug we expect the AP to be picked out fast + ap = self.wait_ap(timeout=100) + self.assertTrue(ap.get_path().startswith('/org/freedesktop/NetworkManager')) + self.assertEqual(ap.get_mode(), getattr(NetworkManager, '80211Mode').INFRA) + self.assertEqual(ap.get_max_bitrate(), expected_max_bitrate) + #self.assertEqual(ap.get_flags(), ) + + # should not auto-connect + self.assertEqual(self.nmclient.get_active_connections(), []) + + # connect to that AP + (conn, active_conn) = self.connect_to_ap(ap, secret, ipv6_mode, ip6_privacy) + + # check NMActiveConnection object + self.assertIn(active_conn.get_uuid(), [c.get_uuid() for c in self.nmclient.get_active_connections()]) + self.assertEqual([d.get_udi() for d in active_conn.get_devices()], [self.nmdev_w.get_udi()]) + + self.check_connected_device_config(ipv6_mode, self.nmdev_w) + + # check corresponding NMConnection object + wireless_setting = conn.get_setting_wireless() + self.assertEqual(wireless_setting.get_ssid(), SSID.encode()) + self.assertEqual(wireless_setting.get_hidden(), False) + if secret: + self.assertEqual(wireless_setting.get_security(), NetworkManager.SETTING_WIRELESS_SECURITY_SETTING_NAME) + else: + self.assertEqual(wireless_setting.get_security(), None) + # for debugging + #conn.dump() + + # for IPv6, check privacy setting + if ipv6_mode is not None and ip6_privacy != NetworkManager.SettingIP6ConfigPrivacy.UNKNOWN: + assert ip6_privacy is not None, 'for IPv6 tests you need to specify ip6_privacy flag' + ip6_setting = conn.get_setting_ip6_config() + self.assertEqual(ip6_setting.props.ip6_privacy, ip6_privacy) + + self.check_low_level_config(self.dev_w_client, ipv6_mode, ip6_privacy) + + +class ColdplugEthernet(NetworkManagerTest): + '''Ethernet: In these tests NM starts after setting up the router''' + + # not run by default; run "nm-wifi ColdplugEthernet.shell" to get this + @network_test_base.run_in_subprocess + def shell(self): + '''Start router and NM, then run a shell (for debugging)''' + + self.setup_eth(None) + self.start_nm(self.dev_e_client) + print(''' + +client interface: %s, router interface: %s + +You can now run commands like "nmcli dev". +Logs are in '%s'. When done, exit the shell. + +''' % (self.dev_e_client, self.dev_e_ap, self.workdir)) + subprocess.call(['bash', '-i']) + + def test_auto_ip4(self): + '''ethernet: auto-connection, IPv4''' + + self.do_test(None, auto_connect=True) + + def test_auto_ip6_raonly_no_pe(self): + '''ethernet: auto-connection, IPv6 with only RA, PE disabled''' + + self.do_test('ra-only', auto_connect=True, + ip6_privacy=NetworkManager.SettingIP6ConfigPrivacy.DISABLED) + + def test_auto_ip6_dhcp(self): + '''ethernet: auto-connection, IPv6 with DHCP''' + + self.do_test('', auto_connect=True, + ip6_privacy=NetworkManager.SettingIP6ConfigPrivacy.UNKNOWN) + + def test_manual_ip4(self): + '''ethernet: manual connection, IPv4''' + + self.do_test(None, auto_connect=False) + + def test_manual_ip6_raonly_tmpaddr(self): + '''ethernet: manual connection, IPv6 with only RA, preferring temp address''' + + self.do_test('ra-only', auto_connect=False, + ip6_privacy=NetworkManager.SettingIP6ConfigPrivacy.PREFER_TEMP_ADDR) + + def test_manual_ip6_raonly_pubaddr(self): + '''ethernet: manual connection, IPv6 with only RA, preferring public address''' + + self.do_test('ra-only', auto_connect=False, + ip6_privacy=NetworkManager.SettingIP6ConfigPrivacy.PREFER_PUBLIC_ADDR) + + # + # Common test code + # + + @network_test_base.run_in_subprocess + def do_test(self, ipv6_mode, ip6_privacy=None, auto_connect=True): + '''Actual test code, parameterized for the particular test case''' + + self.setup_eth(ipv6_mode) + self.start_nm(self.dev_e_client, auto_connect=auto_connect) + + ip4_method = NetworkManager.SETTING_IP4_CONFIG_METHOD_DISABLED + ip6_method = NetworkManager.SETTING_IP6_CONFIG_METHOD_IGNORE + if ipv6_mode is None: + ip4_method = NetworkManager.SETTING_IP4_CONFIG_METHOD_AUTO + else: + ip6_method = NetworkManager.SETTING_IP6_CONFIG_METHOD_AUTO + + if auto_connect: + # ethernet should auto-connect quickly without an existing defined connection + self.assertEventually(lambda: len(self.nmclient.get_active_connections()) > 0, + timeout=100) + active_conn = self.nmclient.get_active_connections()[0] + else: + # auto-connection was disabled, set up manual connection + partial_conn = NetworkManager.Connection.new() + partial_conn.add_setting(NetworkManager.SettingIP4Config(method=ip4_method)) + if ip6_privacy is not None: + partial_conn.add_setting(NetworkManager.SettingIP6Config(ip6_privacy=ip6_privacy, + method=ip6_method)) + + ml = GLib.MainLoop() + self.cb_conn = None + + def add_activate_cb(client, conn, conn_path, error, data): + self.cb_conn = conn + self.cb_error = error + ml.quit() + self.nmclient.add_and_activate_connection(partial_conn, self.nmdev_e, None, add_activate_cb, None) + ml.run() + self.assertEqual(self.cb_error, None) + active_conn = self.cb_conn + self.cb_conn = None + + # we are usually ACTIVATING at this point; wait for completion + # TODO: 5s is not enough, argh slow DHCP client + self.assertEventually(lambda: active_conn.get_state() == NetworkManager.ActiveConnectionState.ACTIVATED, + 'timed out waiting for %s to get activated' % active_conn.get_connection(), + timeout=150) + self.assertEqual(self.nmdev_e.get_state(), NetworkManager.DeviceState.ACTIVATED) + + conn = self.conn_from_active_conn(active_conn) + self.assertTrue(conn.verify()) + + # check NMActiveConnection object + self.assertIn(active_conn.get_uuid(), [c.get_uuid() for c in self.nmclient.get_active_connections()]) + self.assertEqual([d.get_udi() for d in active_conn.get_devices()], [self.nmdev_e.get_udi()]) + + self.check_connected_device_config(ipv6_mode, self.nmdev_e) + + # for IPv6, check privacy setting + if ipv6_mode is not None: + assert ip6_privacy is not None, 'for IPv6 tests you need to specify ip6_privacy flag' + if ip6_privacy not in (NetworkManager.SettingIP6ConfigPrivacy.UNKNOWN, + NetworkManager.SettingIP6ConfigPrivacy.DISABLED): + ip6_setting = conn.get_setting_ip6_config() + self.assertEqual(ip6_setting.props.ip6_privacy, ip6_privacy) + + self.check_low_level_config(self.dev_e_client, ipv6_mode, ip6_privacy) + + +class Hotplug(NetworkManagerTest): + '''In these tests APs are set up while NM is already running''' + + @network_test_base.run_in_subprocess + @unittest.expectedFailure + def test_auto_detect_ap(self): + '''new AP is being detected automatically within 30s''' + + self.setup_ap('hw_mode=b\nchannel=1\nssid=' + SSID, None) + self.start_nm() + ap = self.wait_ap(timeout=300) + # get_ssid returns a byte array + self.assertEqual(ap.get_ssid(), SSID.encode()) + self.assertEqual(self.nmdev_w.get_active_access_point(), None) + + @network_test_base.run_in_subprocess + @unittest.expectedFailure + def test_auto_detect_eth(self): + '''new eth router is being detected automatically within 30s''' + + self.start_nm() + self.setup_eth(None) + self.assertEventually(lambda: len(self.nmclient.get_active_connections()) > 0, + timeout=300) + active_conn = self.nmclient.get_active_connections()[0] + + self.assertEventually(lambda: active_conn.get_state() == NetworkManager.ActiveConnectionState.ACTIVATED, + 'timed out waiting for %s to get activated' % active_conn.get_connection(), + timeout=80) + self.assertEqual(self.nmdev_e.get_state(), NetworkManager.DeviceState.ACTIVATED) + + conn = self.conn_from_active_conn(active_conn) + self.assertTrue(conn.verify()) + + +@unittest.skipIf(DBusTestCase is object, + 'WARNING: python-dbusmock not installed, skipping suspend tests; get it from https://pypi.python.org/pypi/python-dbusmock') +class Suspend(NetworkManagerTest, DBusTestCase): + '''These tests run under a mock logind on a private system D-BUS''' + + @classmethod + def setUpClass(klass): + klass.start_system_bus() + NetworkManagerTest.setUpClass() + + @classmethod + def tearDownClass(klass): + NetworkManagerTest.tearDownClass() + DBusTestCase.tearDownClass() + + def setUp(self): + NetworkManagerTest.setUp(self) + + # start mock polkit and logind processes, so that we can + # intercept/control suspend + (p_polkit, self.obj_polkit) = self.spawn_server_template('polkitd', {}, stdout=subprocess.PIPE) + # by default we are not concerned about restricting access in the tests + self.obj_polkit.AllowUnknown(True) + self.addCleanup(p_polkit.wait) + self.addCleanup(p_polkit.terminate) + + (p_logind, self.obj_logind) = self.spawn_server_template('logind', {}, stdout=subprocess.PIPE) + self.addCleanup(p_logind.wait) + self.addCleanup(p_logind.terminate) + + # we have to manually start wpa_supplicant, as D-BUS activation does + # not happen for the fake D-BUS + log = os.path.join(self.workdir, 'wpasupplicant.log') + p_wpasupp = subprocess.Popen(['wpa_supplicant', '-u', '-d', '-e', '-K', + self.entropy_file, '-f', log]) + self.addCleanup(p_wpasupp.wait) + self.addCleanup(p_wpasupp.terminate) + + def fixme_test_active_ip4(self): + '''suspend during active IPv4 connection''' + + self.do_test('hw_mode=b\nchannel=1\nssid=' + SSID, None, + ['inet 192.168.5.\d+/24']) + + def fixme_test_active_ip6(self): + '''suspend during active IPv6 connection''' + + self.do_test('hw_mode=b\nchannel=1\nssid=' + SSID, 'ra-only', + ['inet6 2600::']) + + # + # Common test code + # + + @network_test_base.run_in_subprocess + def do_test(self, hostapd_conf, ipv6_mode, expected_ip_a): + '''Actual test code, parameterized for the particular test case''' + + self.setup_ap(hostapd_conf, ipv6_mode) + self.start_nm(self.dev_w_client) + ap = self.wait_ap(timeout=1800) + (conn, active_conn) = self.connect_to_ap(ap, None, ipv6_mode, None) + + # send logind signal that we are about to suspend + self.obj_logind.EmitSignal('', 'PrepareForSleep', 'b', [True]) + + # disabling should be fast, give it one second + self.assertEventually(lambda: self.nmdev_w.get_state() == NetworkManager.DeviceState.UNMANAGED, + timeout=10) + self.assert_iface_down(self.dev_w_client) + + # send logind signal that we resumed + self.obj_logind.EmitSignal('', 'PrepareForSleep', 'b', [False]) + + # this involves DHCP, use same timeout as for regular connection + self.assertEventually(lambda: self.nmdev_w.get_state() == NetworkManager.DeviceState.ACTIVATED, + timeout=100) + + # dev_w_client should be back up + self.assert_iface_up(self.dev_w_client, expected_ip_a) + + +# avoid unintelligible error messages, and breaking "make check" when not being +# root +if os.getuid() != 0: + sys.stderr.write('This integration test suite needs to be run as root\n') + sys.exit(1) + +# AppArmor currently does not allow us to access the system D-BUS from an +# unshared file system. Hack the policy to allow that until that gets fixed +# properly. See https://launchpad.net/bugs/1244157 +subprocess.check_call("sed '/nm-dhcp-client.action {/ s/{/flags=(attach_disconnected) {/'" + " /etc/apparmor.d/sbin.dhclient > $ADTTMP/sbin.dhclient", + shell=True) +subprocess.check_call('apparmor_parser -Kr $ADTTMP/sbin.dhclient', shell=True) + +# unshare the mount namespace, so that our tmpfs mounts are guaranteed to get +# cleaned up, and don't influence the production system +libc6 = ctypes.cdll.LoadLibrary('libc.so.6') +assert libc6.unshare(ctypes.c_int(0x00020000)) == 0, 'failed to unshare mount namespace' + +# stop system-wide NetworkManager to avoid interfering with tests +nm_running = subprocess.call('service network-manager stop 2>&1', shell=True) == 0 + +# write to stdout, not stderr +runner = unittest.TextTestRunner(stream=sys.stdout, verbosity=2) +try: + unittest.main(testRunner=runner) +finally: + subprocess.call('dhclient eth0', shell=True) + subprocess.call('sleep 10', shell=True) diff --git a/debian/tests/urfkill-integration b/debian/tests/urfkill-integration new file mode 100644 index 00000000..32c009f9 --- /dev/null +++ b/debian/tests/urfkill-integration @@ -0,0 +1,103 @@ +#!/bin/sh + +set -e + +echo "+++ Building / adding fake-rfkill.ko" +make -f debian/tests/Makefile fake-rfkill +# poor man's dependency resolver +DEPS=$(modinfo debian/tests/fake-rfkill.ko | sed -n '/depends:/ {s/^.*://; s/[[:space:]]*$//; p}') +[ -z "$DEPS" ] || modprobe "$DEPS" +insmod debian/tests/fake-rfkill.ko + +fake_id=$(rfkill list | grep fake | awk -F: '{ print $1; }') +echo "+++ fake-rfkill.ko is device $fake_id" +echo + + +echo "--- Testing killswitch bringup to match NM state: when URFKILL doesn't run" +echo "+++ stopping urfkill" +# ignore failure, only here, because urfkill probably isn't running yet. +service urfkill stop || true +echo "+++ unblocking device $fake_id" +rfkill unblock $fake_id +echo "+++ stopping network-manager" +service network-manager stop || true +echo "+++ blocking device $fake_id" +rfkill block $fake_id +echo "+++ starting network-manager" +service network-manager start + +echo "+++ Waiting for the devices to settle" +sleep 30 + +echo -n "=== NetworkManager state should now be \"enabled\": " +LC_MESSAGES=C nmcli radio wifi +LC_MESSAGES=C nmcli radio wifi | grep -qc enabled +echo + +echo -n "=== NM saved state: " +grep WirelessEnabled /var/lib/NetworkManager/NetworkManager.state +grep -qc WirelessEnabled=true /var/lib/NetworkManager/NetworkManager.state +echo +echo + +echo "--- Testing killswitch bringup when URFKILL is running: follow URfkill signals" +echo "+++ stopping network-manager" +service network-manager stop +echo "+++ starting urfkill" +service urfkill start +sleep 15 +echo "+++ blocking device $fake_id" +rfkill block $fake_id +rfkill list +sleep 5 +echo "+++ starting network-manager" +service network-manager start + +echo "+++ Waiting for the devices to settle" +sleep 30 + +echo -n "=== NetworkManager state should now be \"disabled\": " +LC_MESSAGES=C nmcli radio wifi +LC_MESSAGES=C nmcli radio wifi | grep -qc disabled +echo + +echo -n "=== NM saved state: " +grep WirelessEnabled /var/lib/NetworkManager/NetworkManager.state +#grep -qc WirelessEnabled=false /var/lib/NetworkManager/NetworkManager.state +echo +echo + +echo "+++ Asking urfkill to unblock device $fake_id" +dbus-send --print-reply --system --dest=org.freedesktop.URfkill /org/freedesktop/URfkill org.freedesktop.URfkill.BlockIdx uint32:$fake_id boolean:false +sleep 5 +echo -n "=== NetworkManager state should now be \"enabled\": " +LC_MESSAGES=C nmcli radio wifi +LC_MESSAGES=C nmcli radio wifi | grep -qc enabled +echo + +echo "+++ Asking urfkill to block device $fake_id again" +dbus-send --print-reply --system --dest=org.freedesktop.URfkill /org/freedesktop/URfkill org.freedesktop.URfkill.BlockIdx uint32:$fake_id boolean:true +sleep 5 +echo -n "=== NetworkManager state should now be \"disabled\": " +LC_MESSAGES=C nmcli radio wifi +LC_MESSAGES=C nmcli radio wifi | grep -qc disabled +echo +echo + +echo "--- Removing fake-rfkill, aggregate state should get back to enabled" +rmmod fake-rfkill +sleep 5 +echo -n "=== Checking that the fake device $fake_id has disappeared: " +rfkill list | ( ! grep -qc fake || exit 1 ) && echo yes || echo no + +echo -n "=== NetworkManager state should now be \"enabled\": " +LC_MESSAGES=C nmcli radio wifi +LC_MESSAGES=C nmcli radio wifi | grep -qc enabled +echo + +#cleanup +make -f debian/tests/Makefile clean-rfkill + +echo OK +exit 0 diff --git a/debian/tests/wpa-dhclient b/debian/tests/wpa-dhclient new file mode 100755 index 00000000..a9235f97 --- /dev/null +++ b/debian/tests/wpa-dhclient @@ -0,0 +1,239 @@ +#!/usr/bin/python3 +# Test wpa_supplicant and dhclient in various modes + +__author__ = 'Martin Pitt <martin.pitt@ubuntu.com>' +__copyright__ = '(C) 2013 Canonical Ltd.' +__license__ = 'GPL v2 or later' + +import sys +import os +import os.path +import time +import subprocess +import unittest + +sys.path.append(os.path.dirname(__file__)) +import network_test_base + +SSID = 'fake net' + + +class T(network_test_base.NetworkTestBase): + @unittest.expectedFailure + def test_open_a_ip4(self): + '''Open network, 802.11a, IPv4''' + + # channel 36 ought to work everywhere + self.do_test('hw_mode=a\nchannel=36\n\nssid=' + SSID, + 'ssid="%s"\nkey_mgmt=NONE' % SSID, + None, + ['54.0']) + + @unittest.expectedFailure + def test_open_a_ip6(self): + '''Open network, 802.11a, IPv6''' + + self.do_test('hw_mode=a\nchannel=36\n\nssid=' + SSID, + 'ssid="%s"\nkey_mgmt=NONE' % SSID, + '', + ['54.0']) + + def test_open_b_ip4(self): + '''Open network, 802.11b, IPv4''' + + self.do_test('hw_mode=b\nchannel=1\nssid=' + SSID, + 'ssid="%s"\nkey_mgmt=NONE' % SSID, + None, + ['11.0']) + + def test_open_b_ip6_dhcp(self): + '''Open network, 802.11b, IPv6 with DHCP''' + + self.do_test('hw_mode=b\nchannel=1\nssid=' + SSID, + 'ssid="%s"\nkey_mgmt=NONE' % SSID, + '', + ['11.0']) + + def test_open_b_ip6_raonly(self): + '''Open network, 802.11b, IPv6 with only RA''' + + self.do_test('hw_mode=b\nchannel=1\nssid=' + SSID, + 'ssid="%s"\nkey_mgmt=NONE' % SSID, + 'ra-only', + ['11.0']) + + def test_open_g_ip4(self): + '''Open network, 802.11g, IPv4''' + + self.do_test('hw_mode=g\nchannel=1\nssid=' + SSID, + 'ssid="%s"\nkey_mgmt=NONE' % SSID, + None, + ['54.0']) + + def test_wpa1_ip4(self): + '''WPA1, 802.11g, IPv4''' + + self.do_test('''hw_mode=g +channel=1 +ssid=%s +wpa=1 +wpa_key_mgmt=WPA-PSK +wpa_pairwise=TKIP +wpa_passphrase=12345678 +''' % SSID, + '''ssid="%s" +psk="12345678" +key_mgmt=WPA-PSK +proto=WPA +pairwise=TKIP +group=TKIP''' % SSID, + None, + ['54.0', + 'Pairwise ciphers: TKIP', + 'Authentication suites: PSK']) + + def test_wpa2_ip4(self): + '''WPA2, 802.11g, IPv4''' + + self.do_test('''hw_mode=g +channel=1 +ssid=%s +wpa=2 +wpa_key_mgmt=WPA-PSK +wpa_pairwise=CCMP +wpa_passphrase=12345678 +''' % SSID, + '''ssid="%s" +psk="12345678" +key_mgmt=WPA-PSK +proto=WPA2 +pairwise=CCMP +group=CCMP''' % SSID, + None, + ['54.0', + 'Pairwise ciphers: CCMP', + 'Authentication suites: PSK']) + + def test_wpa2_ip6(self): + '''WPA2, 802.11g, IPv6''' + + self.do_test('''hw_mode=g +channel=1 +ssid=%s +wpa=2 +wpa_key_mgmt=WPA-PSK +wpa_pairwise=CCMP +wpa_passphrase=12345678 +''' % SSID, + '''ssid="%s" +psk="12345678" +key_mgmt=WPA-PSK +proto=WPA2 +pairwise=CCMP +group=CCMP''' % SSID, + '', + ['54.0', + 'Pairwise ciphers: CCMP', + 'Authentication suites: PSK']) + + # + # Common for all tests + # + + def do_test(self, hostapd_conf, wpa_conf, ipv6, exp_iw_scan): + self.setup_ap(hostapd_conf, ipv6) + self.check_ssid_avail(exp_iw_scan) + self.start_wpasupp(wpa_conf) + self.check_iw_link() + self.check_communication(ipv6) + self.check_address(ipv6) + + def check_ssid_avail(self, expected_strings): + subprocess.check_call(['ip', 'link', 'set', self.dev_w_client, 'up']) + out = subprocess.check_output(['iw', 'dev', self.dev_w_client, 'scan'], + universal_newlines=True) + # down it again, wpa_supplicant is supposed to up it by itself + subprocess.check_call(['ip', 'link', 'set', self.dev_w_client, 'down']) + + self.assertRegex(out, 'SSID: ' + SSID) + for s in expected_strings: + self.assertRegex(out, s) + + def check_iw_link(self): + tries = 10 + while tries > 0: + out = subprocess.check_output(['iw', 'dev', self.dev_w_client, 'link'], + universal_newlines=True) + if 'SSID' in out: + break + tries -= 1 + time.sleep(1) + else: + self.fail('timed out on iwconfig showing connected status') + + self.assertRegex(out, 'Connected to ' + self.mac_w_ap) + self.assertRegex(out, 'SSID: ' + SSID) + + def check_communication(self, ipv6_mode): + '''Verify that communication works between AP and client + + This proves that wpa_supplicant set up a working link. We use a DHCP + request for IPv4 and IPv6 in DHCP mode. For IPv6 in other (i. + e. ra-only/slaac) modes, check that dnsmasq received a router solicit + and sends a router advertisement. + ''' + if ipv6_mode is not None: + self.poll_text(self.dnsmasq_log, 'RTR-SOLICIT(%s)' % self.dev_w_ap, timeout=50) + self.poll_text(self.dnsmasq_log, 'RTR-ADVERT(%s)' % self.dev_w_ap, timeout=5) + + # stop here for non-DHCP modes in IPv6 + if ipv6_mode not in (None, '', 'slaac'): + return + + # FIXME: sometimes takes more than 5 s in IPv6 mode + if ipv6_mode is not None: + mode = '-6' + timeout = 10 + else: + mode = '-4' + timeout = 5 + + # run DHCP client on client + out = subprocess.check_output(['dhclient', mode, '-1', '-v', + '-lf', '/dev/null', self.dev_w_client], + universal_newlines=True, timeout=timeout, + stderr=subprocess.STDOUT) + # stop DHCP client + subprocess.call(['dhclient', '-x', mode, self.dev_w_client], + stderr=subprocess.STDOUT) + + if ipv6_mode is not None: + self.assertRegex(out, 'status code Success') + self.assertRegex(out, 'IAADDR 2600::') + else: + self.assertRegex(out, 'DHCPACK of') + self.assertRegex(out, 'bound to') + + def check_address(self, ipv6_mode): + '''Verify that the interface got an appropriate address assigned''' + + out = subprocess.check_output(['ip', 'a', 'show', 'dev', self.dev_w_client], + universal_newlines=True) + self.assertRegex(out, 'state UP') + if ipv6_mode is None: + self.assertRegex(out, 'inet 192.168.5.\d+/24') + else: + if not ipv6_mode: + # has global address from our DHCP server + self.assertRegex(out, 'inet6 2600::[0-9a-z]+/\d') + else: + # has address with our prefix and MAC + self.assertRegex(out, 'inet6 2600::ff:fe00:[0-9a-z]+/64 scope global (?:tentative )?(?:mngtmpaddr )?dynamic') + + # has a link-local address + self.assertRegex(out, 'inet6 fe80::ff:fe00:[0-9a-z:]+/64 scope link') + + +# write to stdout, not stderr +runner = unittest.TextTestRunner(stream=sys.stdout, verbosity=2) +unittest.main(testRunner=runner) |