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