test/networkd-test.py

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