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
 724 set -eu
 725 mkdir -p /run/systemd/network
 726 mkdir -p /run/systemd/netif
 727 mount -t tmpfs none /run/systemd/network
 728 mount -t tmpfs none /run/systemd/netif
 729 [ ! -e /run/dbus ] || mount -t tmpfs none /run/dbus
 730 # create router/client veth pair
 731 cat << EOF > /run/systemd/network/test.netdev
 732 [NetDev]
 733 Name=%(ifr)s
 734 Kind=veth
 735
 736 [Peer]
 737 Name=%(ifc)s
 738 EOF
 739
 740 cat << EOF > /run/systemd/network/test.network
 741 [Match]
 742 Name=%(ifr)s
 743
 744 [Network]
 745 Address=192.168.5.1/24
 746 %(addr6)s
 747 DHCPServer=yes
 748
 749 [DHCPServer]
 750 PoolOffset=10
 751 PoolSize=50
 752 DNS=192.168.5.1
 753 %(dhopts)s
 754 EOF
 755
 756 # run networkd as in systemd-networkd.service
 757 exec $(systemctl cat systemd-networkd.service | sed -n '/^ExecStart=/ { s/^.*=//; s/^[@+-]//; s/^!*//; p}')
 758 ''' % {'ifr': self.if_router, 'ifc': self.iface, 'addr6': ipv6 and 'Address=2600::1/64' or '',
 759        'dhopts': dhcpserver_opts or ''})
 760
 761             os.fchmod(fd, 0o755)
 762
 763         subprocess.check_call(['systemd-run', '--unit=networkd-test-router.service',
 764                                '-p', 'InaccessibleDirectories=-/etc/systemd/network',
 765                                '-p', 'InaccessibleDirectories=-/run/systemd/network',
 766                                '-p', 'InaccessibleDirectories=-/run/systemd/netif',
 767                                '--service-type=notify', script])
 768
 769         # wait until devices got created
 770         for _ in range(50):
 771             out = subprocess.check_output(['ip', 'a', 'show', 'dev', self.if_router])
 772             if b'state UP' in out and b'scope global' in out:
 773                 break
 774             time.sleep(0.1)
 775
 776     def shutdown_iface(self):
 777         '''Remove test interface and stop DHCP server'''
 778
 779         if self.if_router:
 780             subprocess.check_call(['systemctl', 'stop', 'networkd-test-router.service'])
 781             # ensure failed transient unit does not stay around
 782             subprocess.call(['systemctl', 'reset-failed', 'networkd-test-router.service'])
 783             subprocess.call(['ip', 'link', 'del', 'dev', self.if_router])
 784             self.if_router = None
 785
 786     def print_server_log(self):
 787         '''Print DHCP server log for debugging failures'''
 788
 789         self.show_journal('networkd-test-router.service')
 790
 791     @unittest.skip('networkd does not have DHCPv6 server support')
 792     def test_hotplug_dhcp_ip6(self):
 793         pass
 794
 795     @unittest.skip('networkd does not have DHCPv6 server support')
 796     def test_coldplug_dhcp_ip6(self):
 797         pass
 798
 799     def test_search_domains(self):
 800
 801         # we don't use this interface for this test
 802         self.if_router = None
 803
 804         self.write_network('test.netdev', '''\
 805 [NetDev]
 806 Name=dummy0
 807 Kind=dummy
 808 MACAddress=12:34:56:78:9a:bc''')
 809         self.write_network('test.network', '''\
 810 [Match]
 811 Name=dummy0
 812 [Network]
 813 Address=192.168.42.100
 814 DNS=192.168.42.1
 815 Domains= one two three four five six seven eight nine ten''')
 816
 817         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 818
 819         for timeout in range(50):
 820             with open(RESOLV_CONF) as f:
 821                 contents = f.read()
 822             if ' one' in contents:
 823                 break
 824             time.sleep(0.1)
 825         self.assertRegex(contents, 'search .*one two three four')
 826         self.assertNotIn('seven\n', contents)
 827         self.assertIn('# Too many search domains configured, remaining ones ignored.\n', contents)
 828
 829     def test_search_domains_too_long(self):
 830
 831         # we don't use this interface for this test
 832         self.if_router = None
 833
 834         name_prefix = 'a' * 60
 835
 836         self.write_network('test.netdev', '''\
 837 [NetDev]
 838 Name=dummy0
 839 Kind=dummy
 840 MACAddress=12:34:56:78:9a:bc''')
 841         self.write_network('test.network', '''\
 842 [Match]
 843 Name=dummy0
 844 [Network]
 845 Address=192.168.42.100
 846 DNS=192.168.42.1
 847 Domains={p}0 {p}1 {p}2 {p}3 {p}4'''.format(p=name_prefix))
 848
 849         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 850
 851         for timeout in range(50):
 852             with open(RESOLV_CONF) as f:
 853                 contents = f.read()
 854             if ' one' in contents:
 855                 break
 856             time.sleep(0.1)
 857         self.assertRegex(contents, 'search .*{p}0 {p}1 {p}2'.format(p=name_prefix))
 858         self.assertIn('# Total length of all search domains is too long, remaining ones ignored.', contents)
 859
 860     def test_dropin(self):
 861         # we don't use this interface for this test
 862         self.if_router = None
 863
 864         self.write_network('test.netdev', '''\
 865 [NetDev]
 866 Name=dummy0
 867 Kind=dummy
 868 MACAddress=12:34:56:78:9a:bc''')
 869         self.write_network('test.network', '''\
 870 [Match]
 871 Name=dummy0
 872 [Network]
 873 Address=192.168.42.100
 874 DNS=192.168.42.1''')
 875         self.write_network_dropin('test.network', 'dns', '''\
 876 [Network]
 877 DNS=127.0.0.1''')
 878
 879         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 880
 881         for timeout in range(50):
 882             with open(RESOLV_CONF) as f:
 883                 contents = f.read()
 884             if ' 127.0.0.1' in contents:
 885                 break
 886             time.sleep(0.1)
 887         self.assertIn('nameserver 192.168.42.1\n', contents)
 888         self.assertIn('nameserver 127.0.0.1\n', contents)
 889
 890     def test_dhcp_timezone(self):
 891         '''networkd sets time zone from DHCP'''
 892
 893         def get_tz():
 894             out = subprocess.check_output(['busctl', 'get-property', 'org.freedesktop.timedate1',
 895                                            '/org/freedesktop/timedate1', 'org.freedesktop.timedate1', 'Timezone'])
 896             assert out.startswith(b's "')
 897             out = out.strip()
 898             assert out.endswith(b'"')
 899             return out[3:-1].decode()
 900
 901         orig_timezone = get_tz()
 902         self.addCleanup(subprocess.call, ['timedatectl', 'set-timezone', orig_timezone])
 903
 904         self.create_iface(dhcpserver_opts='EmitTimezone=yes\nTimezone=Pacific/Honolulu')
 905         self.do_test(coldplug=None, extra_opts='IPv6AcceptRA=false\n[DHCP]\nUseTimezone=true', dhcp_mode='ipv4')
 906
 907         # should have applied the received timezone
 908         try:
 909             self.assertEqual(get_tz(), 'Pacific/Honolulu')
 910         except AssertionError:
 911             self.show_journal('systemd-networkd.service')
 912             self.show_journal('systemd-hostnamed.service')
 913             raise
 914
 915
 916 class MatchClientTest(unittest.TestCase, NetworkdTestingUtilities):
 917     """Test [Match] sections in .network files.
 918
 919     Be aware that matching the test host's interfaces will wipe their
 920     configuration, so as a precaution, all network files should have a
 921     restrictive [Match] section to only ever interfere with the
 922     temporary veth interfaces created here.
 923     """
 924
 925     def tearDown(self):
 926         """Stop networkd."""
 927         subprocess.call(['systemctl', 'stop', 'systemd-networkd'])
 928
 929     def test_basic_matching(self):
 930         """Verify the Name= line works throughout this class."""
 931         self.add_veth_pair('test_if1', 'fake_if2')
 932         self.write_network('test.network', "[Match]\nName=test_*\n[Network]")
 933         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 934         self.assert_link_states(test_if1='managed', fake_if2='unmanaged')
 935
 936     def test_inverted_matching(self):
 937         """Verify that a '!'-prefixed value inverts the match."""
 938         # Use a MAC address as the interfaces' common matching attribute
 939         # to avoid depending on udev, to support testing in containers.
 940         mac = '00:01:02:03:98:99'
 941         self.add_veth_pair('test_veth', 'test_peer',
 942                            ['addr', mac], ['addr', mac])
 943         self.write_network('no-veth.network', """\
 944 [Match]
 945 MACAddress=%s
 946 Name=!nonexistent *peer*
 947 [Network]""" % mac)
 948         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 949         self.assert_link_states(test_veth='managed', test_peer='unmanaged')
 950
 951
 952 class UnmanagedClientTest(unittest.TestCase, NetworkdTestingUtilities):
 953     """Test if networkd manages the correct interfaces."""
 954
 955     def setUp(self):
 956         """Write .network files to match the named veth devices."""
 957         # Define the veth+peer pairs to be created.
 958         # Their pairing doesn't actually matter, only their names do.
 959         self.veths = {
 960             'm1def': 'm0unm',
 961             'm1man': 'm1unm',
 962         }
 963
 964         # Define the contents of .network files to be read in order.
 965         self.configs = (
 966             "[Match]\nName=m1def\n",
 967             "[Match]\nName=m1unm\n[Link]\nUnmanaged=yes\n",
 968             "[Match]\nName=m1*\n[Link]\nUnmanaged=no\n",
 969         )
 970
 971         # Write out the .network files to be cleaned up automatically.
 972         for i, config in enumerate(self.configs):
 973             self.write_network("%02d-test.network" % i, config)
 974
 975     def tearDown(self):
 976         """Stop networkd."""
 977         subprocess.call(['systemctl', 'stop', 'systemd-networkd'])
 978
 979     def create_iface(self):
 980         """Create temporary veth pairs for interface matching."""
 981         for veth, peer in self.veths.items():
 982             self.add_veth_pair(veth, peer)
 983
 984     def test_unmanaged_setting(self):
 985         """Verify link states with Unmanaged= settings, hot-plug."""
 986         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 987         self.create_iface()
 988         self.assert_link_states(m1def='managed',
 989                                 m1man='managed',
 990                                 m1unm='unmanaged',
 991                                 m0unm='unmanaged')
 992
 993     def test_unmanaged_setting_coldplug(self):
 994         """Verify link states with Unmanaged= settings, cold-plug."""
 995         self.create_iface()
 996         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
 997         self.assert_link_states(m1def='managed',
 998                                 m1man='managed',
 999                                 m1unm='unmanaged',
1000                                 m0unm='unmanaged')
1001
1002     def test_catchall_config(self):
1003         """Verify link states with a catch-all config, hot-plug."""
1004         # Don't actually catch ALL interfaces.  It messes up the host.
1005         self.write_network('all.network', "[Match]\nName=m[01]???\n")
1006         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
1007         self.create_iface()
1008         self.assert_link_states(m1def='managed',
1009                                 m1man='managed',
1010                                 m1unm='unmanaged',
1011                                 m0unm='managed')
1012
1013     def test_catchall_config_coldplug(self):
1014         """Verify link states with a catch-all config, cold-plug."""
1015         # Don't actually catch ALL interfaces.  It messes up the host.
1016         self.write_network('all.network', "[Match]\nName=m[01]???\n")
1017         self.create_iface()
1018         subprocess.check_call(['systemctl', 'start', 'systemd-networkd'])
1019         self.assert_link_states(m1def='managed',
1020                                 m1man='managed',
1021                                 m1unm='unmanaged',
1022                                 m0unm='managed')
1023
1024
1025 if __name__ == '__main__':
1026     unittest.main(testRunner=unittest.TextTestRunner(stream=sys.stdout,
1027                                                      verbosity=2))