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_resolved_etc_hosts(self):
 508         '''resolved queries to /etc/hosts'''
 509
 510         # FIXME: -t MX query fails with enabled DNSSEC (even when using
 511         # the known negative trust anchor .internal instead of .example)
 512         conf = '/run/systemd/resolved.conf.d/test-disable-dnssec.conf'
 513         os.makedirs(os.path.dirname(conf), exist_ok=True)
 514         with open(conf, 'w') as f:
 515             f.write('[Resolve]\nDNSSEC=no')
 516         self.addCleanup(os.remove, conf)
 517
 518         # create /etc/hosts bind mount which resolves my.example for IPv4
 519         hosts = os.path.join(self.workdir, 'hosts')
 520         with open(hosts, 'w') as f:
 521             f.write('172.16.99.99  my.example\n')
 522         subprocess.check_call(['mount', '--bind', hosts, '/etc/hosts'])
 523         self.addCleanup(subprocess.call, ['umount', '/etc/hosts'])
 524         subprocess.check_call(['systemctl', 'stop', 'systemd-resolved.service'])
 525
 526         # note: different IPv4 address here, so that it's easy to tell apart
 527         # what resolved the query
 528         self.create_iface(dnsmasq_opts=['--host-record=my.example,172.16.99.1,2600::99:99',
 529                                         '--host-record=other.example,172.16.0.42,2600::42',
 530                                         '--mx-host=example,mail.example'],
 531                           ipv6=True)
 532         self.do_test(coldplug=None, ipv6=True)
 533
 534         try:
 535             # family specific queries
 536             out = subprocess.check_output(['systemd-resolve', '-4', 'my.example'])
 537             self.assertIn(b'my.example: 172.16.99.99', out)
 538             # we don't expect an IPv6 answer; if /etc/hosts has any IP address,
 539             # it's considered a sufficient source
 540             self.assertNotEqual(subprocess.call(['systemd-resolve', '-6', 'my.example']), 0)
 541             # "any family" query; IPv4 should come from /etc/hosts
 542             out = subprocess.check_output(['systemd-resolve', 'my.example'])
 543             self.assertIn(b'my.example: 172.16.99.99', out)
 544             # IP → name lookup; again, takes the /etc/hosts one
 545             out = subprocess.check_output(['systemd-resolve', '172.16.99.99'])
 546             self.assertIn(b'172.16.99.99: my.example', out)
 547
 548             # non-address RRs should fall back to DNS
 549             out = subprocess.check_output(['systemd-resolve', '--type=MX', 'example'])
 550             self.assertIn(b'example IN MX 1 mail.example', out)
 551
 552             # other domains query DNS
 553             out = subprocess.check_output(['systemd-resolve', 'other.example'])
 554             self.assertIn(b'172.16.0.42', out)
 555             out = subprocess.check_output(['systemd-resolve', '172.16.0.42'])
 556             self.assertIn(b'172.16.0.42: other.example', out)
 557         except (AssertionError, subprocess.CalledProcessError):
 558             self.show_journal('systemd-resolved.service')
 559             self.print_server_log()
 560             raise
 561
 562     def test_transient_hostname(self):
 563         '''networkd sets transient hostname from DHCP'''
 564
 565         orig_hostname = socket.gethostname()
 566         self.addCleanup(socket.sethostname, orig_hostname)
 567         # temporarily move /etc/hostname away; restart hostnamed to pick it up
 568         if os.path.exists('/etc/hostname'):
 569             subprocess.check_call(['mount', '--bind', '/dev/null', '/etc/hostname'])
 570             self.addCleanup(subprocess.call, ['umount', '/etc/hostname'])
 571         subprocess.check_call(['systemctl', 'stop', 'systemd-hostnamed.service'])
 572
 573         self.create_iface(dnsmasq_opts=['--dhcp-host=%s,192.168.5.210,testgreen' % self.iface_mac])
 574         self.do_test(coldplug=None, extra_opts='IPv6AcceptRA=False', dhcp_mode='ipv4')
 575
 576         try:
 577             # should have received the fixed IP above
 578             out = subprocess.check_output(['ip', '-4', 'a', 'show', 'dev', self.iface])
 579             self.assertRegex(out, b'inet 192.168.5.210/24 .* scope global dynamic')
 580             # should have set transient hostname in hostnamed; this is
 581             # sometimes a bit lagging (issue #4753), so retry a few times
 582             for retry in range(1, 6):
 583                 out = subprocess.check_output(['hostnamectl'])
 584                 if b'testgreen' in out:
 585                     break
 586                 time.sleep(5)
 587                 sys.stdout.write('[retry %i] ' % retry)
 588                 sys.stdout.flush()
 589             else:
 590                 self.fail('Transient hostname not found in hostnamectl:\n%s' % out.decode())
 591             # and also applied to the system
 592             self.assertEqual(socket.gethostname(), 'testgreen')
 593         except AssertionError:
 594             self.show_journal('systemd-networkd.service')
 595             self.show_journal('systemd-hostnamed.service')
 596             self.print_server_log()
 597             raise
 598
 599     def test_transient_hostname_with_static(self):
 600         '''transient hostname is not applied if static hostname exists'''
 601
 602         orig_hostname = socket.gethostname()
 603         self.addCleanup(socket.sethostname, orig_hostname)
 604         if not os.path.exists('/etc/hostname'):
 605             self.writeConfig('/etc/hostname', orig_hostname)
 606         subprocess.check_call(['systemctl', 'stop', 'systemd-hostnamed.service'])
 607
 608         self.create_iface(dnsmasq_opts=['--dhcp-host=%s,192.168.5.210,testgreen' % self.iface_mac])
 609         self.do_test(coldplug=None, extra_opts='IPv6AcceptRA=False', dhcp_mode='ipv4')
 610
 611         try:
 612             # should have received the fixed IP above
 613             out = subprocess.check_output(['ip', '-4', 'a', 'show', 'dev', self.iface])
 614             self.assertRegex(out, b'inet 192.168.5.210/24 .* scope global dynamic')
 615             # static hostname wins over transient one, thus *not* applied
 616             self.assertEqual(socket.gethostname(), orig_hostname)
 617         except AssertionError:
 618             self.show_journal('systemd-networkd.service')
 619             self.show_journal('systemd-hostnamed.service')
 620             self.print_server_log()
 621             raise
 622
 623
 624 class NetworkdClientTest(ClientTestBase, unittest.TestCase):
 625     '''Test networkd client against networkd server'''
 626
 627     def setUp(self):
 628         super().setUp()
 629         self.dnsmasq = None
 630
 631     def create_iface(self, ipv6=False, dhcpserver_opts=None):
 632         '''Create test interface with DHCP server behind it'''
 633
 634         # run "router-side" networkd in own mount namespace to shield it from
 635         # "client-side" configuration and networkd
 636         (fd, script) = tempfile.mkstemp(prefix='networkd-router.sh')
 637         self.addCleanup(os.remove, script)
 638         with os.fdopen(fd, 'w+') as f:
 639             f.write('''\
 640 #!/bin/sh -eu
 641 mkdir -p /run/systemd/network
 642 mkdir -p /run/systemd/netif
 643 mount -t tmpfs none /run/systemd/network
 644 mount -t tmpfs none /run/systemd/netif
 645 [ ! -e /run/dbus ] || mount -t tmpfs none /run/dbus
 646 # create router/client veth pair
 647 cat << EOF > /run/systemd/network/test.netdev
 648 [NetDev]
 649 Name=%(ifr)s
 650 Kind=veth
 651
 652 [Peer]
 653 Name=%(ifc)s
 654 EOF
 655
 656 cat << EOF > /run/systemd/network/test.network
 657 [Match]
 658 Name=%(ifr)s
 659
 660 [Network]
 661 Address=192.168.5.1/24
 662 %(addr6)s
 663 DHCPServer=yes
 664
 665 [DHCPServer]
 666 PoolOffset=10
 667 PoolSize=50
 668 DNS=192.168.5.1
 669 %(dhopts)s
 670 EOF
 671
 672 # run networkd as in systemd-networkd.service
 673 exec $(systemctl cat systemd-networkd.service | sed -n '/^ExecStart=/ { s/^.*=//; p}')
 674 ''' % {'ifr': self.if_router, 'ifc': self.iface, 'addr6': ipv6 and 'Address=2600::1/64' or '',
 675        'dhopts': dhcpserver_opts or ''})
 676
 677             os.fchmod(fd, 0o755)
 678
 679         subprocess.check_call(['systemd-run', '--unit=networkd-test-router.service',
 680                                '-p', 'InaccessibleDirectories=-/etc/systemd/network',
 681                                '-p', 'InaccessibleDirectories=-/run/systemd/network',
 682                                '-p', 'InaccessibleDirectories=-/run/systemd/netif',
 683                                '--service-type=notify', script])
 684
 685         # wait until devices got created
 686         for _ in range(50):
 687             out = subprocess.check_output(['ip', 'a', 'show', 'dev', self.if_router])
 688             if b'state UP' in out and b'scope global' in out:
 689                 break
 690             time.sleep(0.1)
 691
 692     def shutdown_iface(self):
 693         '''Remove test interface and stop DHCP server'''
 694
 695         if self.if_router:
 696             subprocess.check_call(['systemctl', 'stop', 'networkd-test-router.service'])
 697             # ensure failed transient unit does not stay around
 698             subprocess.call(['systemctl', 'reset-failed', 'networkd-test-router.service'])
 699             subprocess.call(['ip', 'link', 'del', 'dev', self.if_router])
 700             self.if_router = None
 701
 702     def print_server_log(self):
 703         '''Print DHCP server log for debugging failures'''
 704
 705         self.show_journal('networkd-test-router.service')
 706
 707     @unittest.skip('networkd does not have DHCPv6 server support')
 708     def test_hotplug_dhcp_ip6(self):
 709         pass
 710
 711     @unittest.skip('networkd does not have DHCPv6 server support')
 712     def test_coldplug_dhcp_ip6(self):
 713         pass
 714
 715     def test_search_domains(self):
 716
 717         # we don't use this interface for this test
 718         self.if_router = None
 719
 720         self.write_network('test.netdev', '''\
 721 [NetDev]
 722 Name=dummy0
 723 Kind=dummy
 724 MACAddress=12:34:56:78:9a:bc''')
 725         self.write_network('test.network', '''\
 726 [Match]
 727 Name=dummy0
 728 [Network]
 729 Address=192.168.42.100
 730 DNS=192.168.42.1
 731 Domains= one two three four five six seven eight nine ten''')
 732
 733         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 734
 735         for timeout in range(50):
 736             with open(RESOLV_CONF) as f:
 737                 contents = f.read()
 738             if ' one' in contents:
 739                 break
 740             time.sleep(0.1)
 741         self.assertRegex(contents, 'search .*one two three four')
 742         self.assertNotIn('seven\n', contents)
 743         self.assertIn('# Too many search domains configured, remaining ones ignored.\n', contents)
 744
 745     def test_search_domains_too_long(self):
 746
 747         # we don't use this interface for this test
 748         self.if_router = None
 749
 750         name_prefix = 'a' * 60
 751
 752         self.write_network('test.netdev', '''\
 753 [NetDev]
 754 Name=dummy0
 755 Kind=dummy
 756 MACAddress=12:34:56:78:9a:bc''')
 757         self.write_network('test.network', '''\
 758 [Match]
 759 Name=dummy0
 760 [Network]
 761 Address=192.168.42.100
 762 DNS=192.168.42.1
 763 Domains={p}0 {p}1 {p}2 {p}3 {p}4'''.format(p=name_prefix))
 764
 765         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 766
 767         for timeout in range(50):
 768             with open(RESOLV_CONF) as f:
 769                 contents = f.read()
 770             if ' one' in contents:
 771                 break
 772             time.sleep(0.1)
 773         self.assertRegex(contents, 'search .*{p}0 {p}1 {p}2'.format(p=name_prefix))
 774         self.assertIn('# Total length of all search domains is too long, remaining ones ignored.', contents)
 775
 776     def test_dropin(self):
 777         # we don't use this interface for this test
 778         self.if_router = None
 779
 780         self.write_network('test.netdev', '''\
 781 [NetDev]
 782 Name=dummy0
 783 Kind=dummy
 784 MACAddress=12:34:56:78:9a:bc''')
 785         self.write_network('test.network', '''\
 786 [Match]
 787 Name=dummy0
 788 [Network]
 789 Address=192.168.42.100
 790 DNS=192.168.42.1''')
 791         self.write_network_dropin('test.network', 'dns', '''\
 792 [Network]
 793 DNS=127.0.0.1''')
 794
 795         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 796
 797         for timeout in range(50):
 798             with open(RESOLV_CONF) as f:
 799                 contents = f.read()
 800             if ' 127.0.0.1' in contents:
 801                 break
 802             time.sleep(0.1)
 803         self.assertIn('nameserver 192.168.42.1\n', contents)
 804         self.assertIn('nameserver 127.0.0.1\n', contents)
 805
 806     def test_dhcp_timezone(self):
 807         '''networkd sets time zone from DHCP'''
 808
 809         def get_tz():
 810             out = subprocess.check_output(['busctl', 'get-property', 'org.freedesktop.timedate1',
 811                                            '/org/freedesktop/timedate1', 'org.freedesktop.timedate1', 'Timezone'])
 812             assert out.startswith(b's "')
 813             out = out.strip()
 814             assert out.endswith(b'"')
 815             return out[3:-1].decode()
 816
 817         orig_timezone = get_tz()
 818         self.addCleanup(subprocess.call, ['timedatectl', 'set-timezone', orig_timezone])
 819
 820         self.create_iface(dhcpserver_opts='EmitTimezone=yes\nTimezone=Pacific/Honolulu')
 821         self.do_test(coldplug=None, extra_opts='IPv6AcceptRA=false\n[DHCP]\nUseTimezone=true', dhcp_mode='ipv4')
 822
 823         # should have applied the received timezone
 824         try:
 825             self.assertEqual(get_tz(), 'Pacific/Honolulu')
 826         except AssertionError:
 827             self.show_journal('systemd-networkd.service')
 828             self.show_journal('systemd-hostnamed.service')
 829             raise
 830
 831
 832 class MatchClientTest(unittest.TestCase, NetworkdTestingUtilities):
 833     """Test [Match] sections in .network files.
 834
 835     Be aware that matching the test host's interfaces will wipe their
 836     configuration, so as a precaution, all network files should have a
 837     restrictive [Match] section to only ever interfere with the
 838     temporary veth interfaces created here.
 839     """
 840
 841     def tearDown(self):
 842         """Stop networkd."""
 843         subprocess.call(['systemctl', 'stop', 'systemd-networkd'])
 844
 845     def test_basic_matching(self):
 846         """Verify the Name= line works throughout this class."""
 847         self.add_veth_pair('test_if1', 'fake_if2')
 848         self.write_network('test.network', "[Match]\nName=test_*\n[Network]")
 849         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 850         self.assert_link_states(test_if1='managed', fake_if2='unmanaged')
 851
 852     def test_inverted_matching(self):
 853         """Verify that a '!'-prefixed value inverts the match."""
 854         # Use a MAC address as the interfaces' common matching attribute
 855         # to avoid depending on udev, to support testing in containers.
 856         mac = '00:01:02:03:98:99'
 857         self.add_veth_pair('test_veth', 'test_peer',
 858                            ['addr', mac], ['addr', mac])
 859         self.write_network('no-veth.network', """\
 860 [Match]
 861 MACAddress=%s
 862 Name=!nonexistent *peer*
 863 [Network]""" % mac)
 864         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 865         self.assert_link_states(test_veth='managed', test_peer='unmanaged')
 866
 867
 868 class UnmanagedClientTest(unittest.TestCase, NetworkdTestingUtilities):
 869     """Test if networkd manages the correct interfaces."""
 870
 871     def setUp(self):
 872         """Write .network files to match the named veth devices."""
 873         # Define the veth+peer pairs to be created.
 874         # Their pairing doesn't actually matter, only their names do.
 875         self.veths = {
 876             'm1def': 'm0unm',
 877             'm1man': 'm1unm',
 878         }
 879
 880         # Define the contents of .network files to be read in order.
 881         self.configs = (
 882             "[Match]\nName=m1def\n",
 883             "[Match]\nName=m1unm\n[Link]\nUnmanaged=yes\n",
 884             "[Match]\nName=m1*\n[Link]\nUnmanaged=no\n",
 885         )
 886
 887         # Write out the .network files to be cleaned up automatically.
 888         for i, config in enumerate(self.configs):
 889             self.write_network("%02d-test.network" % i, config)
 890
 891     def tearDown(self):
 892         """Stop networkd."""
 893         subprocess.call(['systemctl', 'stop', 'systemd-networkd'])
 894
 895     def create_iface(self):
 896         """Create temporary veth pairs for interface matching."""
 897         for veth, peer in self.veths.items():
 898             self.add_veth_pair(veth, peer)
 899
 900     def test_unmanaged_setting(self):
 901         """Verify link states with Unmanaged= settings, hot-plug."""
 902         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 903         self.create_iface()
 904         self.assert_link_states(m1def='managed',
 905                                 m1man='managed',
 906                                 m1unm='unmanaged',
 907                                 m0unm='unmanaged')
 908
 909     def test_unmanaged_setting_coldplug(self):
 910         """Verify link states with Unmanaged= settings, cold-plug."""
 911         self.create_iface()
 912         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 913         self.assert_link_states(m1def='managed',
 914                                 m1man='managed',
 915                                 m1unm='unmanaged',
 916                                 m0unm='unmanaged')
 917
 918     def test_catchall_config(self):
 919         """Verify link states with a catch-all config, hot-plug."""
 920         # Don't actually catch ALL interfaces.  It messes up the host.
 921         self.write_network('all.network', "[Match]\nName=m[01]???\n")
 922         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 923         self.create_iface()
 924         self.assert_link_states(m1def='managed',
 925                                 m1man='managed',
 926                                 m1unm='unmanaged',
 927                                 m0unm='managed')
 928
 929     def test_catchall_config_coldplug(self):
 930         """Verify link states with a catch-all config, cold-plug."""
 931         # Don't actually catch ALL interfaces.  It messes up the host.
 932         self.write_network('all.network', "[Match]\nName=m[01]???\n")
 933         self.create_iface()
 934         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 935         self.assert_link_states(m1def='managed',
 936                                 m1man='managed',
 937                                 m1unm='unmanaged',
 938                                 m0unm='managed')
 939
 940
 941 if __name__ == '__main__':
 942     unittest.main(testRunner=unittest.TextTestRunner(stream=sys.stdout,
 943                                                      verbosity=2))