test/networkd-test.py

   1 #!/usr/bin/env python3
   2 # SPDX-License-Identifier: LGPL-2.1+
   3 #
   4 # networkd integration test
   5 # This uses temporary configuration in /run and temporary veth devices, and
   6 # does not write anything on disk or change any system configuration;
   7 # but it assumes (and checks at the beginning) that networkd is not currently
   8 # running.
   9 #
  10 # This can be run on a normal installation, in QEMU, nspawn (with
  11 # --private-network), LXD (with "--config raw.lxc=lxc.aa_profile=unconfined"),
  12 # or LXC system containers. You need at least the "ip" tool from the iproute
  13 # package; it is recommended to install dnsmasq too to get full test coverage.
  14 #
  15 # ATTENTION: This uses the *installed* networkd, not the one from the built
  16 # source tree.
  17 #
  18 # (C) 2015 Canonical Ltd.
  19 # Author: Martin Pitt <martin.pitt@ubuntu.com>
  20 #
  21 # systemd is free software; you can redistribute it and/or modify it
  22 # under the terms of the GNU Lesser General Public License as published by
  23 # the Free Software Foundation; either version 2.1 of the License, or
  24 # (at your option) any later version.
  25
  26 # systemd is distributed in the hope that it will be useful, but
  27 # WITHOUT ANY WARRANTY; without even the implied warranty of
  28 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  29 # Lesser General Public License for more details.
  30 #
  31 # You should have received a copy of the GNU Lesser General Public License
  32 # along with systemd; If not, see <http://www.gnu.org/licenses/>.
  33
  34 import errno
  35 import os
  36 import sys
  37 import time
  38 import unittest
  39 import tempfile
  40 import subprocess
  41 import shutil
  42 import socket
  43
  44 HAVE_DNSMASQ = shutil.which('dnsmasq') is not None
  45
  46 NETWORK_UNITDIR = '/run/systemd/network'
  47
  48 NETWORKD_WAIT_ONLINE = shutil.which('systemd-networkd-wait-online',
  49                                     path='/usr/lib/systemd:/lib/systemd')
  50
  51 RESOLV_CONF = '/run/systemd/resolve/resolv.conf'
  52
  53
  54 def setUpModule():
  55     """Initialize the environment, and perform sanity checks on it."""
  56     if NETWORKD_WAIT_ONLINE is None:
  57         raise OSError(errno.ENOENT, 'systemd-networkd-wait-online not found')
  58
  59     # Do not run any tests if the system is using networkd already.
  60     if subprocess.call(['systemctl', 'is-active', '--quiet',
  61                         'systemd-networkd.service']) == 0:
  62         raise unittest.SkipTest('networkd is already active')
  63
  64     # Avoid "Failed to open /dev/tty" errors in containers.
  65     os.environ['SYSTEMD_LOG_TARGET'] = 'journal'
  66
  67     # Ensure the unit directory exists so tests can dump files into it.
  68     os.makedirs(NETWORK_UNITDIR, exist_ok=True)
  69
  70
  71 class NetworkdTestingUtilities:
  72     """Provide a set of utility functions to facilitate networkd tests.
  73
  74     This class must be inherited along with unittest.TestCase to define
  75     some required methods.
  76     """
  77
  78     def add_veth_pair(self, veth, peer, veth_options=(), peer_options=()):
  79         """Add a veth interface pair, and queue them to be removed."""
  80         subprocess.check_call(['ip', 'link', 'add', 'name', veth] +
  81                               list(veth_options) +
  82                               ['type', 'veth', 'peer', 'name', peer] +
  83                               list(peer_options))
  84         self.addCleanup(subprocess.call, ['ip', 'link', 'del', 'dev', peer])
  85
  86     def write_network(self, unit_name, contents):
  87         """Write a network unit file, and queue it to be removed."""
  88         unit_path = os.path.join(NETWORK_UNITDIR, unit_name)
  89
  90         with open(unit_path, 'w') as unit:
  91             unit.write(contents)
  92         self.addCleanup(os.remove, unit_path)
  93
  94     def write_network_dropin(self, unit_name, dropin_name, contents):
  95         """Write a network unit drop-in, and queue it to be removed."""
  96         dropin_dir = os.path.join(NETWORK_UNITDIR, "%s.d" % unit_name)
  97         dropin_path = os.path.join(dropin_dir, "%s.conf" % dropin_name)
  98
  99         os.makedirs(dropin_dir, exist_ok=True)
 100         self.addCleanup(os.rmdir, dropin_dir)
 101         with open(dropin_path, 'w') as dropin:
 102             dropin.write(contents)
 103         self.addCleanup(os.remove, dropin_path)
 104
 105     def read_attr(self, link, attribute):
 106         """Read a link attributed from the sysfs."""
 107         # Note we we don't want to check if interface `link' is managed, we
 108         # want to evaluate link variable and pass the value of the link to
 109         # assert_link_states e.g. eth0=managed.
 110         self.assert_link_states(**{link:'managed'})
 111         with open(os.path.join('/sys/class/net', link, attribute)) as f:
 112             return f.readline().strip()
 113
 114     def assert_link_states(self, **kwargs):
 115         """Match networkctl link states to the given ones.
 116
 117         Each keyword argument should be the name of a network interface
 118         with its expected value of the "SETUP" column in output from
 119         networkctl.  The interfaces have five seconds to come online
 120         before the check is performed.  Every specified interface must
 121         be present in the output, and any other interfaces found in the
 122         output are ignored.
 123
 124         A special interface state "managed" is supported, which matches
 125         any value in the "SETUP" column other than "unmanaged".
 126         """
 127         if not kwargs:
 128             return
 129         interfaces = set(kwargs)
 130
 131         # Wait for the requested interfaces, but don't fail for them.
 132         subprocess.call([NETWORKD_WAIT_ONLINE, '--timeout=5'] +
 133                         ['--interface=%s' % iface for iface in kwargs])
 134
 135         # Validate each link state found in the networkctl output.
 136         out = subprocess.check_output(['networkctl', '--no-legend']).rstrip()
 137         for line in out.decode('utf-8').split('\n'):
 138             fields = line.split()
 139             if len(fields) >= 5 and fields[1] in kwargs:
 140                 iface = fields[1]
 141                 expected = kwargs[iface]
 142                 actual = fields[-1]
 143                 if (actual != expected and
 144                         not (expected == 'managed' and actual != 'unmanaged')):
 145                     self.fail("Link %s expects state %s, found %s" %
 146                               (iface, expected, actual))
 147                 interfaces.remove(iface)
 148
 149         # Ensure that all requested interfaces have been covered.
 150         if interfaces:
 151             self.fail("Missing links in status output: %s" % interfaces)
 152
 153
 154 class BridgeTest(NetworkdTestingUtilities, unittest.TestCase):
 155     """Provide common methods for testing networkd against servers."""
 156
 157     def setUp(self):
 158         self.write_network('port1.netdev', '''\
 159 [NetDev]
 160 Name=port1
 161 Kind=dummy
 162 MACAddress=12:34:56:78:9a:bc''')
 163         self.write_network('port2.netdev', '''\
 164 [NetDev]
 165 Name=port2
 166 Kind=dummy
 167 MACAddress=12:34:56:78:9a:bd''')
 168         self.write_network('mybridge.netdev', '''\
 169 [NetDev]
 170 Name=mybridge
 171 Kind=bridge''')
 172         self.write_network('port1.network', '''\
 173 [Match]
 174 Name=port1
 175 [Network]
 176 Bridge=mybridge''')
 177         self.write_network('port2.network', '''\
 178 [Match]
 179 Name=port2
 180 [Network]
 181 Bridge=mybridge''')
 182         self.write_network('mybridge.network', '''\
 183 [Match]
 184 Name=mybridge
 185 [Network]
 186 DNS=192.168.250.1
 187 Address=192.168.250.33/24
 188 Gateway=192.168.250.1''')
 189         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 190
 191     def tearDown(self):
 192         subprocess.check_call(['systemctl', 'stop', 'systemd-networkd'])
 193         subprocess.check_call(['ip', 'link', 'del', 'mybridge'])
 194         subprocess.check_call(['ip', 'link', 'del', 'port1'])
 195         subprocess.check_call(['ip', 'link', 'del', 'port2'])
 196
 197     def test_bridge_init(self):
 198         self.assert_link_states(
 199             port1='managed',
 200             port2='managed',
 201             mybridge='managed')
 202
 203     def test_bridge_port_priority(self):
 204         self.assertEqual(self.read_attr('port1', 'brport/priority'), '32')
 205         self.write_network_dropin('port1.network', 'priority', '''\
 206 [Bridge]
 207 Priority=28
 208 ''')
 209         subprocess.check_call(['systemctl', 'restart', 'systemd-networkd'])
 210         self.assertEqual(self.read_attr('port1', 'brport/priority'), '28')
 211
 212     def test_bridge_port_priority_set_zero(self):
 213         """It should be possible to set the bridge port priority to 0"""
 214         self.assertEqual(self.read_attr('port2', 'brport/priority'), '32')
 215         self.write_network_dropin('port2.network', 'priority', '''\
 216 [Bridge]
 217 Priority=0
 218 ''')
 219         subprocess.check_call(['systemctl', 'restart', 'systemd-networkd'])
 220         self.assertEqual(self.read_attr('port2', 'brport/priority'), '0')
 221
 222 class ClientTestBase(NetworkdTestingUtilities):
 223     """Provide common methods for testing networkd against servers."""
 224
 225     @classmethod
 226     def setUpClass(klass):
 227         klass.orig_log_level = subprocess.check_output(
 228             ['systemctl', 'show', '--value', '--property', 'LogLevel'],
 229             universal_newlines=True).strip()
 230         subprocess.check_call(['systemd-analyze', 'set-log-level', 'debug'])
 231
 232     @classmethod
 233     def tearDownClass(klass):
 234         subprocess.check_call(['systemd-analyze', 'set-log-level', klass.orig_log_level])
 235
 236     def setUp(self):
 237         self.iface = 'test_eth42'
 238         self.if_router = 'router_eth42'
 239         self.workdir_obj = tempfile.TemporaryDirectory()
 240         self.workdir = self.workdir_obj.name
 241         self.config = 'test_eth42.network'
 242
 243         # get current journal cursor
 244         subprocess.check_output(['journalctl', '--sync'])
 245         out = subprocess.check_output(['journalctl', '-b', '--quiet',
 246                                        '--no-pager', '-n0', '--show-cursor'],
 247                                       universal_newlines=True)
 248         self.assertTrue(out.startswith('-- cursor:'))
 249         self.journal_cursor = out.split()[-1]
 250
 251     def tearDown(self):
 252         self.shutdown_iface()
 253         subprocess.call(['systemctl', 'stop', 'systemd-networkd'])
 254         subprocess.call(['ip', 'link', 'del', 'dummy0'],
 255                         stderr=subprocess.DEVNULL)
 256
 257     def show_journal(self, unit):
 258         '''Show journal of given unit since start of the test'''
 259
 260         print('---- %s ----' % unit)
 261         subprocess.check_output(['journalctl', '--sync'])
 262         sys.stdout.flush()
 263         subprocess.call(['journalctl', '-b', '--no-pager', '--quiet',
 264                          '--cursor', self.journal_cursor, '-u', unit])
 265
 266     def create_iface(self, ipv6=False):
 267         '''Create test interface with DHCP server behind it'''
 268
 269         raise NotImplementedError('must be implemented by a subclass')
 270
 271     def shutdown_iface(self):
 272         '''Remove test interface and stop DHCP server'''
 273
 274         raise NotImplementedError('must be implemented by a subclass')
 275
 276     def print_server_log(self):
 277         '''Print DHCP server log for debugging failures'''
 278
 279         raise NotImplementedError('must be implemented by a subclass')
 280
 281     def do_test(self, coldplug=True, ipv6=False, extra_opts='',
 282                 online_timeout=10, dhcp_mode='yes'):
 283         try:
 284             subprocess.check_call(['systemctl', 'start', 'systemd-resolved'])
 285         except subprocess.CalledProcessError:
 286             self.show_journal('systemd-resolved.service')
 287             raise
 288         self.write_network(self.config, '''\
 289 [Match]
 290 Name=%s
 291 [Network]
 292 DHCP=%s
 293 %s''' % (self.iface, dhcp_mode, extra_opts))
 294
 295         if coldplug:
 296             # create interface first, then start networkd
 297             self.create_iface(ipv6=ipv6)
 298             subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 299         elif coldplug is not None:
 300             # start networkd first, then create interface
 301             subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 302             self.create_iface(ipv6=ipv6)
 303         else:
 304             # "None" means test sets up interface by itself
 305             subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 306
 307         try:
 308             subprocess.check_call([NETWORKD_WAIT_ONLINE, '--interface',
 309                                    self.iface, '--timeout=%i' % online_timeout])
 310
 311             if ipv6:
 312                 # check iface state and IP 6 address; FIXME: we need to wait a bit
 313                 # longer, as the iface is "configured" already with IPv4 *or*
 314                 # IPv6, but we want to wait for both
 315                 for _ in range(10):
 316                     out = subprocess.check_output(['ip', 'a', 'show', 'dev', self.iface])
 317                     if b'state UP' in out and b'inet6 2600' in out and b'inet 192.168' in out:
 318                         break
 319                     time.sleep(1)
 320                 else:
 321                     self.fail('timed out waiting for IPv6 configuration')
 322
 323                 self.assertRegex(out, b'inet6 2600::.* scope global .*dynamic')
 324                 self.assertRegex(out, b'inet6 fe80::.* scope link')
 325             else:
 326                 # should have link-local address on IPv6 only
 327                 out = subprocess.check_output(['ip', '-6', 'a', 'show', 'dev', self.iface])
 328                 self.assertRegex(out, br'inet6 fe80::.* scope link')
 329                 self.assertNotIn(b'scope global', out)
 330
 331             # should have IPv4 address
 332             out = subprocess.check_output(['ip', '-4', 'a', 'show', 'dev', self.iface])
 333             self.assertIn(b'state UP', out)
 334             self.assertRegex(out, br'inet 192.168.5.\d+/.* scope global dynamic')
 335
 336             # check networkctl state
 337             out = subprocess.check_output(['networkctl'])
 338             self.assertRegex(out, (r'%s\s+ether\s+[a-z-]+\s+unmanaged' % self.if_router).encode())
 339             self.assertRegex(out, (r'%s\s+ether\s+routable\s+configured' % self.iface).encode())
 340
 341             out = subprocess.check_output(['networkctl', 'status', self.iface])
 342             self.assertRegex(out, br'Type:\s+ether')
 343             self.assertRegex(out, br'State:\s+routable.*configured')
 344             self.assertRegex(out, br'Address:\s+192.168.5.\d+')
 345             if ipv6:
 346                 self.assertRegex(out, br'2600::')
 347             else:
 348                 self.assertNotIn(br'2600::', out)
 349             self.assertRegex(out, br'fe80::')
 350             self.assertRegex(out, br'Gateway:\s+192.168.5.1')
 351             self.assertRegex(out, br'DNS:\s+192.168.5.1')
 352         except (AssertionError, subprocess.CalledProcessError):
 353             # show networkd status, journal, and DHCP server log on failure
 354             with open(os.path.join(NETWORK_UNITDIR, self.config)) as f:
 355                 print('\n---- %s ----\n%s' % (self.config, f.read()))
 356             print('---- interface status ----')
 357             sys.stdout.flush()
 358             subprocess.call(['ip', 'a', 'show', 'dev', self.iface])
 359             print('---- networkctl status %s ----' % self.iface)
 360             sys.stdout.flush()
 361             subprocess.call(['networkctl', 'status', self.iface])
 362             self.show_journal('systemd-networkd.service')
 363             self.print_server_log()
 364             raise
 365
 366         for timeout in range(50):
 367             with open(RESOLV_CONF) as f:
 368                 contents = f.read()
 369             if 'nameserver 192.168.5.1\n' in contents:
 370                 break
 371             time.sleep(0.1)
 372         else:
 373             self.fail('nameserver 192.168.5.1 not found in ' + RESOLV_CONF)
 374
 375         if coldplug is False:
 376             # check post-down.d hook
 377             self.shutdown_iface()
 378
 379     def test_coldplug_dhcp_yes_ip4(self):
 380         # we have a 12s timeout on RA, so we need to wait longer
 381         self.do_test(coldplug=True, ipv6=False, online_timeout=15)
 382
 383     def test_coldplug_dhcp_yes_ip4_no_ra(self):
 384         # with disabling RA explicitly things should be fast
 385         self.do_test(coldplug=True, ipv6=False,
 386                      extra_opts='IPv6AcceptRA=False')
 387
 388     def test_coldplug_dhcp_ip4_only(self):
 389         # we have a 12s timeout on RA, so we need to wait longer
 390         self.do_test(coldplug=True, ipv6=False, dhcp_mode='ipv4',
 391                      online_timeout=15)
 392
 393     def test_coldplug_dhcp_ip4_only_no_ra(self):
 394         # with disabling RA explicitly things should be fast
 395         self.do_test(coldplug=True, ipv6=False, dhcp_mode='ipv4',
 396                      extra_opts='IPv6AcceptRA=False')
 397
 398     def test_coldplug_dhcp_ip6(self):
 399         self.do_test(coldplug=True, ipv6=True)
 400
 401     def test_hotplug_dhcp_ip4(self):
 402         # With IPv4 only we have a 12s timeout on RA, so we need to wait longer
 403         self.do_test(coldplug=False, ipv6=False, online_timeout=15)
 404
 405     def test_hotplug_dhcp_ip6(self):
 406         self.do_test(coldplug=False, ipv6=True)
 407
 408     def test_route_only_dns(self):
 409         self.write_network('myvpn.netdev', '''\
 410 [NetDev]
 411 Name=dummy0
 412 Kind=dummy
 413 MACAddress=12:34:56:78:9a:bc''')
 414         self.write_network('myvpn.network', '''\
 415 [Match]
 416 Name=dummy0
 417 [Network]
 418 Address=192.168.42.100
 419 DNS=192.168.42.1
 420 Domains= ~company''')
 421
 422         self.do_test(coldplug=True, ipv6=False,
 423                      extra_opts='IPv6AcceptRouterAdvertisements=False')
 424
 425         with open(RESOLV_CONF) as f:
 426             contents = f.read()
 427             # ~company is not a search domain, only a routing domain
 428             self.assertNotRegex(contents, 'search.*company')
 429             # our global server should appear
 430             self.assertIn('nameserver 192.168.5.1\n', contents)
 431             # should not have domain-restricted server as global server
 432             self.assertNotIn('nameserver 192.168.42.1\n', contents)
 433
 434     def test_route_only_dns_all_domains(self):
 435         self.write_network('myvpn.netdev', '''[NetDev]
 436 Name=dummy0
 437 Kind=dummy
 438 MACAddress=12:34:56:78:9a:bc''')
 439         self.write_network('myvpn.network', '''[Match]
 440 Name=dummy0
 441 [Network]
 442 Address=192.168.42.100
 443 DNS=192.168.42.1
 444 Domains= ~company ~.''')
 445
 446         self.do_test(coldplug=True, ipv6=False,
 447                      extra_opts='IPv6AcceptRouterAdvertisements=False')
 448
 449         with open(RESOLV_CONF) as f:
 450             contents = f.read()
 451
 452         # ~company is not a search domain, only a routing domain
 453         self.assertNotRegex(contents, 'search.*company')
 454
 455         # our global server should appear
 456         self.assertIn('nameserver 192.168.5.1\n', contents)
 457         # should have company server as global server due to ~.
 458         self.assertIn('nameserver 192.168.42.1\n', contents)
 459
 460
 461 @unittest.skipUnless(HAVE_DNSMASQ, 'dnsmasq not installed')
 462 class DnsmasqClientTest(ClientTestBase, unittest.TestCase):
 463     '''Test networkd client against dnsmasq'''
 464
 465     def setUp(self):
 466         super().setUp()
 467         self.dnsmasq = None
 468         self.iface_mac = 'de:ad:be:ef:47:11'
 469
 470     def create_iface(self, ipv6=False, dnsmasq_opts=None):
 471         '''Create test interface with DHCP server behind it'''
 472
 473         # add veth pair
 474         subprocess.check_call(['ip', 'link', 'add', 'name', self.iface,
 475                                'address', self.iface_mac,
 476                                'type', 'veth', 'peer', 'name', self.if_router])
 477
 478         # give our router an IP
 479         subprocess.check_call(['ip', 'a', 'flush', 'dev', self.if_router])
 480         subprocess.check_call(['ip', 'a', 'add', '192.168.5.1/24', 'dev', self.if_router])
 481         if ipv6:
 482             subprocess.check_call(['ip', 'a', 'add', '2600::1/64', 'dev', self.if_router])
 483         subprocess.check_call(['ip', 'link', 'set', self.if_router, 'up'])
 484
 485         # add DHCP server
 486         self.dnsmasq_log = os.path.join(self.workdir, 'dnsmasq.log')
 487         lease_file = os.path.join(self.workdir, 'dnsmasq.leases')
 488         if ipv6:
 489             extra_opts = ['--enable-ra', '--dhcp-range=2600::10,2600::20']
 490         else:
 491             extra_opts = []
 492         if dnsmasq_opts:
 493             extra_opts += dnsmasq_opts
 494         self.dnsmasq = subprocess.Popen(
 495             ['dnsmasq', '--keep-in-foreground', '--log-queries',
 496              '--log-facility=' + self.dnsmasq_log, '--conf-file=/dev/null',
 497              '--dhcp-leasefile=' + lease_file, '--bind-interfaces',
 498              '--interface=' + self.if_router, '--except-interface=lo',
 499              '--dhcp-range=192.168.5.10,192.168.5.200'] + extra_opts)
 500
 501     def shutdown_iface(self):
 502         '''Remove test interface and stop DHCP server'''
 503
 504         if self.if_router:
 505             subprocess.check_call(['ip', 'link', 'del', 'dev', self.if_router])
 506             self.if_router = None
 507         if self.dnsmasq:
 508             self.dnsmasq.kill()
 509             self.dnsmasq.wait()
 510             self.dnsmasq = None
 511
 512     def print_server_log(self):
 513         '''Print DHCP server log for debugging failures'''
 514
 515         with open(self.dnsmasq_log) as f:
 516             sys.stdout.write('\n\n---- dnsmasq log ----\n%s\n------\n\n' % f.read())
 517
 518     def test_resolved_domain_restricted_dns(self):
 519         '''resolved: domain-restricted DNS servers'''
 520
 521         # create interface for generic connections; this will map all DNS names
 522         # to 192.168.42.1
 523         self.create_iface(dnsmasq_opts=['--address=/#/192.168.42.1'])
 524         self.write_network('general.network', '''\
 525 [Match]
 526 Name=%s
 527 [Network]
 528 DHCP=ipv4
 529 IPv6AcceptRA=False''' % self.iface)
 530
 531         # create second device/dnsmasq for a .company/.lab VPN interface
 532         # static IPs for simplicity
 533         self.add_veth_pair('testvpnclient', 'testvpnrouter')
 534         subprocess.check_call(['ip', 'a', 'flush', 'dev', 'testvpnrouter'])
 535         subprocess.check_call(['ip', 'a', 'add', '10.241.3.1/24', 'dev', 'testvpnrouter'])
 536         subprocess.check_call(['ip', 'link', 'set', 'testvpnrouter', 'up'])
 537
 538         vpn_dnsmasq_log = os.path.join(self.workdir, 'dnsmasq-vpn.log')
 539         vpn_dnsmasq = subprocess.Popen(
 540             ['dnsmasq', '--keep-in-foreground', '--log-queries',
 541              '--log-facility=' + vpn_dnsmasq_log, '--conf-file=/dev/null',
 542              '--dhcp-leasefile=/dev/null', '--bind-interfaces',
 543              '--interface=testvpnrouter', '--except-interface=lo',
 544              '--address=/math.lab/10.241.3.3', '--address=/cantina.company/10.241.4.4'])
 545         self.addCleanup(vpn_dnsmasq.wait)
 546         self.addCleanup(vpn_dnsmasq.kill)
 547
 548         self.write_network('vpn.network', '''\
 549 [Match]
 550 Name=testvpnclient
 551 [Network]
 552 IPv6AcceptRA=False
 553 Address=10.241.3.2/24
 554 DNS=10.241.3.1
 555 Domains= ~company ~lab''')
 556
 557         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 558         subprocess.check_call([NETWORKD_WAIT_ONLINE, '--interface', self.iface,
 559                                '--interface=testvpnclient', '--timeout=20'])
 560
 561         # ensure we start fresh with every test
 562         subprocess.check_call(['systemctl', 'restart', 'systemd-resolved'])
 563
 564         # test vpnclient specific domains; these should *not* be answered by
 565         # the general DNS
 566         out = subprocess.check_output(['systemd-resolve', 'math.lab'])
 567         self.assertIn(b'math.lab: 10.241.3.3', out)
 568         out = subprocess.check_output(['systemd-resolve', 'kettle.cantina.company'])
 569         self.assertIn(b'kettle.cantina.company: 10.241.4.4', out)
 570
 571         # test general domains
 572         out = subprocess.check_output(['systemd-resolve', 'megasearch.net'])
 573         self.assertIn(b'megasearch.net: 192.168.42.1', out)
 574
 575         with open(self.dnsmasq_log) as f:
 576             general_log = f.read()
 577         with open(vpn_dnsmasq_log) as f:
 578             vpn_log = f.read()
 579
 580         # VPN domains should only be sent to VPN DNS
 581         self.assertRegex(vpn_log, 'query.*math.lab')
 582         self.assertRegex(vpn_log, 'query.*cantina.company')
 583         self.assertNotIn('.lab', general_log)
 584         self.assertNotIn('.company', general_log)
 585
 586         # general domains should not be sent to the VPN DNS
 587         self.assertRegex(general_log, 'query.*megasearch.net')
 588         self.assertNotIn('megasearch.net', vpn_log)
 589
 590     def test_resolved_etc_hosts(self):
 591         '''resolved queries to /etc/hosts'''
 592
 593         # FIXME: -t MX query fails with enabled DNSSEC (even when using
 594         # the known negative trust anchor .internal instead of .example)
 595         conf = '/run/systemd/resolved.conf.d/test-disable-dnssec.conf'
 596         os.makedirs(os.path.dirname(conf), exist_ok=True)
 597         with open(conf, 'w') as f:
 598             f.write('[Resolve]\nDNSSEC=no')
 599         self.addCleanup(os.remove, conf)
 600
 601         # create /etc/hosts bind mount which resolves my.example for IPv4
 602         hosts = os.path.join(self.workdir, 'hosts')
 603         with open(hosts, 'w') as f:
 604             f.write('172.16.99.99  my.example\n')
 605         subprocess.check_call(['mount', '--bind', hosts, '/etc/hosts'])
 606         self.addCleanup(subprocess.call, ['umount', '/etc/hosts'])
 607         subprocess.check_call(['systemctl', 'stop', 'systemd-resolved.service'])
 608
 609         # note: different IPv4 address here, so that it's easy to tell apart
 610         # what resolved the query
 611         self.create_iface(dnsmasq_opts=['--host-record=my.example,172.16.99.1,2600::99:99',
 612                                         '--host-record=other.example,172.16.0.42,2600::42',
 613                                         '--mx-host=example,mail.example'],
 614                           ipv6=True)
 615         self.do_test(coldplug=None, ipv6=True)
 616
 617         try:
 618             # family specific queries
 619             out = subprocess.check_output(['systemd-resolve', '-4', 'my.example'])
 620             self.assertIn(b'my.example: 172.16.99.99', out)
 621             # we don't expect an IPv6 answer; if /etc/hosts has any IP address,
 622             # it's considered a sufficient source
 623             self.assertNotEqual(subprocess.call(['systemd-resolve', '-6', 'my.example']), 0)
 624             # "any family" query; IPv4 should come from /etc/hosts
 625             out = subprocess.check_output(['systemd-resolve', 'my.example'])
 626             self.assertIn(b'my.example: 172.16.99.99', out)
 627             # IP → name lookup; again, takes the /etc/hosts one
 628             out = subprocess.check_output(['systemd-resolve', '172.16.99.99'])
 629             self.assertIn(b'172.16.99.99: my.example', out)
 630
 631             # non-address RRs should fall back to DNS
 632             out = subprocess.check_output(['systemd-resolve', '--type=MX', 'example'])
 633             self.assertIn(b'example IN MX 1 mail.example', out)
 634
 635             # other domains query DNS
 636             out = subprocess.check_output(['systemd-resolve', 'other.example'])
 637             self.assertIn(b'172.16.0.42', out)
 638             out = subprocess.check_output(['systemd-resolve', '172.16.0.42'])
 639             self.assertIn(b'172.16.0.42: other.example', out)
 640         except (AssertionError, subprocess.CalledProcessError):
 641             self.show_journal('systemd-resolved.service')
 642             self.print_server_log()
 643             raise
 644
 645     def test_transient_hostname(self):
 646         '''networkd sets transient hostname from DHCP'''
 647
 648         orig_hostname = socket.gethostname()
 649         self.addCleanup(socket.sethostname, orig_hostname)
 650         # temporarily move /etc/hostname away; restart hostnamed to pick it up
 651         if os.path.exists('/etc/hostname'):
 652             subprocess.check_call(['mount', '--bind', '/dev/null', '/etc/hostname'])
 653             self.addCleanup(subprocess.call, ['umount', '/etc/hostname'])
 654         subprocess.check_call(['systemctl', 'stop', 'systemd-hostnamed.service'])
 655
 656         self.create_iface(dnsmasq_opts=['--dhcp-host=%s,192.168.5.210,testgreen' % self.iface_mac])
 657         self.do_test(coldplug=None, extra_opts='IPv6AcceptRA=False', dhcp_mode='ipv4')
 658
 659         try:
 660             # should have received the fixed IP above
 661             out = subprocess.check_output(['ip', '-4', 'a', 'show', 'dev', self.iface])
 662             self.assertRegex(out, b'inet 192.168.5.210/24 .* scope global dynamic')
 663             # should have set transient hostname in hostnamed; this is
 664             # sometimes a bit lagging (issue #4753), so retry a few times
 665             for retry in range(1, 6):
 666                 out = subprocess.check_output(['hostnamectl'])
 667                 if b'testgreen' in out:
 668                     break
 669                 time.sleep(5)
 670                 sys.stdout.write('[retry %i] ' % retry)
 671                 sys.stdout.flush()
 672             else:
 673                 self.fail('Transient hostname not found in hostnamectl:\n%s' % out.decode())
 674             # and also applied to the system
 675             self.assertEqual(socket.gethostname(), 'testgreen')
 676         except AssertionError:
 677             self.show_journal('systemd-networkd.service')
 678             self.show_journal('systemd-hostnamed.service')
 679             self.print_server_log()
 680             raise
 681
 682     def test_transient_hostname_with_static(self):
 683         '''transient hostname is not applied if static hostname exists'''
 684
 685         orig_hostname = socket.gethostname()
 686         self.addCleanup(socket.sethostname, orig_hostname)
 687         if not os.path.exists('/etc/hostname'):
 688             self.writeConfig('/etc/hostname', orig_hostname)
 689         subprocess.check_call(['systemctl', 'stop', 'systemd-hostnamed.service'])
 690
 691         self.create_iface(dnsmasq_opts=['--dhcp-host=%s,192.168.5.210,testgreen' % self.iface_mac])
 692         self.do_test(coldplug=None, extra_opts='IPv6AcceptRA=False', dhcp_mode='ipv4')
 693
 694         try:
 695             # should have received the fixed IP above
 696             out = subprocess.check_output(['ip', '-4', 'a', 'show', 'dev', self.iface])
 697             self.assertRegex(out, b'inet 192.168.5.210/24 .* scope global dynamic')
 698             # static hostname wins over transient one, thus *not* applied
 699             self.assertEqual(socket.gethostname(), orig_hostname)
 700         except AssertionError:
 701             self.show_journal('systemd-networkd.service')
 702             self.show_journal('systemd-hostnamed.service')
 703             self.print_server_log()
 704             raise
 705
 706
 707 class NetworkdClientTest(ClientTestBase, unittest.TestCase):
 708     '''Test networkd client against networkd server'''
 709
 710     def setUp(self):
 711         super().setUp()
 712         self.dnsmasq = None
 713
 714     def create_iface(self, ipv6=False, dhcpserver_opts=None):
 715         '''Create test interface with DHCP server behind it'''
 716
 717         # run "router-side" networkd in own mount namespace to shield it from
 718         # "client-side" configuration and networkd
 719         (fd, script) = tempfile.mkstemp(prefix='networkd-router.sh')
 720         self.addCleanup(os.remove, script)
 721         with os.fdopen(fd, 'w+') as f:
 722             f.write('''\
 723 #!/bin/sh -eu
 724 mkdir -p /run/systemd/network
 725 mkdir -p /run/systemd/netif
 726 mount -t tmpfs none /run/systemd/network
 727 mount -t tmpfs none /run/systemd/netif
 728 [ ! -e /run/dbus ] || mount -t tmpfs none /run/dbus
 729 # create router/client veth pair
 730 cat << EOF > /run/systemd/network/test.netdev
 731 [NetDev]
 732 Name=%(ifr)s
 733 Kind=veth
 734
 735 [Peer]
 736 Name=%(ifc)s
 737 EOF
 738
 739 cat << EOF > /run/systemd/network/test.network
 740 [Match]
 741 Name=%(ifr)s
 742
 743 [Network]
 744 Address=192.168.5.1/24
 745 %(addr6)s
 746 DHCPServer=yes
 747
 748 [DHCPServer]
 749 PoolOffset=10
 750 PoolSize=50
 751 DNS=192.168.5.1
 752 %(dhopts)s
 753 EOF
 754
 755 # run networkd as in systemd-networkd.service
 756 exec $(systemctl cat systemd-networkd.service | sed -n '/^ExecStart=/ { s/^.*=//; s/^[@+-]//; s/^!*//; p}')
 757 ''' % {'ifr': self.if_router, 'ifc': self.iface, 'addr6': ipv6 and 'Address=2600::1/64' or '',
 758        'dhopts': dhcpserver_opts or ''})
 759
 760             os.fchmod(fd, 0o755)
 761
 762         subprocess.check_call(['systemd-run', '--unit=networkd-test-router.service',
 763                                '-p', 'InaccessibleDirectories=-/etc/systemd/network',
 764                                '-p', 'InaccessibleDirectories=-/run/systemd/network',
 765                                '-p', 'InaccessibleDirectories=-/run/systemd/netif',
 766                                '--service-type=notify', script])
 767
 768         # wait until devices got created
 769         for _ in range(50):
 770             out = subprocess.check_output(['ip', 'a', 'show', 'dev', self.if_router])
 771             if b'state UP' in out and b'scope global' in out:
 772                 break
 773             time.sleep(0.1)
 774
 775     def shutdown_iface(self):
 776         '''Remove test interface and stop DHCP server'''
 777
 778         if self.if_router:
 779             subprocess.check_call(['systemctl', 'stop', 'networkd-test-router.service'])
 780             # ensure failed transient unit does not stay around
 781             subprocess.call(['systemctl', 'reset-failed', 'networkd-test-router.service'])
 782             subprocess.call(['ip', 'link', 'del', 'dev', self.if_router])
 783             self.if_router = None
 784
 785     def print_server_log(self):
 786         '''Print DHCP server log for debugging failures'''
 787
 788         self.show_journal('networkd-test-router.service')
 789
 790     @unittest.skip('networkd does not have DHCPv6 server support')
 791     def test_hotplug_dhcp_ip6(self):
 792         pass
 793
 794     @unittest.skip('networkd does not have DHCPv6 server support')
 795     def test_coldplug_dhcp_ip6(self):
 796         pass
 797
 798     def test_search_domains(self):
 799
 800         # we don't use this interface for this test
 801         self.if_router = None
 802
 803         self.write_network('test.netdev', '''\
 804 [NetDev]
 805 Name=dummy0
 806 Kind=dummy
 807 MACAddress=12:34:56:78:9a:bc''')
 808         self.write_network('test.network', '''\
 809 [Match]
 810 Name=dummy0
 811 [Network]
 812 Address=192.168.42.100
 813 DNS=192.168.42.1
 814 Domains= one two three four five six seven eight nine ten''')
 815
 816         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 817
 818         for timeout in range(50):
 819             with open(RESOLV_CONF) as f:
 820                 contents = f.read()
 821             if ' one' in contents:
 822                 break
 823             time.sleep(0.1)
 824         self.assertRegex(contents, 'search .*one two three four')
 825         self.assertNotIn('seven\n', contents)
 826         self.assertIn('# Too many search domains configured, remaining ones ignored.\n', contents)
 827
 828     def test_search_domains_too_long(self):
 829
 830         # we don't use this interface for this test
 831         self.if_router = None
 832
 833         name_prefix = 'a' * 60
 834
 835         self.write_network('test.netdev', '''\
 836 [NetDev]
 837 Name=dummy0
 838 Kind=dummy
 839 MACAddress=12:34:56:78:9a:bc''')
 840         self.write_network('test.network', '''\
 841 [Match]
 842 Name=dummy0
 843 [Network]
 844 Address=192.168.42.100
 845 DNS=192.168.42.1
 846 Domains={p}0 {p}1 {p}2 {p}3 {p}4'''.format(p=name_prefix))
 847
 848         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 849
 850         for timeout in range(50):
 851             with open(RESOLV_CONF) as f:
 852                 contents = f.read()
 853             if ' one' in contents:
 854                 break
 855             time.sleep(0.1)
 856         self.assertRegex(contents, 'search .*{p}0 {p}1 {p}2'.format(p=name_prefix))
 857         self.assertIn('# Total length of all search domains is too long, remaining ones ignored.', contents)
 858
 859     def test_dropin(self):
 860         # we don't use this interface for this test
 861         self.if_router = None
 862
 863         self.write_network('test.netdev', '''\
 864 [NetDev]
 865 Name=dummy0
 866 Kind=dummy
 867 MACAddress=12:34:56:78:9a:bc''')
 868         self.write_network('test.network', '''\
 869 [Match]
 870 Name=dummy0
 871 [Network]
 872 Address=192.168.42.100
 873 DNS=192.168.42.1''')
 874         self.write_network_dropin('test.network', 'dns', '''\
 875 [Network]
 876 DNS=127.0.0.1''')
 877
 878         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 879
 880         for timeout in range(50):
 881             with open(RESOLV_CONF) as f:
 882                 contents = f.read()
 883             if ' 127.0.0.1' in contents:
 884                 break
 885             time.sleep(0.1)
 886         self.assertIn('nameserver 192.168.42.1\n', contents)
 887         self.assertIn('nameserver 127.0.0.1\n', contents)
 888
 889     def test_dhcp_timezone(self):
 890         '''networkd sets time zone from DHCP'''
 891
 892         def get_tz():
 893             out = subprocess.check_output(['busctl', 'get-property', 'org.freedesktop.timedate1',
 894                                            '/org/freedesktop/timedate1', 'org.freedesktop.timedate1', 'Timezone'])
 895             assert out.startswith(b's "')
 896             out = out.strip()
 897             assert out.endswith(b'"')
 898             return out[3:-1].decode()
 899
 900         orig_timezone = get_tz()
 901         self.addCleanup(subprocess.call, ['timedatectl', 'set-timezone', orig_timezone])
 902
 903         self.create_iface(dhcpserver_opts='EmitTimezone=yes\nTimezone=Pacific/Honolulu')
 904         self.do_test(coldplug=None, extra_opts='IPv6AcceptRA=false\n[DHCP]\nUseTimezone=true', dhcp_mode='ipv4')
 905
 906         # should have applied the received timezone
 907         try:
 908             self.assertEqual(get_tz(), 'Pacific/Honolulu')
 909         except AssertionError:
 910             self.show_journal('systemd-networkd.service')
 911             self.show_journal('systemd-hostnamed.service')
 912             raise
 913
 914
 915 class MatchClientTest(unittest.TestCase, NetworkdTestingUtilities):
 916     """Test [Match] sections in .network files.
 917
 918     Be aware that matching the test host's interfaces will wipe their
 919     configuration, so as a precaution, all network files should have a
 920     restrictive [Match] section to only ever interfere with the
 921     temporary veth interfaces created here.
 922     """
 923
 924     def tearDown(self):
 925         """Stop networkd."""
 926         subprocess.call(['systemctl', 'stop', 'systemd-networkd'])
 927
 928     def test_basic_matching(self):
 929         """Verify the Name= line works throughout this class."""
 930         self.add_veth_pair('test_if1', 'fake_if2')
 931         self.write_network('test.network', "[Match]\nName=test_*\n[Network]")
 932         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 933         self.assert_link_states(test_if1='managed', fake_if2='unmanaged')
 934
 935     def test_inverted_matching(self):
 936         """Verify that a '!'-prefixed value inverts the match."""
 937         # Use a MAC address as the interfaces' common matching attribute
 938         # to avoid depending on udev, to support testing in containers.
 939         mac = '00:01:02:03:98:99'
 940         self.add_veth_pair('test_veth', 'test_peer',
 941                            ['addr', mac], ['addr', mac])
 942         self.write_network('no-veth.network', """\
 943 [Match]
 944 MACAddress=%s
 945 Name=!nonexistent *peer*
 946 [Network]""" % mac)
 947         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 948         self.assert_link_states(test_veth='managed', test_peer='unmanaged')
 949
 950
 951 class UnmanagedClientTest(unittest.TestCase, NetworkdTestingUtilities):
 952     """Test if networkd manages the correct interfaces."""
 953
 954     def setUp(self):
 955         """Write .network files to match the named veth devices."""
 956         # Define the veth+peer pairs to be created.
 957         # Their pairing doesn't actually matter, only their names do.
 958         self.veths = {
 959             'm1def': 'm0unm',
 960             'm1man': 'm1unm',
 961         }
 962
 963         # Define the contents of .network files to be read in order.
 964         self.configs = (
 965             "[Match]\nName=m1def\n",
 966             "[Match]\nName=m1unm\n[Link]\nUnmanaged=yes\n",
 967             "[Match]\nName=m1*\n[Link]\nUnmanaged=no\n",
 968         )
 969
 970         # Write out the .network files to be cleaned up automatically.
 971         for i, config in enumerate(self.configs):
 972             self.write_network("%02d-test.network" % i, config)
 973
 974     def tearDown(self):
 975         """Stop networkd."""
 976         subprocess.call(['systemctl', 'stop', 'systemd-networkd'])
 977
 978     def create_iface(self):
 979         """Create temporary veth pairs for interface matching."""
 980         for veth, peer in self.veths.items():
 981             self.add_veth_pair(veth, peer)
 982
 983     def test_unmanaged_setting(self):
 984         """Verify link states with Unmanaged= settings, hot-plug."""
 985         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 986         self.create_iface()
 987         self.assert_link_states(m1def='managed',
 988                                 m1man='managed',
 989                                 m1unm='unmanaged',
 990                                 m0unm='unmanaged')
 991
 992     def test_unmanaged_setting_coldplug(self):
 993         """Verify link states with Unmanaged= settings, cold-plug."""
 994         self.create_iface()
 995         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 996         self.assert_link_states(m1def='managed',
 997                                 m1man='managed',
 998                                 m1unm='unmanaged',
 999                                 m0unm='unmanaged')
1000
1001     def test_catchall_config(self):
1002         """Verify link states with a catch-all config, hot-plug."""
1003         # Don't actually catch ALL interfaces.  It messes up the host.
1004         self.write_network('all.network', "[Match]\nName=m[01]???\n")
1005         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
1006         self.create_iface()
1007         self.assert_link_states(m1def='managed',
1008                                 m1man='managed',
1009                                 m1unm='unmanaged',
1010                                 m0unm='managed')
1011
1012     def test_catchall_config_coldplug(self):
1013         """Verify link states with a catch-all config, cold-plug."""
1014         # Don't actually catch ALL interfaces.  It messes up the host.
1015         self.write_network('all.network', "[Match]\nName=m[01]???\n")
1016         self.create_iface()
1017         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
1018         self.assert_link_states(m1def='managed',
1019                                 m1man='managed',
1020                                 m1unm='unmanaged',
1021                                 m0unm='managed')
1022
1023
1024 if __name__ == '__main__':
1025     unittest.main(testRunner=unittest.TextTestRunner(stream=sys.stdout,
1026                                                      verbosity=2))