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 add_veth_pair(self, veth, peer, veth_options=(), peer_options=()):
  78         """Add a veth interface pair, and queue them to be removed."""
  79         subprocess.check_call(['ip', 'link', 'add', 'name', veth] +
  80                               list(veth_options) +
  81                               ['type', 'veth', 'peer', 'name', peer] +
  82                               list(peer_options))
  83         self.addCleanup(subprocess.call, ['ip', 'link', 'del', 'dev', peer])
  84
  85     def write_network(self, unit_name, contents):
  86         """Write a network unit file, and queue it to be removed."""
  87         unit_path = os.path.join(NETWORK_UNITDIR, unit_name)
  88
  89         with open(unit_path, 'w') as unit:
  90             unit.write(contents)
  91         self.addCleanup(os.remove, unit_path)
  92
  93     def write_network_dropin(self, unit_name, dropin_name, contents):
  94         """Write a network unit drop-in, and queue it to be removed."""
  95         dropin_dir = os.path.join(NETWORK_UNITDIR, "%s.d" % unit_name)
  96         dropin_path = os.path.join(dropin_dir, "%s.conf" % dropin_name)
  97
  98         os.makedirs(dropin_dir, exist_ok=True)
  99         with open(dropin_path, 'w') as dropin:
 100             dropin.write(contents)
 101         self.addCleanup(os.remove, dropin_path)
 102
 103     def assert_link_states(self, **kwargs):
 104         """Match networkctl link states to the given ones.
 105
 106         Each keyword argument should be the name of a network interface
 107         with its expected value of the "SETUP" column in output from
 108         networkctl.  The interfaces have five seconds to come online
 109         before the check is performed.  Every specified interface must
 110         be present in the output, and any other interfaces found in the
 111         output are ignored.
 112
 113         A special interface state "managed" is supported, which matches
 114         any value in the "SETUP" column other than "unmanaged".
 115         """
 116         if not kwargs:
 117             return
 118         interfaces = set(kwargs)
 119
 120         # Wait for the requested interfaces, but don't fail for them.
 121         subprocess.call([NETWORKD_WAIT_ONLINE, '--timeout=5'] +
 122                         ['--interface=%s' % iface for iface in kwargs])
 123
 124         # Validate each link state found in the networkctl output.
 125         out = subprocess.check_output(['networkctl', '--no-legend']).rstrip()
 126         for line in out.decode('utf-8').split('\n'):
 127             fields = line.split()
 128             if len(fields) >= 5 and fields[1] in kwargs:
 129                 iface = fields[1]
 130                 expected = kwargs[iface]
 131                 actual = fields[-1]
 132                 if (actual != expected and
 133                         not (expected == 'managed' and actual != 'unmanaged')):
 134                     self.fail("Link %s expects state %s, found %s" %
 135                               (iface, expected, actual))
 136                 interfaces.remove(iface)
 137
 138         # Ensure that all requested interfaces have been covered.
 139         if interfaces:
 140             self.fail("Missing links in status output: %s" % interfaces)
 141
 142
 143 class ClientTestBase(NetworkdTestingUtilities):
 144     """Provide common methods for testing networkd against servers."""
 145
 146     @classmethod
 147     def setUpClass(klass):
 148         klass.orig_log_level = subprocess.check_output(
 149             ['systemctl', 'show', '--value', '--property', 'LogLevel'],
 150             universal_newlines=True).strip()
 151         subprocess.check_call(['systemd-analyze', 'set-log-level', 'debug'])
 152
 153     @classmethod
 154     def tearDownClass(klass):
 155         subprocess.check_call(['systemd-analyze', 'set-log-level', klass.orig_log_level])
 156
 157     def setUp(self):
 158         self.iface = 'test_eth42'
 159         self.if_router = 'router_eth42'
 160         self.workdir_obj = tempfile.TemporaryDirectory()
 161         self.workdir = self.workdir_obj.name
 162         self.config = 'test_eth42.network'
 163
 164         # get current journal cursor
 165         subprocess.check_output(['journalctl', '--sync'])
 166         out = subprocess.check_output(['journalctl', '-b', '--quiet',
 167                                        '--no-pager', '-n0', '--show-cursor'],
 168                                       universal_newlines=True)
 169         self.assertTrue(out.startswith('-- cursor:'))
 170         self.journal_cursor = out.split()[-1]
 171
 172     def tearDown(self):
 173         self.shutdown_iface()
 174         subprocess.call(['systemctl', 'stop', 'systemd-networkd'])
 175         subprocess.call(['ip', 'link', 'del', 'dummy0'],
 176                         stderr=subprocess.DEVNULL)
 177
 178     def show_journal(self, unit):
 179         '''Show journal of given unit since start of the test'''
 180
 181         print('---- %s ----' % unit)
 182         subprocess.check_output(['journalctl', '--sync'])
 183         sys.stdout.flush()
 184         subprocess.call(['journalctl', '-b', '--no-pager', '--quiet',
 185                          '--cursor', self.journal_cursor, '-u', unit])
 186
 187     def create_iface(self, ipv6=False):
 188         '''Create test interface with DHCP server behind it'''
 189
 190         raise NotImplementedError('must be implemented by a subclass')
 191
 192     def shutdown_iface(self):
 193         '''Remove test interface and stop DHCP server'''
 194
 195         raise NotImplementedError('must be implemented by a subclass')
 196
 197     def print_server_log(self):
 198         '''Print DHCP server log for debugging failures'''
 199
 200         raise NotImplementedError('must be implemented by a subclass')
 201
 202     def do_test(self, coldplug=True, ipv6=False, extra_opts='',
 203                 online_timeout=10, dhcp_mode='yes'):
 204         subprocess.check_call(['systemctl', 'start', 'systemd-resolved'])
 205         self.write_network(self.config, '''\
 206 [Match]
 207 Name=%s
 208 [Network]
 209 DHCP=%s
 210 %s''' % (self.iface, dhcp_mode, extra_opts))
 211
 212         if coldplug:
 213             # create interface first, then start networkd
 214             self.create_iface(ipv6=ipv6)
 215             subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 216         elif coldplug is not None:
 217             # start networkd first, then create interface
 218             subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 219             self.create_iface(ipv6=ipv6)
 220         else:
 221             # "None" means test sets up interface by itself
 222             subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 223
 224         try:
 225             subprocess.check_call([NETWORKD_WAIT_ONLINE, '--interface',
 226                                    self.iface, '--timeout=%i' % online_timeout])
 227
 228             if ipv6:
 229                 # check iface state and IP 6 address; FIXME: we need to wait a bit
 230                 # longer, as the iface is "configured" already with IPv4 *or*
 231                 # IPv6, but we want to wait for both
 232                 for _ in range(10):
 233                     out = subprocess.check_output(['ip', 'a', 'show', 'dev', self.iface])
 234                     if b'state UP' in out and b'inet6 2600' in out and b'inet 192.168' in out:
 235                         break
 236                     time.sleep(1)
 237                 else:
 238                     self.fail('timed out waiting for IPv6 configuration')
 239
 240                 self.assertRegex(out, b'inet6 2600::.* scope global .*dynamic')
 241                 self.assertRegex(out, b'inet6 fe80::.* scope link')
 242             else:
 243                 # should have link-local address on IPv6 only
 244                 out = subprocess.check_output(['ip', '-6', 'a', 'show', 'dev', self.iface])
 245                 self.assertRegex(out, br'inet6 fe80::.* scope link')
 246                 self.assertNotIn(b'scope global', out)
 247
 248             # should have IPv4 address
 249             out = subprocess.check_output(['ip', '-4', 'a', 'show', 'dev', self.iface])
 250             self.assertIn(b'state UP', out)
 251             self.assertRegex(out, br'inet 192.168.5.\d+/.* scope global dynamic')
 252
 253             # check networkctl state
 254             out = subprocess.check_output(['networkctl'])
 255             self.assertRegex(out, (r'%s\s+ether\s+routable\s+unmanaged' % self.if_router).encode())
 256             self.assertRegex(out, (r'%s\s+ether\s+routable\s+configured' % self.iface).encode())
 257
 258             out = subprocess.check_output(['networkctl', 'status', self.iface])
 259             self.assertRegex(out, br'Type:\s+ether')
 260             self.assertRegex(out, br'State:\s+routable.*configured')
 261             self.assertRegex(out, br'Address:\s+192.168.5.\d+')
 262             if ipv6:
 263                 self.assertRegex(out, br'2600::')
 264             else:
 265                 self.assertNotIn(br'2600::', out)
 266             self.assertRegex(out, br'fe80::')
 267             self.assertRegex(out, br'Gateway:\s+192.168.5.1')
 268             self.assertRegex(out, br'DNS:\s+192.168.5.1')
 269         except (AssertionError, subprocess.CalledProcessError):
 270             # show networkd status, journal, and DHCP server log on failure
 271             with open(os.path.join(NETWORK_UNITDIR, self.config)) as f:
 272                 print('\n---- %s ----\n%s' % (self.config, f.read()))
 273             print('---- interface status ----')
 274             sys.stdout.flush()
 275             subprocess.call(['ip', 'a', 'show', 'dev', self.iface])
 276             print('---- networkctl status %s ----' % self.iface)
 277             sys.stdout.flush()
 278             subprocess.call(['networkctl', 'status', self.iface])
 279             self.show_journal('systemd-networkd.service')
 280             self.print_server_log()
 281             raise
 282
 283         for timeout in range(50):
 284             with open(RESOLV_CONF) as f:
 285                 contents = f.read()
 286             if 'nameserver 192.168.5.1\n' in contents:
 287                 break
 288             time.sleep(0.1)
 289         else:
 290             self.fail('nameserver 192.168.5.1 not found in ' + RESOLV_CONF)
 291
 292         if coldplug is False:
 293             # check post-down.d hook
 294             self.shutdown_iface()
 295
 296     def test_coldplug_dhcp_yes_ip4(self):
 297         # we have a 12s timeout on RA, so we need to wait longer
 298         self.do_test(coldplug=True, ipv6=False, online_timeout=15)
 299
 300     def test_coldplug_dhcp_yes_ip4_no_ra(self):
 301         # with disabling RA explicitly things should be fast
 302         self.do_test(coldplug=True, ipv6=False,
 303                      extra_opts='IPv6AcceptRA=False')
 304
 305     def test_coldplug_dhcp_ip4_only(self):
 306         # we have a 12s timeout on RA, so we need to wait longer
 307         self.do_test(coldplug=True, ipv6=False, dhcp_mode='ipv4',
 308                      online_timeout=15)
 309
 310     def test_coldplug_dhcp_ip4_only_no_ra(self):
 311         # with disabling RA explicitly things should be fast
 312         self.do_test(coldplug=True, ipv6=False, dhcp_mode='ipv4',
 313                      extra_opts='IPv6AcceptRA=False')
 314
 315     def test_coldplug_dhcp_ip6(self):
 316         self.do_test(coldplug=True, ipv6=True)
 317
 318     def test_hotplug_dhcp_ip4(self):
 319         # With IPv4 only we have a 12s timeout on RA, so we need to wait longer
 320         self.do_test(coldplug=False, ipv6=False, online_timeout=15)
 321
 322     def test_hotplug_dhcp_ip6(self):
 323         self.do_test(coldplug=False, ipv6=True)
 324
 325     def test_route_only_dns(self):
 326         self.write_network('myvpn.netdev', '''\
 327 [NetDev]
 328 Name=dummy0
 329 Kind=dummy
 330 MACAddress=12:34:56:78:9a:bc''')
 331         self.write_network('myvpn.network', '''\
 332 [Match]
 333 Name=dummy0
 334 [Network]
 335 Address=192.168.42.100
 336 DNS=192.168.42.1
 337 Domains= ~company''')
 338
 339         self.do_test(coldplug=True, ipv6=False,
 340                      extra_opts='IPv6AcceptRouterAdvertisements=False')
 341
 342         with open(RESOLV_CONF) as f:
 343             contents = f.read()
 344             # ~company is not a search domain, only a routing domain
 345             self.assertNotRegex(contents, 'search.*company')
 346             # our global server should appear
 347             self.assertIn('nameserver 192.168.5.1\n', contents)
 348             # should not have domain-restricted server as global server
 349             self.assertNotIn('nameserver 192.168.42.1\n', contents)
 350
 351     def test_route_only_dns_all_domains(self):
 352         self.write_network('myvpn.netdev', '''[NetDev]
 353 Name=dummy0
 354 Kind=dummy
 355 MACAddress=12:34:56:78:9a:bc''')
 356         self.write_network('myvpn.network', '''[Match]
 357 Name=dummy0
 358 [Network]
 359 Address=192.168.42.100
 360 DNS=192.168.42.1
 361 Domains= ~company ~.''')
 362
 363         self.do_test(coldplug=True, ipv6=False,
 364                      extra_opts='IPv6AcceptRouterAdvertisements=False')
 365
 366         with open(RESOLV_CONF) as f:
 367             contents = f.read()
 368
 369         # ~company is not a search domain, only a routing domain
 370         self.assertNotRegex(contents, 'search.*company')
 371
 372         # our global server should appear
 373         self.assertIn('nameserver 192.168.5.1\n', contents)
 374         # should have company server as global server due to ~.
 375         self.assertIn('nameserver 192.168.42.1\n', contents)
 376
 377
 378 @unittest.skipUnless(HAVE_DNSMASQ, 'dnsmasq not installed')
 379 class DnsmasqClientTest(ClientTestBase, unittest.TestCase):
 380     '''Test networkd client against dnsmasq'''
 381
 382     def setUp(self):
 383         super().setUp()
 384         self.dnsmasq = None
 385         self.iface_mac = 'de:ad:be:ef:47:11'
 386
 387     def create_iface(self, ipv6=False, dnsmasq_opts=None):
 388         '''Create test interface with DHCP server behind it'''
 389
 390         # add veth pair
 391         subprocess.check_call(['ip', 'link', 'add', 'name', self.iface,
 392                                'address', self.iface_mac,
 393                                'type', 'veth', 'peer', 'name', self.if_router])
 394
 395         # give our router an IP
 396         subprocess.check_call(['ip', 'a', 'flush', 'dev', self.if_router])
 397         subprocess.check_call(['ip', 'a', 'add', '192.168.5.1/24', 'dev', self.if_router])
 398         if ipv6:
 399             subprocess.check_call(['ip', 'a', 'add', '2600::1/64', 'dev', self.if_router])
 400         subprocess.check_call(['ip', 'link', 'set', self.if_router, 'up'])
 401
 402         # add DHCP server
 403         self.dnsmasq_log = os.path.join(self.workdir, 'dnsmasq.log')
 404         lease_file = os.path.join(self.workdir, 'dnsmasq.leases')
 405         if ipv6:
 406             extra_opts = ['--enable-ra', '--dhcp-range=2600::10,2600::20']
 407         else:
 408             extra_opts = []
 409         if dnsmasq_opts:
 410             extra_opts += dnsmasq_opts
 411         self.dnsmasq = subprocess.Popen(
 412             ['dnsmasq', '--keep-in-foreground', '--log-queries',
 413              '--log-facility=' + self.dnsmasq_log, '--conf-file=/dev/null',
 414              '--dhcp-leasefile=' + lease_file, '--bind-interfaces',
 415              '--interface=' + self.if_router, '--except-interface=lo',
 416              '--dhcp-range=192.168.5.10,192.168.5.200'] + extra_opts)
 417
 418     def shutdown_iface(self):
 419         '''Remove test interface and stop DHCP server'''
 420
 421         if self.if_router:
 422             subprocess.check_call(['ip', 'link', 'del', 'dev', self.if_router])
 423             self.if_router = None
 424         if self.dnsmasq:
 425             self.dnsmasq.kill()
 426             self.dnsmasq.wait()
 427             self.dnsmasq = None
 428
 429     def print_server_log(self):
 430         '''Print DHCP server log for debugging failures'''
 431
 432         with open(self.dnsmasq_log) as f:
 433             sys.stdout.write('\n\n---- dnsmasq log ----\n%s\n------\n\n' % f.read())
 434
 435     def test_resolved_domain_restricted_dns(self):
 436         '''resolved: domain-restricted DNS servers'''
 437
 438         # create interface for generic connections; this will map all DNS names
 439         # to 192.168.42.1
 440         self.create_iface(dnsmasq_opts=['--address=/#/192.168.42.1'])
 441         self.write_network('general.network', '''\
 442 [Match]
 443 Name=%s
 444 [Network]
 445 DHCP=ipv4
 446 IPv6AcceptRA=False''' % self.iface)
 447
 448         # create second device/dnsmasq for a .company/.lab VPN interface
 449         # static IPs for simplicity
 450         self.add_veth_pair('testvpnclient', 'testvpnrouter')
 451         subprocess.check_call(['ip', 'a', 'flush', 'dev', 'testvpnrouter'])
 452         subprocess.check_call(['ip', 'a', 'add', '10.241.3.1/24', 'dev', 'testvpnrouter'])
 453         subprocess.check_call(['ip', 'link', 'set', 'testvpnrouter', 'up'])
 454
 455         vpn_dnsmasq_log = os.path.join(self.workdir, 'dnsmasq-vpn.log')
 456         vpn_dnsmasq = subprocess.Popen(
 457             ['dnsmasq', '--keep-in-foreground', '--log-queries',
 458              '--log-facility=' + vpn_dnsmasq_log, '--conf-file=/dev/null',
 459              '--dhcp-leasefile=/dev/null', '--bind-interfaces',
 460              '--interface=testvpnrouter', '--except-interface=lo',
 461              '--address=/math.lab/10.241.3.3', '--address=/cantina.company/10.241.4.4'])
 462         self.addCleanup(vpn_dnsmasq.wait)
 463         self.addCleanup(vpn_dnsmasq.kill)
 464
 465         self.write_network('vpn.network', '''\
 466 [Match]
 467 Name=testvpnclient
 468 [Network]
 469 IPv6AcceptRA=False
 470 Address=10.241.3.2/24
 471 DNS=10.241.3.1
 472 Domains= ~company ~lab''')
 473
 474         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 475         subprocess.check_call([NETWORKD_WAIT_ONLINE, '--interface', self.iface,
 476                                '--interface=testvpnclient', '--timeout=20'])
 477
 478         # ensure we start fresh with every test
 479         subprocess.check_call(['systemctl', 'restart', 'systemd-resolved'])
 480
 481         # test vpnclient specific domains; these should *not* be answered by
 482         # the general DNS
 483         out = subprocess.check_output(['systemd-resolve', 'math.lab'])
 484         self.assertIn(b'math.lab: 10.241.3.3', out)
 485         out = subprocess.check_output(['systemd-resolve', 'kettle.cantina.company'])
 486         self.assertIn(b'kettle.cantina.company: 10.241.4.4', out)
 487
 488         # test general domains
 489         out = subprocess.check_output(['systemd-resolve', 'megasearch.net'])
 490         self.assertIn(b'megasearch.net: 192.168.42.1', out)
 491
 492         with open(self.dnsmasq_log) as f:
 493             general_log = f.read()
 494         with open(vpn_dnsmasq_log) as f:
 495             vpn_log = f.read()
 496
 497         # VPN domains should only be sent to VPN DNS
 498         self.assertRegex(vpn_log, 'query.*math.lab')
 499         self.assertRegex(vpn_log, 'query.*cantina.company')
 500         self.assertNotIn('lab', general_log)
 501         self.assertNotIn('company', general_log)
 502
 503         # general domains should not be sent to the VPN DNS
 504         self.assertRegex(general_log, 'query.*megasearch.net')
 505         self.assertNotIn('megasearch.net', vpn_log)
 506
 507     def test_transient_hostname(self):
 508         '''networkd sets transient hostname from DHCP'''
 509
 510         orig_hostname = socket.gethostname()
 511         self.addCleanup(socket.sethostname, orig_hostname)
 512         # temporarily move /etc/hostname away; restart hostnamed to pick it up
 513         if os.path.exists('/etc/hostname'):
 514             subprocess.check_call(['mount', '--bind', '/dev/null', '/etc/hostname'])
 515             self.addCleanup(subprocess.call, ['umount', '/etc/hostname'])
 516         subprocess.check_call(['systemctl', 'stop', 'systemd-hostnamed.service'])
 517
 518         self.create_iface(dnsmasq_opts=['--dhcp-host=%s,192.168.5.210,testgreen' % self.iface_mac])
 519         self.do_test(coldplug=None, extra_opts='IPv6AcceptRA=False', dhcp_mode='ipv4')
 520
 521         try:
 522             # should have received the fixed IP above
 523             out = subprocess.check_output(['ip', '-4', 'a', 'show', 'dev', self.iface])
 524             self.assertRegex(out, b'inet 192.168.5.210/24 .* scope global dynamic')
 525             # should have set transient hostname in hostnamed; this is
 526             # sometimes a bit lagging (issue #4753), so retry a few times
 527             for retry in range(1, 6):
 528                 out = subprocess.check_output(['hostnamectl'])
 529                 if b'testgreen' in out:
 530                     break
 531                 time.sleep(5)
 532                 sys.stdout.write('[retry %i] ' % retry)
 533                 sys.stdout.flush()
 534             else:
 535                 self.fail('Transient hostname not found in hostnamectl:\n%s' % out.decode())
 536             # and also applied to the system
 537             self.assertEqual(socket.gethostname(), 'testgreen')
 538         except AssertionError:
 539             self.show_journal('systemd-networkd.service')
 540             self.show_journal('systemd-hostnamed.service')
 541             self.print_server_log()
 542             raise
 543
 544     def test_transient_hostname_with_static(self):
 545         '''transient hostname is not applied if static hostname exists'''
 546
 547         orig_hostname = socket.gethostname()
 548         self.addCleanup(socket.sethostname, orig_hostname)
 549         if not os.path.exists('/etc/hostname'):
 550             self.writeConfig('/etc/hostname', orig_hostname)
 551         subprocess.check_call(['systemctl', 'stop', 'systemd-hostnamed.service'])
 552
 553         self.create_iface(dnsmasq_opts=['--dhcp-host=%s,192.168.5.210,testgreen' % self.iface_mac])
 554         self.do_test(coldplug=None, extra_opts='IPv6AcceptRA=False', dhcp_mode='ipv4')
 555
 556         try:
 557             # should have received the fixed IP above
 558             out = subprocess.check_output(['ip', '-4', 'a', 'show', 'dev', self.iface])
 559             self.assertRegex(out, b'inet 192.168.5.210/24 .* scope global dynamic')
 560             # static hostname wins over transient one, thus *not* applied
 561             self.assertEqual(socket.gethostname(), orig_hostname)
 562         except AssertionError:
 563             self.show_journal('systemd-networkd.service')
 564             self.show_journal('systemd-hostnamed.service')
 565             self.print_server_log()
 566             raise
 567
 568
 569 class NetworkdClientTest(ClientTestBase, unittest.TestCase):
 570     '''Test networkd client against networkd server'''
 571
 572     def setUp(self):
 573         super().setUp()
 574         self.dnsmasq = None
 575
 576     def create_iface(self, ipv6=False, dhcpserver_opts=None):
 577         '''Create test interface with DHCP server behind it'''
 578
 579         # run "router-side" networkd in own mount namespace to shield it from
 580         # "client-side" configuration and networkd
 581         (fd, script) = tempfile.mkstemp(prefix='networkd-router.sh')
 582         self.addCleanup(os.remove, script)
 583         with os.fdopen(fd, 'w+') as f:
 584             f.write('''\
 585 #!/bin/sh -eu
 586 mkdir -p /run/systemd/network
 587 mkdir -p /run/systemd/netif
 588 mount -t tmpfs none /run/systemd/network
 589 mount -t tmpfs none /run/systemd/netif
 590 [ ! -e /run/dbus ] || mount -t tmpfs none /run/dbus
 591 # create router/client veth pair
 592 cat << EOF > /run/systemd/network/test.netdev
 593 [NetDev]
 594 Name=%(ifr)s
 595 Kind=veth
 596
 597 [Peer]
 598 Name=%(ifc)s
 599 EOF
 600
 601 cat << EOF > /run/systemd/network/test.network
 602 [Match]
 603 Name=%(ifr)s
 604
 605 [Network]
 606 Address=192.168.5.1/24
 607 %(addr6)s
 608 DHCPServer=yes
 609
 610 [DHCPServer]
 611 PoolOffset=10
 612 PoolSize=50
 613 DNS=192.168.5.1
 614 %(dhopts)s
 615 EOF
 616
 617 # run networkd as in systemd-networkd.service
 618 exec $(systemctl cat systemd-networkd.service | sed -n '/^ExecStart=/ { s/^.*=//; p}')
 619 ''' % {'ifr': self.if_router, 'ifc': self.iface, 'addr6': ipv6 and 'Address=2600::1/64' or '',
 620        'dhopts': dhcpserver_opts or ''})
 621
 622             os.fchmod(fd, 0o755)
 623
 624         subprocess.check_call(['systemd-run', '--unit=networkd-test-router.service',
 625                                '-p', 'InaccessibleDirectories=-/etc/systemd/network',
 626                                '-p', 'InaccessibleDirectories=-/run/systemd/network',
 627                                '-p', 'InaccessibleDirectories=-/run/systemd/netif',
 628                                '--service-type=notify', script])
 629
 630         # wait until devices got created
 631         for _ in range(50):
 632             out = subprocess.check_output(['ip', 'a', 'show', 'dev', self.if_router])
 633             if b'state UP' in out and b'scope global' in out:
 634                 break
 635             time.sleep(0.1)
 636
 637     def shutdown_iface(self):
 638         '''Remove test interface and stop DHCP server'''
 639
 640         if self.if_router:
 641             subprocess.check_call(['systemctl', 'stop', 'networkd-test-router.service'])
 642             # ensure failed transient unit does not stay around
 643             subprocess.call(['systemctl', 'reset-failed', 'networkd-test-router.service'])
 644             subprocess.call(['ip', 'link', 'del', 'dev', self.if_router])
 645             self.if_router = None
 646
 647     def print_server_log(self):
 648         '''Print DHCP server log for debugging failures'''
 649
 650         self.show_journal('networkd-test-router.service')
 651
 652     @unittest.skip('networkd does not have DHCPv6 server support')
 653     def test_hotplug_dhcp_ip6(self):
 654         pass
 655
 656     @unittest.skip('networkd does not have DHCPv6 server support')
 657     def test_coldplug_dhcp_ip6(self):
 658         pass
 659
 660     def test_search_domains(self):
 661
 662         # we don't use this interface for this test
 663         self.if_router = None
 664
 665         self.write_network('test.netdev', '''\
 666 [NetDev]
 667 Name=dummy0
 668 Kind=dummy
 669 MACAddress=12:34:56:78:9a:bc''')
 670         self.write_network('test.network', '''\
 671 [Match]
 672 Name=dummy0
 673 [Network]
 674 Address=192.168.42.100
 675 DNS=192.168.42.1
 676 Domains= one two three four five six seven eight nine ten''')
 677
 678         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 679
 680         for timeout in range(50):
 681             with open(RESOLV_CONF) as f:
 682                 contents = f.read()
 683             if ' one' in contents:
 684                 break
 685             time.sleep(0.1)
 686         self.assertRegex(contents, 'search .*one two three four')
 687         self.assertNotIn('seven\n', contents)
 688         self.assertIn('# Too many search domains configured, remaining ones ignored.\n', contents)
 689
 690     def test_search_domains_too_long(self):
 691
 692         # we don't use this interface for this test
 693         self.if_router = None
 694
 695         name_prefix = 'a' * 60
 696
 697         self.write_network('test.netdev', '''\
 698 [NetDev]
 699 Name=dummy0
 700 Kind=dummy
 701 MACAddress=12:34:56:78:9a:bc''')
 702         self.write_network('test.network', '''\
 703 [Match]
 704 Name=dummy0
 705 [Network]
 706 Address=192.168.42.100
 707 DNS=192.168.42.1
 708 Domains={p}0 {p}1 {p}2 {p}3 {p}4'''.format(p=name_prefix))
 709
 710         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 711
 712         for timeout in range(50):
 713             with open(RESOLV_CONF) as f:
 714                 contents = f.read()
 715             if ' one' in contents:
 716                 break
 717             time.sleep(0.1)
 718         self.assertRegex(contents, 'search .*{p}0 {p}1 {p}2'.format(p=name_prefix))
 719         self.assertIn('# Total length of all search domains is too long, remaining ones ignored.', contents)
 720
 721     def test_dropin(self):
 722         # we don't use this interface for this test
 723         self.if_router = None
 724
 725         self.write_network('test.netdev', '''\
 726 [NetDev]
 727 Name=dummy0
 728 Kind=dummy
 729 MACAddress=12:34:56:78:9a:bc''')
 730         self.write_network('test.network', '''\
 731 [Match]
 732 Name=dummy0
 733 [Network]
 734 Address=192.168.42.100
 735 DNS=192.168.42.1''')
 736         self.write_network_dropin('test.network', 'dns', '''\
 737 [Network]
 738 DNS=127.0.0.1''')
 739
 740         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 741
 742         for timeout in range(50):
 743             with open(RESOLV_CONF) as f:
 744                 contents = f.read()
 745             if ' 127.0.0.1' in contents:
 746                 break
 747             time.sleep(0.1)
 748         self.assertIn('nameserver 192.168.42.1\n', contents)
 749         self.assertIn('nameserver 127.0.0.1\n', contents)
 750
 751     def test_dhcp_timezone(self):
 752         '''networkd sets time zone from DHCP'''
 753
 754         def get_tz():
 755             out = subprocess.check_output(['busctl', 'get-property', 'org.freedesktop.timedate1',
 756                                            '/org/freedesktop/timedate1', 'org.freedesktop.timedate1', 'Timezone'])
 757             assert out.startswith(b's "')
 758             out = out.strip()
 759             assert out.endswith(b'"')
 760             return out[3:-1].decode()
 761
 762         orig_timezone = get_tz()
 763         self.addCleanup(subprocess.call, ['timedatectl', 'set-timezone', orig_timezone])
 764
 765         self.create_iface(dhcpserver_opts='EmitTimezone=yes\nTimezone=Pacific/Honolulu')
 766         self.do_test(coldplug=None, extra_opts='IPv6AcceptRA=false\n[DHCP]\nUseTimezone=true', dhcp_mode='ipv4')
 767
 768         # should have applied the received timezone
 769         try:
 770             self.assertEqual(get_tz(), 'Pacific/Honolulu')
 771         except AssertionError:
 772             self.show_journal('systemd-networkd.service')
 773             self.show_journal('systemd-hostnamed.service')
 774             raise
 775
 776
 777 class MatchClientTest(unittest.TestCase, NetworkdTestingUtilities):
 778     """Test [Match] sections in .network files.
 779
 780     Be aware that matching the test host's interfaces will wipe their
 781     configuration, so as a precaution, all network files should have a
 782     restrictive [Match] section to only ever interfere with the
 783     temporary veth interfaces created here.
 784     """
 785
 786     def tearDown(self):
 787         """Stop networkd."""
 788         subprocess.call(['systemctl', 'stop', 'systemd-networkd'])
 789
 790     def test_basic_matching(self):
 791         """Verify the Name= line works throughout this class."""
 792         self.add_veth_pair('test_if1', 'fake_if2')
 793         self.write_network('test.network', "[Match]\nName=test_*\n[Network]")
 794         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 795         self.assert_link_states(test_if1='managed', fake_if2='unmanaged')
 796
 797     def test_inverted_matching(self):
 798         """Verify that a '!'-prefixed value inverts the match."""
 799         # Use a MAC address as the interfaces' common matching attribute
 800         # to avoid depending on udev, to support testing in containers.
 801         mac = '00:01:02:03:98:99'
 802         self.add_veth_pair('test_veth', 'test_peer',
 803                            ['addr', mac], ['addr', mac])
 804         self.write_network('no-veth.network', """\
 805 [Match]
 806 MACAddress=%s
 807 Name=!nonexistent *peer*
 808 [Network]""" % mac)
 809         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 810         self.assert_link_states(test_veth='managed', test_peer='unmanaged')
 811
 812
 813 class UnmanagedClientTest(unittest.TestCase, NetworkdTestingUtilities):
 814     """Test if networkd manages the correct interfaces."""
 815
 816     def setUp(self):
 817         """Write .network files to match the named veth devices."""
 818         # Define the veth+peer pairs to be created.
 819         # Their pairing doesn't actually matter, only their names do.
 820         self.veths = {
 821             'm1def': 'm0unm',
 822             'm1man': 'm1unm',
 823         }
 824
 825         # Define the contents of .network files to be read in order.
 826         self.configs = (
 827             "[Match]\nName=m1def\n",
 828             "[Match]\nName=m1unm\n[Link]\nUnmanaged=yes\n",
 829             "[Match]\nName=m1*\n[Link]\nUnmanaged=no\n",
 830         )
 831
 832         # Write out the .network files to be cleaned up automatically.
 833         for i, config in enumerate(self.configs):
 834             self.write_network("%02d-test.network" % i, config)
 835
 836     def tearDown(self):
 837         """Stop networkd."""
 838         subprocess.call(['systemctl', 'stop', 'systemd-networkd'])
 839
 840     def create_iface(self):
 841         """Create temporary veth pairs for interface matching."""
 842         for veth, peer in self.veths.items():
 843             self.add_veth_pair(veth, peer)
 844
 845     def test_unmanaged_setting(self):
 846         """Verify link states with Unmanaged= settings, hot-plug."""
 847         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 848         self.create_iface()
 849         self.assert_link_states(m1def='managed',
 850                                 m1man='managed',
 851                                 m1unm='unmanaged',
 852                                 m0unm='unmanaged')
 853
 854     def test_unmanaged_setting_coldplug(self):
 855         """Verify link states with Unmanaged= settings, cold-plug."""
 856         self.create_iface()
 857         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 858         self.assert_link_states(m1def='managed',
 859                                 m1man='managed',
 860                                 m1unm='unmanaged',
 861                                 m0unm='unmanaged')
 862
 863     def test_catchall_config(self):
 864         """Verify link states with a catch-all config, hot-plug."""
 865         # Don't actually catch ALL interfaces.  It messes up the host.
 866         self.write_network('all.network', "[Match]\nName=m[01]???\n")
 867         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 868         self.create_iface()
 869         self.assert_link_states(m1def='managed',
 870                                 m1man='managed',
 871                                 m1unm='unmanaged',
 872                                 m0unm='managed')
 873
 874     def test_catchall_config_coldplug(self):
 875         """Verify link states with a catch-all config, cold-plug."""
 876         # Don't actually catch ALL interfaces.  It messes up the host.
 877         self.write_network('all.network', "[Match]\nName=m[01]???\n")
 878         self.create_iface()
 879         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 880         self.assert_link_states(m1def='managed',
 881                                 m1man='managed',
 882                                 m1unm='unmanaged',
 883                                 m0unm='managed')
 884
 885
 886 if __name__ == '__main__':
 887     unittest.main(testRunner=unittest.TextTestRunner(stream=sys.stdout,
 888                                                      verbosity=2))