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