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 import socket
  41
  42 networkd_active = subprocess.call(['systemctl', 'is-active', '--quiet',
  43                                    'systemd-networkd']) == 0
  44 have_dnsmasq = shutil.which('dnsmasq')
  45
  46 RESOLV_CONF = '/run/systemd/resolve/resolv.conf'
  47
  48
  49 @unittest.skipIf(networkd_active,
  50                  'networkd is already active')
  51 class ClientTestBase:
  52     @classmethod
  53     def setUpClass(klass):
  54         klass.orig_log_level = subprocess.check_output(
  55             ['systemctl', 'show', '--value', '--property', 'LogLevel'],
  56             universal_newlines=True).strip()
  57         subprocess.check_call(['systemd-analyze', 'set-log-level', 'debug'])
  58
  59     @classmethod
  60     def tearDownClass(klass):
  61         subprocess.check_call(['systemd-analyze', 'set-log-level', klass.orig_log_level])
  62
  63     def setUp(self):
  64         self.iface = 'test_eth42'
  65         self.if_router = 'router_eth42'
  66         self.workdir_obj = tempfile.TemporaryDirectory()
  67         self.workdir = self.workdir_obj.name
  68         self.config = '/run/systemd/network/test_eth42.network'
  69
  70         # avoid "Failed to open /dev/tty" errors in containers
  71         os.environ['SYSTEMD_LOG_TARGET'] = 'journal'
  72
  73         # determine path to systemd-networkd-wait-online
  74         for p in ['/usr/lib/systemd/systemd-networkd-wait-online',
  75                   '/lib/systemd/systemd-networkd-wait-online']:
  76             if os.path.exists(p):
  77                 self.networkd_wait_online = p
  78                 break
  79         else:
  80             self.fail('systemd-networkd-wait-online not found')
  81
  82         # get current journal cursor
  83         subprocess.check_output(['journalctl', '--sync'])
  84         out = subprocess.check_output(['journalctl', '-b', '--quiet',
  85                                        '--no-pager', '-n0', '--show-cursor'],
  86                                       universal_newlines=True)
  87         self.assertTrue(out.startswith('-- cursor:'))
  88         self.journal_cursor = out.split()[-1]
  89
  90     def tearDown(self):
  91         self.shutdown_iface()
  92         subprocess.call(['systemctl', 'stop', 'systemd-networkd'])
  93         subprocess.call(['ip', 'link', 'del', 'dummy0'],
  94                         stderr=subprocess.DEVNULL)
  95
  96     def writeConfig(self, fname, contents):
  97         os.makedirs(os.path.dirname(fname), exist_ok=True)
  98         with open(fname, 'w') as f:
  99             f.write(contents)
 100         self.addCleanup(os.remove, fname)
 101
 102     def show_journal(self, unit):
 103         '''Show journal of given unit since start of the test'''
 104
 105         print('---- %s ----' % unit)
 106         subprocess.check_output(['journalctl', '--sync'])
 107         sys.stdout.flush()
 108         subprocess.call(['journalctl', '-b', '--no-pager', '--quiet',
 109                          '--cursor', self.journal_cursor, '-u', unit])
 110
 111     def create_iface(self, ipv6=False):
 112         '''Create test interface with DHCP server behind it'''
 113
 114         raise NotImplementedError('must be implemented by a subclass')
 115
 116     def shutdown_iface(self):
 117         '''Remove test interface and stop DHCP server'''
 118
 119         raise NotImplementedError('must be implemented by a subclass')
 120
 121     def print_server_log(self):
 122         '''Print DHCP server log for debugging failures'''
 123
 124         raise NotImplementedError('must be implemented by a subclass')
 125
 126     def do_test(self, coldplug=True, ipv6=False, extra_opts='',
 127                 online_timeout=10, dhcp_mode='yes'):
 128         subprocess.check_call(['systemctl', 'start', 'systemd-resolved'])
 129         self.writeConfig(self.config, '''\
 130 [Match]
 131 Name=%s
 132 [Network]
 133 DHCP=%s
 134 %s''' % (self.iface, dhcp_mode, extra_opts))
 135
 136         if coldplug:
 137             # create interface first, then start networkd
 138             self.create_iface(ipv6=ipv6)
 139             subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 140         elif coldplug is not None:
 141             # start networkd first, then create interface
 142             subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 143             self.create_iface(ipv6=ipv6)
 144         else:
 145             # "None" means test sets up interface by itself
 146             subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 147
 148         try:
 149             subprocess.check_call([self.networkd_wait_online, '--interface',
 150                                    self.iface, '--timeout=%i' % online_timeout])
 151
 152             if ipv6:
 153                 # check iface state and IP 6 address; FIXME: we need to wait a bit
 154                 # longer, as the iface is "configured" already with IPv4 *or*
 155                 # IPv6, but we want to wait for both
 156                 for _ in range(10):
 157                     out = subprocess.check_output(['ip', 'a', 'show', 'dev', self.iface])
 158                     if b'state UP' in out and b'inet6 2600' in out and b'inet 192.168' in out:
 159                         break
 160                     time.sleep(1)
 161                 else:
 162                     self.fail('timed out waiting for IPv6 configuration')
 163
 164                 self.assertRegex(out, b'inet6 2600::.* scope global .*dynamic')
 165                 self.assertRegex(out, b'inet6 fe80::.* scope link')
 166             else:
 167                 # should have link-local address on IPv6 only
 168                 out = subprocess.check_output(['ip', '-6', 'a', 'show', 'dev', self.iface])
 169                 self.assertRegex(out, b'inet6 fe80::.* scope link')
 170                 self.assertNotIn(b'scope global', out)
 171
 172             # should have IPv4 address
 173             out = subprocess.check_output(['ip', '-4', 'a', 'show', 'dev', self.iface])
 174             self.assertIn(b'state UP', out)
 175             self.assertRegex(out, b'inet 192.168.5.\d+/.* scope global dynamic')
 176
 177             # check networkctl state
 178             out = subprocess.check_output(['networkctl'])
 179             self.assertRegex(out, ('%s\s+ether\s+routable\s+unmanaged' % self.if_router).encode())
 180             self.assertRegex(out, ('%s\s+ether\s+routable\s+configured' % self.iface).encode())
 181
 182             out = subprocess.check_output(['networkctl', 'status', self.iface])
 183             self.assertRegex(out, b'Type:\s+ether')
 184             self.assertRegex(out, b'State:\s+routable.*configured')
 185             self.assertRegex(out, b'Address:\s+192.168.5.\d+')
 186             if ipv6:
 187                 self.assertRegex(out, b'2600::')
 188             else:
 189                 self.assertNotIn(b'2600::', out)
 190             self.assertRegex(out, b'fe80::')
 191             self.assertRegex(out, b'Gateway:\s+192.168.5.1')
 192             self.assertRegex(out, b'DNS:\s+192.168.5.1')
 193         except (AssertionError, subprocess.CalledProcessError):
 194             # show networkd status, journal, and DHCP server log on failure
 195             with open(self.config) as f:
 196                 print('\n---- %s ----\n%s' % (self.config, f.read()))
 197             print('---- interface status ----')
 198             sys.stdout.flush()
 199             subprocess.call(['ip', 'a', 'show', 'dev', self.iface])
 200             print('---- networkctl status %s ----' % self.iface)
 201             sys.stdout.flush()
 202             subprocess.call(['networkctl', 'status', self.iface])
 203             self.show_journal('systemd-networkd.service')
 204             self.print_server_log()
 205             raise
 206
 207         for timeout in range(50):
 208             with open(RESOLV_CONF) as f:
 209                 contents = f.read()
 210             if 'nameserver 192.168.5.1\n' in contents:
 211                 break
 212             time.sleep(0.1)
 213         else:
 214             self.fail('nameserver 192.168.5.1 not found in ' + RESOLV_CONF)
 215
 216         if coldplug is False:
 217             # check post-down.d hook
 218             self.shutdown_iface()
 219
 220     def test_coldplug_dhcp_yes_ip4(self):
 221         # we have a 12s timeout on RA, so we need to wait longer
 222         self.do_test(coldplug=True, ipv6=False, online_timeout=15)
 223
 224     def test_coldplug_dhcp_yes_ip4_no_ra(self):
 225         # with disabling RA explicitly things should be fast
 226         self.do_test(coldplug=True, ipv6=False,
 227                      extra_opts='IPv6AcceptRA=False')
 228
 229     def test_coldplug_dhcp_ip4_only(self):
 230         # we have a 12s timeout on RA, so we need to wait longer
 231         self.do_test(coldplug=True, ipv6=False, dhcp_mode='ipv4',
 232                      online_timeout=15)
 233
 234     def test_coldplug_dhcp_ip4_only_no_ra(self):
 235         # with disabling RA explicitly things should be fast
 236         self.do_test(coldplug=True, ipv6=False, dhcp_mode='ipv4',
 237                      extra_opts='IPv6AcceptRA=False')
 238
 239     def test_coldplug_dhcp_ip6(self):
 240         self.do_test(coldplug=True, ipv6=True)
 241
 242     def test_hotplug_dhcp_ip4(self):
 243         # With IPv4 only we have a 12s timeout on RA, so we need to wait longer
 244         self.do_test(coldplug=False, ipv6=False, online_timeout=15)
 245
 246     def test_hotplug_dhcp_ip6(self):
 247         self.do_test(coldplug=False, ipv6=True)
 248
 249     def test_route_only_dns(self):
 250         self.writeConfig('/run/systemd/network/myvpn.netdev', '''\
 251 [NetDev]
 252 Name=dummy0
 253 Kind=dummy
 254 MACAddress=12:34:56:78:9a:bc''')
 255         self.writeConfig('/run/systemd/network/myvpn.network', '''\
 256 [Match]
 257 Name=dummy0
 258 [Network]
 259 Address=192.168.42.100
 260 DNS=192.168.42.1
 261 Domains= ~company''')
 262
 263         self.do_test(coldplug=True, ipv6=False,
 264                      extra_opts='IPv6AcceptRouterAdvertisements=False')
 265
 266         with open(RESOLV_CONF) as f:
 267             contents = f.read()
 268             # ~company is not a search domain, only a routing domain
 269             self.assertNotRegex(contents, 'search.*company')
 270             # our global server should appear
 271             self.assertIn('nameserver 192.168.5.1\n', contents)
 272             # should not have domain-restricted server as global server
 273             self.assertNotIn('nameserver 192.168.42.1\n', contents)
 274
 275     def test_route_only_dns_all_domains(self):
 276         with open('/run/systemd/network/myvpn.netdev', 'w') as f:
 277             f.write('''[NetDev]
 278 Name=dummy0
 279 Kind=dummy
 280 MACAddress=12:34:56:78:9a:bc''')
 281         with open('/run/systemd/network/myvpn.network', 'w') as f:
 282             f.write('''[Match]
 283 Name=dummy0
 284 [Network]
 285 Address=192.168.42.100
 286 DNS=192.168.42.1
 287 Domains= ~company ~.''')
 288         self.addCleanup(os.remove, '/run/systemd/network/myvpn.netdev')
 289         self.addCleanup(os.remove, '/run/systemd/network/myvpn.network')
 290
 291         self.do_test(coldplug=True, ipv6=False,
 292                      extra_opts='IPv6AcceptRouterAdvertisements=False')
 293
 294         with open(RESOLV_CONF) as f:
 295             contents = f.read()
 296
 297         # ~company is not a search domain, only a routing domain
 298         self.assertNotRegex(contents, 'search.*company')
 299
 300         # our global server should appear
 301         self.assertIn('nameserver 192.168.5.1\n', contents)
 302         # should have company server as global server due to ~.
 303         self.assertIn('nameserver 192.168.42.1\n', contents)
 304
 305
 306 @unittest.skipUnless(have_dnsmasq, 'dnsmasq not installed')
 307 class DnsmasqClientTest(ClientTestBase, unittest.TestCase):
 308     '''Test networkd client against dnsmasq'''
 309
 310     def setUp(self):
 311         super().setUp()
 312         self.dnsmasq = None
 313         self.iface_mac = 'de:ad:be:ef:47:11'
 314
 315     def create_iface(self, ipv6=False, dnsmasq_opts=None):
 316         '''Create test interface with DHCP server behind it'''
 317
 318         # add veth pair
 319         subprocess.check_call(['ip', 'link', 'add', 'name', self.iface,
 320                                'address', self.iface_mac,
 321                                'type', 'veth', 'peer', 'name', self.if_router])
 322
 323         # give our router an IP
 324         subprocess.check_call(['ip', 'a', 'flush', 'dev', self.if_router])
 325         subprocess.check_call(['ip', 'a', 'add', '192.168.5.1/24', 'dev', self.if_router])
 326         if ipv6:
 327             subprocess.check_call(['ip', 'a', 'add', '2600::1/64', 'dev', self.if_router])
 328         subprocess.check_call(['ip', 'link', 'set', self.if_router, 'up'])
 329
 330         # add DHCP server
 331         self.dnsmasq_log = os.path.join(self.workdir, 'dnsmasq.log')
 332         lease_file = os.path.join(self.workdir, 'dnsmasq.leases')
 333         if ipv6:
 334             extra_opts = ['--enable-ra', '--dhcp-range=2600::10,2600::20']
 335         else:
 336             extra_opts = []
 337         if dnsmasq_opts:
 338             extra_opts += dnsmasq_opts
 339         self.dnsmasq = subprocess.Popen(
 340             ['dnsmasq', '--keep-in-foreground', '--log-queries',
 341              '--log-facility=' + self.dnsmasq_log, '--conf-file=/dev/null',
 342              '--dhcp-leasefile=' + lease_file, '--bind-interfaces',
 343              '--interface=' + self.if_router, '--except-interface=lo',
 344              '--dhcp-range=192.168.5.10,192.168.5.200'] + extra_opts)
 345
 346     def shutdown_iface(self):
 347         '''Remove test interface and stop DHCP server'''
 348
 349         if self.if_router:
 350             subprocess.check_call(['ip', 'link', 'del', 'dev', self.if_router])
 351             self.if_router = None
 352         if self.dnsmasq:
 353             self.dnsmasq.kill()
 354             self.dnsmasq.wait()
 355             self.dnsmasq = None
 356
 357     def print_server_log(self):
 358         '''Print DHCP server log for debugging failures'''
 359
 360         with open(self.dnsmasq_log) as f:
 361             sys.stdout.write('\n\n---- dnsmasq log ----\n%s\n------\n\n' % f.read())
 362
 363     def test_resolved_domain_restricted_dns(self):
 364         '''resolved: domain-restricted DNS servers'''
 365
 366         # create interface for generic connections; this will map all DNS names
 367         # to 192.168.42.1
 368         self.create_iface(dnsmasq_opts=['--address=/#/192.168.42.1'])
 369         self.writeConfig('/run/systemd/network/general.network', '''\
 370 [Match]
 371 Name=%s
 372 [Network]
 373 DHCP=ipv4
 374 IPv6AcceptRA=False''' % self.iface)
 375
 376         # create second device/dnsmasq for a .company/.lab VPN interface
 377         # static IPs for simplicity
 378         subprocess.check_call(['ip', 'link', 'add', 'name', 'testvpnclient', 'type',
 379                                'veth', 'peer', 'name', 'testvpnrouter'])
 380         self.addCleanup(subprocess.call, ['ip', 'link', 'del', 'dev', 'testvpnrouter'])
 381         subprocess.check_call(['ip', 'a', 'flush', 'dev', 'testvpnrouter'])
 382         subprocess.check_call(['ip', 'a', 'add', '10.241.3.1/24', 'dev', 'testvpnrouter'])
 383         subprocess.check_call(['ip', 'link', 'set', 'testvpnrouter', 'up'])
 384
 385         vpn_dnsmasq_log = os.path.join(self.workdir, 'dnsmasq-vpn.log')
 386         vpn_dnsmasq = subprocess.Popen(
 387             ['dnsmasq', '--keep-in-foreground', '--log-queries',
 388              '--log-facility=' + vpn_dnsmasq_log, '--conf-file=/dev/null',
 389              '--dhcp-leasefile=/dev/null', '--bind-interfaces',
 390              '--interface=testvpnrouter', '--except-interface=lo',
 391              '--address=/math.lab/10.241.3.3', '--address=/cantina.company/10.241.4.4'])
 392         self.addCleanup(vpn_dnsmasq.wait)
 393         self.addCleanup(vpn_dnsmasq.kill)
 394
 395         self.writeConfig('/run/systemd/network/vpn.network', '''\
 396 [Match]
 397 Name=testvpnclient
 398 [Network]
 399 IPv6AcceptRA=False
 400 Address=10.241.3.2/24
 401 DNS=10.241.3.1
 402 Domains= ~company ~lab''')
 403
 404         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 405         subprocess.check_call([self.networkd_wait_online, '--interface', self.iface,
 406                                '--interface=testvpnclient', '--timeout=20'])
 407
 408         # ensure we start fresh with every test
 409         subprocess.check_call(['systemctl', 'restart', 'systemd-resolved'])
 410
 411         # test vpnclient specific domains; these should *not* be answered by
 412         # the general DNS
 413         out = subprocess.check_output(['systemd-resolve', 'math.lab'])
 414         self.assertIn(b'math.lab: 10.241.3.3', out)
 415         out = subprocess.check_output(['systemd-resolve', 'kettle.cantina.company'])
 416         self.assertIn(b'kettle.cantina.company: 10.241.4.4', out)
 417
 418         # test general domains
 419         out = subprocess.check_output(['systemd-resolve', 'megasearch.net'])
 420         self.assertIn(b'megasearch.net: 192.168.42.1', out)
 421
 422         with open(self.dnsmasq_log) as f:
 423             general_log = f.read()
 424         with open(vpn_dnsmasq_log) as f:
 425             vpn_log = f.read()
 426
 427         # VPN domains should only be sent to VPN DNS
 428         self.assertRegex(vpn_log, 'query.*math.lab')
 429         self.assertRegex(vpn_log, 'query.*cantina.company')
 430         self.assertNotIn('lab', general_log)
 431         self.assertNotIn('company', general_log)
 432
 433         # general domains should not be sent to the VPN DNS
 434         self.assertRegex(general_log, 'query.*megasearch.net')
 435         self.assertNotIn('megasearch.net', vpn_log)
 436
 437     def test_transient_hostname(self):
 438         '''networkd sets transient hostname from DHCP'''
 439
 440         orig_hostname = socket.gethostname()
 441         self.addCleanup(socket.sethostname, orig_hostname)
 442         # temporarily move /etc/hostname away; restart hostnamed to pick it up
 443         if os.path.exists('/etc/hostname'):
 444             subprocess.check_call(['mount', '--bind', '/dev/null', '/etc/hostname'])
 445             self.addCleanup(subprocess.call, ['umount', '/etc/hostname'])
 446         subprocess.check_call(['systemctl', 'stop', 'systemd-hostnamed.service'])
 447
 448         self.create_iface(dnsmasq_opts=['--dhcp-host=%s,192.168.5.210,testgreen' % self.iface_mac])
 449         self.do_test(coldplug=None, extra_opts='IPv6AcceptRA=False', dhcp_mode='ipv4')
 450
 451         try:
 452             # should have received the fixed IP above
 453             out = subprocess.check_output(['ip', '-4', 'a', 'show', 'dev', self.iface])
 454             self.assertRegex(out, b'inet 192.168.5.210/24 .* scope global dynamic')
 455             # should have set transient hostname in hostnamed; this is
 456             # sometimes a bit lagging (issue #4753), so retry a few times
 457             for retry in range(1, 6):
 458                 out = subprocess.check_output(['hostnamectl'])
 459                 if b'testgreen' in out:
 460                     break
 461                 time.sleep(5)
 462                 sys.stdout.write('[retry %i] ' % retry)
 463                 sys.stdout.flush()
 464             else:
 465                 self.fail('Transient hostname not found in hostnamectl:\n%s' % out.decode())
 466             # and also applied to the system
 467             self.assertEqual(socket.gethostname(), 'testgreen')
 468         except AssertionError:
 469             self.show_journal('systemd-networkd.service')
 470             self.show_journal('systemd-hostnamed.service')
 471             self.print_server_log()
 472             raise
 473
 474     def test_transient_hostname_with_static(self):
 475         '''transient hostname is not applied if static hostname exists'''
 476
 477         orig_hostname = socket.gethostname()
 478         self.addCleanup(socket.sethostname, orig_hostname)
 479         if not os.path.exists('/etc/hostname'):
 480             self.writeConfig('/etc/hostname', orig_hostname)
 481         subprocess.check_call(['systemctl', 'stop', 'systemd-hostnamed.service'])
 482
 483         self.create_iface(dnsmasq_opts=['--dhcp-host=%s,192.168.5.210,testgreen' % self.iface_mac])
 484         self.do_test(coldplug=None, extra_opts='IPv6AcceptRA=False', dhcp_mode='ipv4')
 485
 486         try:
 487             # should have received the fixed IP above
 488             out = subprocess.check_output(['ip', '-4', 'a', 'show', 'dev', self.iface])
 489             self.assertRegex(out, b'inet 192.168.5.210/24 .* scope global dynamic')
 490             # static hostname wins over transient one, thus *not* applied
 491             self.assertEqual(socket.gethostname(), orig_hostname)
 492         except AssertionError:
 493             self.show_journal('systemd-networkd.service')
 494             self.show_journal('systemd-hostnamed.service')
 495             self.print_server_log()
 496             raise
 497
 498
 499 class NetworkdClientTest(ClientTestBase, unittest.TestCase):
 500     '''Test networkd client against networkd server'''
 501
 502     def setUp(self):
 503         super().setUp()
 504         self.dnsmasq = None
 505
 506     def create_iface(self, ipv6=False, dhcpserver_opts=None):
 507         '''Create test interface with DHCP server behind it'''
 508
 509         # run "router-side" networkd in own mount namespace to shield it from
 510         # "client-side" configuration and networkd
 511         (fd, script) = tempfile.mkstemp(prefix='networkd-router.sh')
 512         self.addCleanup(os.remove, script)
 513         with os.fdopen(fd, 'w+') as f:
 514             f.write('''\
 515 #!/bin/sh -eu
 516 mkdir -p /run/systemd/network
 517 mkdir -p /run/systemd/netif
 518 mount -t tmpfs none /run/systemd/network
 519 mount -t tmpfs none /run/systemd/netif
 520 [ ! -e /run/dbus ] || mount -t tmpfs none /run/dbus
 521 # create router/client veth pair
 522 cat << EOF > /run/systemd/network/test.netdev
 523 [NetDev]
 524 Name=%(ifr)s
 525 Kind=veth
 526
 527 [Peer]
 528 Name=%(ifc)s
 529 EOF
 530
 531 cat << EOF > /run/systemd/network/test.network
 532 [Match]
 533 Name=%(ifr)s
 534
 535 [Network]
 536 Address=192.168.5.1/24
 537 %(addr6)s
 538 DHCPServer=yes
 539
 540 [DHCPServer]
 541 PoolOffset=10
 542 PoolSize=50
 543 DNS=192.168.5.1
 544 %(dhopts)s
 545 EOF
 546
 547 # run networkd as in systemd-networkd.service
 548 exec $(systemctl cat systemd-networkd.service | sed -n '/^ExecStart=/ { s/^.*=//; p}')
 549 ''' % {'ifr': self.if_router, 'ifc': self.iface, 'addr6': ipv6 and 'Address=2600::1/64' or '',
 550        'dhopts': dhcpserver_opts or ''})
 551
 552             os.fchmod(fd, 0o755)
 553
 554         subprocess.check_call(['systemd-run', '--unit=networkd-test-router.service',
 555                                '-p', 'InaccessibleDirectories=-/etc/systemd/network',
 556                                '-p', 'InaccessibleDirectories=-/run/systemd/network',
 557                                '-p', 'InaccessibleDirectories=-/run/systemd/netif',
 558                                '--service-type=notify', script])
 559
 560         # wait until devices got created
 561         for _ in range(50):
 562             out = subprocess.check_output(['ip', 'a', 'show', 'dev', self.if_router])
 563             if b'state UP' in out and b'scope global' in out:
 564                 break
 565             time.sleep(0.1)
 566
 567     def shutdown_iface(self):
 568         '''Remove test interface and stop DHCP server'''
 569
 570         if self.if_router:
 571             subprocess.check_call(['systemctl', 'stop', 'networkd-test-router.service'])
 572             # ensure failed transient unit does not stay around
 573             subprocess.call(['systemctl', 'reset-failed', 'networkd-test-router.service'])
 574             subprocess.call(['ip', 'link', 'del', 'dev', self.if_router])
 575             self.if_router = None
 576
 577     def print_server_log(self):
 578         '''Print DHCP server log for debugging failures'''
 579
 580         self.show_journal('networkd-test-router.service')
 581
 582     @unittest.skip('networkd does not have DHCPv6 server support')
 583     def test_hotplug_dhcp_ip6(self):
 584         pass
 585
 586     @unittest.skip('networkd does not have DHCPv6 server support')
 587     def test_coldplug_dhcp_ip6(self):
 588         pass
 589
 590     def test_search_domains(self):
 591
 592         # we don't use this interface for this test
 593         self.if_router = None
 594
 595         self.writeConfig('/run/systemd/network/test.netdev', '''\
 596 [NetDev]
 597 Name=dummy0
 598 Kind=dummy
 599 MACAddress=12:34:56:78:9a:bc''')
 600         self.writeConfig('/run/systemd/network/test.network', '''\
 601 [Match]
 602 Name=dummy0
 603 [Network]
 604 Address=192.168.42.100
 605 DNS=192.168.42.1
 606 Domains= one two three four five six seven eight nine ten''')
 607
 608         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 609
 610         for timeout in range(50):
 611             with open(RESOLV_CONF) as f:
 612                 contents = f.read()
 613             if ' one' in contents:
 614                 break
 615             time.sleep(0.1)
 616         self.assertRegex(contents, 'search .*one two three four')
 617         self.assertNotIn('seven\n', contents)
 618         self.assertIn('# Too many search domains configured, remaining ones ignored.\n', contents)
 619
 620     def test_search_domains_too_long(self):
 621
 622         # we don't use this interface for this test
 623         self.if_router = None
 624
 625         name_prefix = 'a' * 60
 626
 627         self.writeConfig('/run/systemd/network/test.netdev', '''\
 628 [NetDev]
 629 Name=dummy0
 630 Kind=dummy
 631 MACAddress=12:34:56:78:9a:bc''')
 632         self.writeConfig('/run/systemd/network/test.network', '''\
 633 [Match]
 634 Name=dummy0
 635 [Network]
 636 Address=192.168.42.100
 637 DNS=192.168.42.1
 638 Domains={p}0 {p}1 {p}2 {p}3 {p}4'''.format(p=name_prefix))
 639
 640         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 641
 642         for timeout in range(50):
 643             with open(RESOLV_CONF) as f:
 644                 contents = f.read()
 645             if ' one' in contents:
 646                 break
 647             time.sleep(0.1)
 648         self.assertRegex(contents, 'search .*{p}0 {p}1 {p}2'.format(p=name_prefix))
 649         self.assertIn('# Total length of all search domains is too long, remaining ones ignored.', contents)
 650
 651     def test_dropin(self):
 652         # we don't use this interface for this test
 653         self.if_router = None
 654
 655         self.writeConfig('/run/systemd/network/test.netdev', '''\
 656 [NetDev]
 657 Name=dummy0
 658 Kind=dummy
 659 MACAddress=12:34:56:78:9a:bc''')
 660         self.writeConfig('/run/systemd/network/test.network', '''\
 661 [Match]
 662 Name=dummy0
 663 [Network]
 664 Address=192.168.42.100
 665 DNS=192.168.42.1''')
 666         self.writeConfig('/run/systemd/network/test.network.d/dns.conf', '''\
 667 [Network]
 668 DNS=127.0.0.1''')
 669
 670         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 671
 672         for timeout in range(50):
 673             with open(RESOLV_CONF) as f:
 674                 contents = f.read()
 675             if ' 127.0.0.1' in contents:
 676                 break
 677             time.sleep(0.1)
 678         self.assertIn('nameserver 192.168.42.1\n', contents)
 679         self.assertIn('nameserver 127.0.0.1\n', contents)
 680
 681     def test_dhcp_timezone(self):
 682         '''networkd sets time zone from DHCP'''
 683
 684         def get_tz():
 685             out = subprocess.check_output(['busctl', 'get-property', 'org.freedesktop.timedate1',
 686                                            '/org/freedesktop/timedate1', 'org.freedesktop.timedate1', 'Timezone'])
 687             assert out.startswith(b's "')
 688             out = out.strip()
 689             assert out.endswith(b'"')
 690             return out[3:-1].decode()
 691
 692         orig_timezone = get_tz()
 693         self.addCleanup(subprocess.call, ['timedatectl', 'set-timezone', orig_timezone])
 694
 695         self.create_iface(dhcpserver_opts='EmitTimezone=yes\nTimezone=Pacific/Honolulu')
 696         self.do_test(coldplug=None, extra_opts='IPv6AcceptRA=false\n[DHCP]\nUseTimezone=true', dhcp_mode='ipv4')
 697
 698         # should have applied the received timezone
 699         try:
 700             self.assertEqual(get_tz(), 'Pacific/Honolulu')
 701         except AssertionError:
 702             self.show_journal('systemd-networkd.service')
 703             self.show_journal('systemd-hostnamed.service')
 704             raise
 705
 706
 707 if __name__ == '__main__':
 708     unittest.main(testRunner=unittest.TextTestRunner(stream=sys.stdout,
 709                                                      verbosity=2))