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