test/networkd-test.py

   1 #!/usr/bin/env python3
   2 #
   3 # networkd integration test
   4 # This uses temporary configuration in /run and temporary veth devices, and
   5 # does not write anything on disk or change any system configuration;
   6 # but it assumes (and checks at the beginning) that networkd is not currently
   7 # running.
   8 #
   9 # This can be run on a normal installation, in QEMU, nspawn (with
  10 # --private-network), LXD (with "--config raw.lxc=lxc.aa_profile=unconfined"),
  11 # or LXC system containers. You need at least the "ip" tool from the iproute
  12 # package; it is recommended to install dnsmasq too to get full test coverage.
  13 #
  14 # ATTENTION: This uses the *installed* networkd, not the one from the built
  15 # source tree.
  16 #
  17 # (C) 2015 Canonical Ltd.
  18 # Author: Martin Pitt <martin.pitt@ubuntu.com>
  19 #
  20 # systemd is free software; you can redistribute it and/or modify it
  21 # under the terms of the GNU Lesser General Public License as published by
  22 # the Free Software Foundation; either version 2.1 of the License, or
  23 # (at your option) any later version.
  24
  25 # systemd is distributed in the hope that it will be useful, but
  26 # WITHOUT ANY WARRANTY; without even the implied warranty of
  27 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  28 # Lesser General Public License for more details.
  29 #
  30 # You should have received a copy of the GNU Lesser General Public License
  31 # along with systemd; If not, see <http://www.gnu.org/licenses/>.
  32
  33 import os
  34 import sys
  35 import time
  36 import unittest
  37 import tempfile
  38 import subprocess
  39 import shutil
  40
  41 networkd_active = subprocess.call(['systemctl', 'is-active', '--quiet',
  42                                    'systemd-networkd']) == 0
  43 have_dnsmasq = shutil.which('dnsmasq')
  44
  45 RESOLV_CONF = '/run/systemd/resolve/resolv.conf'
  46
  47
  48 @unittest.skipIf(networkd_active,
  49                  'networkd is already active')
  50 class ClientTestBase:
  51     def setUp(self):
  52         self.iface = 'test_eth42'
  53         self.if_router = 'router_eth42'
  54         self.workdir_obj = tempfile.TemporaryDirectory()
  55         self.workdir = self.workdir_obj.name
  56         self.config = '/run/systemd/network/test_eth42.network'
  57         os.makedirs(os.path.dirname(self.config), exist_ok=True)
  58
  59         # avoid "Failed to open /dev/tty" errors in containers
  60         os.environ['SYSTEMD_LOG_TARGET'] = 'journal'
  61
  62         # determine path to systemd-networkd-wait-online
  63         for p in ['/usr/lib/systemd/systemd-networkd-wait-online',
  64                   '/lib/systemd/systemd-networkd-wait-online']:
  65             if os.path.exists(p):
  66                 self.networkd_wait_online = p
  67                 break
  68         else:
  69             self.fail('systemd-networkd-wait-online not found')
  70
  71         # get current journal cursor
  72         out = subprocess.check_output(['journalctl', '-b', '--quiet',
  73                                        '--no-pager', '-n0', '--show-cursor'],
  74                                       universal_newlines=True)
  75         self.assertTrue(out.startswith('-- cursor:'))
  76         self.journal_cursor = out.split()[-1]
  77
  78     def tearDown(self):
  79         self.shutdown_iface()
  80         if os.path.exists(self.config):
  81             os.unlink(self.config)
  82         subprocess.call(['systemctl', 'stop', 'systemd-networkd'])
  83
  84     def show_journal(self, unit):
  85         '''Show journal of given unit since start of the test'''
  86
  87         print('---- %s ----' % unit)
  88         sys.stdout.flush()
  89         subprocess.call(['journalctl', '-b', '--no-pager', '--quiet',
  90                          '--cursor', self.journal_cursor, '-u', unit])
  91
  92     def create_iface(self, ipv6=False):
  93         '''Create test interface with DHCP server behind it'''
  94
  95         raise NotImplementedError('must be implemented by a subclass')
  96
  97     def shutdown_iface(self):
  98         '''Remove test interface and stop DHCP server'''
  99
 100         raise NotImplementedError('must be implemented by a subclass')
 101
 102     def print_server_log(self):
 103         '''Print DHCP server log for debugging failures'''
 104
 105         raise NotImplementedError('must be implemented by a subclass')
 106
 107     def do_test(self, coldplug=True, ipv6=False, extra_opts='',
 108                 online_timeout=10, dhcp_mode='yes'):
 109         subprocess.check_call(['systemctl', 'start', 'systemd-resolved'])
 110         with open(self.config, 'w') as f:
 111             f.write('''[Match]
 112 Name=%s
 113 [Network]
 114 DHCP=%s
 115 %s''' % (self.iface, dhcp_mode, extra_opts))
 116
 117         if coldplug:
 118             # create interface first, then start networkd
 119             self.create_iface(ipv6=ipv6)
 120             subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 121         else:
 122             # start networkd first, then create interface
 123             subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 124             self.create_iface(ipv6=ipv6)
 125
 126         try:
 127             subprocess.check_call([self.networkd_wait_online, '--interface',
 128                                    self.iface, '--timeout=%i' % online_timeout])
 129
 130             if ipv6:
 131                 # check iface state and IP 6 address; FIXME: we need to wait a bit
 132                 # longer, as the iface is "configured" already with IPv4 *or*
 133                 # IPv6, but we want to wait for both
 134                 for timeout in range(10):
 135                     out = subprocess.check_output(['ip', 'a', 'show', 'dev', self.iface])
 136                     if b'state UP' in out and b'inet6 2600' in out and b'inet 192.168' in out:
 137                         break
 138                     time.sleep(1)
 139                 else:
 140                     self.fail('timed out waiting for IPv6 configuration')
 141
 142                 self.assertRegex(out, b'inet6 2600::.* scope global .*dynamic')
 143                 self.assertRegex(out, b'inet6 fe80::.* scope link')
 144             else:
 145                 # should have link-local address on IPv6 only
 146                 out = subprocess.check_output(['ip', '-6', 'a', 'show', 'dev', self.iface])
 147                 self.assertRegex(out, b'inet6 fe80::.* scope link')
 148                 self.assertNotIn(b'scope global', out)
 149
 150             # should have IPv4 address
 151             out = subprocess.check_output(['ip', '-4', 'a', 'show', 'dev', self.iface])
 152             self.assertIn(b'state UP', out)
 153             self.assertRegex(out, b'inet 192.168.5.\d+/.* scope global dynamic')
 154
 155             # check networkctl state
 156             out = subprocess.check_output(['networkctl'])
 157             self.assertRegex(out, ('%s\s+ether\s+routable\s+unmanaged' % self.if_router).encode())
 158             self.assertRegex(out, ('%s\s+ether\s+routable\s+configured' % self.iface).encode())
 159
 160             out = subprocess.check_output(['networkctl', 'status', self.iface])
 161             self.assertRegex(out, b'Type:\s+ether')
 162             self.assertRegex(out, b'State:\s+routable.*configured')
 163             self.assertRegex(out, b'Address:\s+192.168.5.\d+')
 164             if ipv6:
 165                 self.assertRegex(out, b'2600::')
 166             else:
 167                 self.assertNotIn(b'2600::', out)
 168             self.assertRegex(out, b'fe80::')
 169             self.assertRegex(out, b'Gateway:\s+192.168.5.1')
 170             self.assertRegex(out, b'DNS:\s+192.168.5.1')
 171         except (AssertionError, subprocess.CalledProcessError):
 172             # show networkd status, journal, and DHCP server log on failure
 173             with open(self.config) as f:
 174                 print('\n---- %s ----\n%s' % (self.config, f.read()))
 175             print('---- interface status ----')
 176             sys.stdout.flush()
 177             subprocess.call(['ip', 'a', 'show', 'dev', self.iface])
 178             print('---- networkctl status %s ----' % self.iface)
 179             sys.stdout.flush()
 180             subprocess.call(['networkctl', 'status', self.iface])
 181             self.show_journal('systemd-networkd.service')
 182             self.print_server_log()
 183             raise
 184
 185         for timeout in range(50):
 186             with open(RESOLV_CONF) as f:
 187                 contents = f.read()
 188             if 'nameserver 192.168.5.1\n' in contents:
 189                 break
 190             time.sleep(0.1)
 191         else:
 192             self.fail('nameserver 192.168.5.1 not found in ' + RESOLV_CONF)
 193
 194         if not coldplug:
 195             # check post-down.d hook
 196             self.shutdown_iface()
 197
 198     def test_coldplug_dhcp_yes_ip4(self):
 199         # we have a 12s timeout on RA, so we need to wait longer
 200         self.do_test(coldplug=True, ipv6=False, online_timeout=15)
 201
 202     def test_coldplug_dhcp_yes_ip4_no_ra(self):
 203         # with disabling RA explicitly things should be fast
 204         self.do_test(coldplug=True, ipv6=False,
 205                      extra_opts='IPv6AcceptRA=False')
 206
 207     def test_coldplug_dhcp_ip4_only(self):
 208         # we have a 12s timeout on RA, so we need to wait longer
 209         self.do_test(coldplug=True, ipv6=False, dhcp_mode='ipv4',
 210                      online_timeout=15)
 211
 212     def test_coldplug_dhcp_ip4_only_no_ra(self):
 213         # with disabling RA explicitly things should be fast
 214         self.do_test(coldplug=True, ipv6=False, dhcp_mode='ipv4',
 215                      extra_opts='IPv6AcceptRA=False')
 216
 217     def test_coldplug_dhcp_ip6(self):
 218         self.do_test(coldplug=True, ipv6=True)
 219
 220     def test_hotplug_dhcp_ip4(self):
 221         # With IPv4 only we have a 12s timeout on RA, so we need to wait longer
 222         self.do_test(coldplug=False, ipv6=False, online_timeout=15)
 223
 224     def test_hotplug_dhcp_ip6(self):
 225         self.do_test(coldplug=False, ipv6=True)
 226
 227     def test_route_only_dns(self):
 228         with open('/run/systemd/network/myvpn.netdev', 'w') as f:
 229             f.write('''[NetDev]
 230 Name=dummy0
 231 Kind=dummy
 232 MACAddress=12:34:56:78:9a:bc''')
 233         with open('/run/systemd/network/myvpn.network', 'w') as f:
 234             f.write('''[Match]
 235 Name=dummy0
 236 [Network]
 237 Address=192.168.42.100
 238 DNS=192.168.42.1
 239 Domains= ~company''')
 240         self.addCleanup(os.remove, '/run/systemd/network/myvpn.netdev')
 241         self.addCleanup(os.remove, '/run/systemd/network/myvpn.network')
 242
 243         self.do_test(coldplug=True, ipv6=False,
 244                      extra_opts='IPv6AcceptRouterAdvertisements=False')
 245
 246         with open(RESOLV_CONF) as f:
 247             contents = f.read()
 248             # ~company is not a search domain, only a routing domain
 249             self.assertNotRegex(contents, 'search.*company')
 250             # our global server should appear
 251             self.assertIn('nameserver 192.168.5.1\n', contents)
 252
 253
 254 @unittest.skipUnless(have_dnsmasq, 'dnsmasq not installed')
 255 class DnsmasqClientTest(ClientTestBase, unittest.TestCase):
 256     '''Test networkd client against dnsmasq'''
 257
 258     def setUp(self):
 259         super().setUp()
 260         self.dnsmasq = None
 261
 262     def create_iface(self, ipv6=False):
 263         '''Create test interface with DHCP server behind it'''
 264
 265         # add veth pair
 266         subprocess.check_call(['ip', 'link', 'add', 'name', self.iface, 'type',
 267                                'veth', 'peer', 'name', self.if_router])
 268
 269         # give our router an IP
 270         subprocess.check_call(['ip', 'a', 'flush', 'dev', self.if_router])
 271         subprocess.check_call(['ip', 'a', 'add', '192.168.5.1/24', 'dev', self.if_router])
 272         if ipv6:
 273             subprocess.check_call(['ip', 'a', 'add', '2600::1/64', 'dev', self.if_router])
 274         subprocess.check_call(['ip', 'link', 'set', self.if_router, 'up'])
 275
 276         # add DHCP server
 277         self.dnsmasq_log = os.path.join(self.workdir, 'dnsmasq.log')
 278         lease_file = os.path.join(self.workdir, 'dnsmasq.leases')
 279         if ipv6:
 280             extra_opts = ['--enable-ra', '--dhcp-range=2600::10,2600::20']
 281         else:
 282             extra_opts = []
 283         self.dnsmasq = subprocess.Popen(
 284             ['dnsmasq', '--keep-in-foreground', '--log-queries',
 285              '--log-facility=' + self.dnsmasq_log, '--conf-file=/dev/null',
 286              '--dhcp-leasefile=' + lease_file, '--bind-interfaces',
 287              '--interface=' + self.if_router, '--except-interface=lo',
 288              '--dhcp-range=192.168.5.10,192.168.5.200'] + extra_opts)
 289
 290     def shutdown_iface(self):
 291         '''Remove test interface and stop DHCP server'''
 292
 293         if self.if_router:
 294             subprocess.check_call(['ip', 'link', 'del', 'dev', self.if_router])
 295             self.if_router = None
 296         if self.dnsmasq:
 297             self.dnsmasq.kill()
 298             self.dnsmasq.wait()
 299             self.dnsmasq = None
 300
 301     def print_server_log(self):
 302         '''Print DHCP server log for debugging failures'''
 303
 304         with open(self.dnsmasq_log) as f:
 305             sys.stdout.write('\n\n---- dnsmasq log ----\n%s\n------\n\n' % f.read())
 306
 307
 308 class NetworkdClientTest(ClientTestBase, unittest.TestCase):
 309     '''Test networkd client against networkd server'''
 310
 311     def setUp(self):
 312         super().setUp()
 313         self.dnsmasq = None
 314
 315     def create_iface(self, ipv6=False):
 316         '''Create test interface with DHCP server behind it'''
 317
 318         # run "router-side" networkd in own mount namespace to shield it from
 319         # "client-side" configuration and networkd
 320         (fd, script) = tempfile.mkstemp(prefix='networkd-router.sh')
 321         self.addCleanup(os.remove, script)
 322         with os.fdopen(fd, 'w+') as f:
 323             f.write('''#!/bin/sh -eu
 324 mkdir -p /run/systemd/network
 325 mkdir -p /run/systemd/netif
 326 mount -t tmpfs none /run/systemd/network
 327 mount -t tmpfs none /run/systemd/netif
 328 [ ! -e /run/dbus ] || mount -t tmpfs none /run/dbus
 329 # create router/client veth pair
 330 cat << EOF > /run/systemd/network/test.netdev
 331 [NetDev]
 332 Name=%(ifr)s
 333 Kind=veth
 334
 335 [Peer]
 336 Name=%(ifc)s
 337 EOF
 338
 339 cat << EOF > /run/systemd/network/test.network
 340 [Match]
 341 Name=%(ifr)s
 342
 343 [Network]
 344 Address=192.168.5.1/24
 345 %(addr6)s
 346 DHCPServer=yes
 347
 348 [DHCPServer]
 349 PoolOffset=10
 350 PoolSize=50
 351 DNS=192.168.5.1
 352 EOF
 353
 354 # run networkd as in systemd-networkd.service
 355 exec $(systemctl cat systemd-networkd.service | sed -n '/^ExecStart=/ { s/^.*=//; p}')
 356 ''' % {'ifr': self.if_router, 'ifc': self.iface, 'addr6': ipv6 and 'Address=2600::1/64' or ''})
 357
 358             os.fchmod(fd, 0o755)
 359
 360         subprocess.check_call(['systemd-run', '--unit=networkd-test-router.service',
 361                                '-p', 'InaccessibleDirectories=-/etc/systemd/network',
 362                                '-p', 'InaccessibleDirectories=-/run/systemd/network',
 363                                '-p', 'InaccessibleDirectories=-/run/systemd/netif',
 364                                '--service-type=notify', script])
 365
 366         # wait until devices got created
 367         for timeout in range(50):
 368             out = subprocess.check_output(['ip', 'a', 'show', 'dev', self.if_router])
 369             if b'state UP' in out and b'scope global' in out:
 370                 break
 371             time.sleep(0.1)
 372
 373     def shutdown_iface(self):
 374         '''Remove test interface and stop DHCP server'''
 375
 376         if self.if_router:
 377             subprocess.check_call(['systemctl', 'stop', 'networkd-test-router.service'])
 378             # ensure failed transient unit does not stay around
 379             subprocess.call(['systemctl', 'reset-failed', 'networkd-test-router.service'])
 380             subprocess.call(['ip', 'link', 'del', 'dev', self.if_router])
 381             self.if_router = None
 382
 383     def print_server_log(self):
 384         '''Print DHCP server log for debugging failures'''
 385
 386         self.show_journal('networkd-test-router.service')
 387
 388     @unittest.skip('networkd does not have DHCPv6 server support')
 389     def test_hotplug_dhcp_ip6(self):
 390         pass
 391
 392     @unittest.skip('networkd does not have DHCPv6 server support')
 393     def test_coldplug_dhcp_ip6(self):
 394         pass
 395
 396     def test_search_domains(self):
 397
 398         # we don't use this interface for this test
 399         self.if_router = None
 400
 401         with open('/run/systemd/network/test.netdev', 'w') as f:
 402             f.write('''[NetDev]
 403 Name=dummy0
 404 Kind=dummy
 405 MACAddress=12:34:56:78:9a:bc''')
 406         with open('/run/systemd/network/test.network', 'w') as f:
 407             f.write('''[Match]
 408 Name=dummy0
 409 [Network]
 410 Address=192.168.42.100
 411 DNS=192.168.42.1
 412 Domains= one two three four five six seven eight nine ten''')
 413         self.addCleanup(os.remove, '/run/systemd/network/test.netdev')
 414         self.addCleanup(os.remove, '/run/systemd/network/test.network')
 415
 416         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 417
 418         for timeout in range(50):
 419             with open(RESOLV_CONF) as f:
 420                 contents = f.read()
 421             if ' one' in contents:
 422                 break
 423             time.sleep(0.1)
 424         self.assertRegex(contents, 'search .*one two three four')
 425         self.assertNotIn('seven\n', contents)
 426         self.assertIn('# Too many search domains configured, remaining ones ignored.\n', contents)
 427
 428     def test_search_domains_too_long(self):
 429
 430         # we don't use this interface for this test
 431         self.if_router = None
 432
 433         name_prefix = 'a' * 60
 434
 435         with open('/run/systemd/network/test.netdev', 'w') as f:
 436             f.write('''[NetDev]
 437 Name=dummy0
 438 Kind=dummy
 439 MACAddress=12:34:56:78:9a:bc''')
 440         with open('/run/systemd/network/test.network', 'w') as f:
 441             f.write('''[Match]
 442 Name=dummy0
 443 [Network]
 444 Address=192.168.42.100
 445 DNS=192.168.42.1
 446 Domains=''')
 447             for i in range(5):
 448                 f.write('%s%i ' % (name_prefix, i))
 449
 450         self.addCleanup(os.remove, '/run/systemd/network/test.netdev')
 451         self.addCleanup(os.remove, '/run/systemd/network/test.network')
 452
 453         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 454
 455         for timeout in range(50):
 456             with open(RESOLV_CONF) as f:
 457                 contents = f.read()
 458             if ' one' in contents:
 459                 break
 460             time.sleep(0.1)
 461         self.assertRegex(contents, 'search .*%(p)s0 %(p)s1 %(p)s2' % {'p': name_prefix})
 462         self.assertIn('# Total length of all search domains is too long, remaining ones ignored.', contents)
 463
 464
 465 if __name__ == '__main__':
 466     unittest.main(testRunner=unittest.TextTestRunner(stream=sys.stdout,
 467                                                      verbosity=2))