]>
git.ipfire.org Git - thirdparty/systemd.git/blob - test/networkd-test.py
2 # SPDX-License-Identifier: LGPL-2.1-or-later
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
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.
15 # ATTENTION: This uses the *installed* networkd, not the one from the built
18 # © 2015 Canonical Ltd.
19 # Author: Martin Pitt <martin.pitt@ubuntu.com>
31 HAVE_DNSMASQ
= shutil
.which('dnsmasq') is not None
32 IS_CONTAINER
= subprocess
.call(['systemd-detect-virt', '--quiet', '--container']) == 0
34 NETWORK_UNITDIR
= '/run/systemd/network'
36 NETWORKD_WAIT_ONLINE
= shutil
.which('systemd-networkd-wait-online',
37 path
='/usr/lib/systemd:/lib/systemd')
39 RESOLV_CONF
= '/run/systemd/resolve/resolv.conf'
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')
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')
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
)
64 stopped_units
.append(u
)
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'])
72 for d
in ['/etc/systemd/network', '/run/systemd/network',
73 '/run/systemd/netif', '/run/systemd/resolve']:
75 subprocess
.check_call(["mount", "-t", "tmpfs", "none", 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')
84 # Avoid "Failed to open /dev/tty" errors in containers.
85 os
.environ
['SYSTEMD_LOG_TARGET'] = 'journal'
87 # Ensure the unit directory exists so tests can dump files into it.
88 os
.makedirs(NETWORK_UNITDIR
, exist_ok
=True)
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
])
101 class NetworkdTestingUtilities
:
102 """Provide a set of utility functions to facilitate networkd tests.
104 This class must be inherited along with unittest.TestCase to define
105 some required methods.
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
] +
112 ['type', 'veth', 'peer', 'name', peer
] +
114 self
.addCleanup(subprocess
.call
, ['ip', 'link', 'del', 'dev', peer
])
116 def write_config(self
, path
, contents
):
117 """"Write a configuration file, and queue it to be removed."""
119 with
open(path
, 'w') as f
:
122 self
.addCleanup(os
.remove
, path
)
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
)
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
))
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
)
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()
148 def assert_link_states(self
, **kwargs
):
149 """Match networkctl link states to the given ones.
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
158 A special interface state "managed" is supported, which matches
159 any value in the "SETUP" column other than "unmanaged".
163 interfaces
= set(kwargs
)
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
])
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
:
175 expected
= kwargs
[iface
]
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
)
182 # Ensure that all requested interfaces have been covered.
184 self
.fail("Missing links in status output: {}".format(interfaces
))
187 class BridgeTest(NetworkdTestingUtilities
, unittest
.TestCase
):
188 """Provide common methods for testing networkd against servers."""
191 self
.write_network('port1.netdev', '''\
195 MACAddress=12:34:56:78:9a:bc
197 self
.write_network('port2.netdev', '''\
201 MACAddress=12:34:56:78:9a:bd
203 self
.write_network('mybridge.netdev', '''\
208 self
.write_network('port1.network', '''\
214 self
.write_network('port2.network', '''\
220 self
.write_network('mybridge.network', '''\
225 Address=192.168.250.33/24
226 Gateway=192.168.250.1
228 subprocess
.call(['systemctl', 'reset-failed', 'systemd-networkd', 'systemd-resolved'])
229 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
232 subprocess
.check_call(['systemctl', 'stop', 'systemd-networkd.socket'])
233 subprocess
.check_call(['systemctl', 'stop', 'systemd-networkd.service'])
234 subprocess
.check_call(['ip', 'link', 'del', 'mybridge'])
235 subprocess
.check_call(['ip', 'link', 'del', 'port1'])
236 subprocess
.check_call(['ip', 'link', 'del', 'port2'])
238 def test_bridge_init(self
):
239 self
.assert_link_states(
244 def test_bridge_port_priority(self
):
245 self
.assertEqual(self
.read_attr('port1', 'brport/priority'), '32')
246 self
.write_network_dropin('port1.network', 'priority', '''\
250 subprocess
.check_call(['ip', 'link', 'set', 'dev', 'port1', 'down'])
251 subprocess
.check_call(['systemctl', 'restart', 'systemd-networkd'])
252 subprocess
.check_call([NETWORKD_WAIT_ONLINE
, '--interface',
253 'port1', '--timeout=5'])
254 self
.assertEqual(self
.read_attr('port1', 'brport/priority'), '28')
256 def test_bridge_port_priority_set_zero(self
):
257 """It should be possible to set the bridge port priority to 0"""
258 self
.assertEqual(self
.read_attr('port2', 'brport/priority'), '32')
259 self
.write_network_dropin('port2.network', 'priority', '''\
263 subprocess
.check_call(['ip', 'link', 'set', 'dev', 'port2', 'down'])
264 subprocess
.check_call(['systemctl', 'restart', 'systemd-networkd'])
265 subprocess
.check_call([NETWORKD_WAIT_ONLINE
, '--interface',
266 'port2', '--timeout=5'])
267 self
.assertEqual(self
.read_attr('port2', 'brport/priority'), '0')
269 def test_bridge_port_property(self
):
270 """Test the "[Bridge]" section keys"""
271 self
.assertEqual(self
.read_attr('port2', 'brport/priority'), '32')
272 self
.write_network_dropin('port2.network', 'property', '''\
278 AllowPortToBeRoot=true
282 subprocess
.check_call(['ip', 'link', 'set', 'dev', 'port2', 'down'])
283 subprocess
.check_call(['systemctl', 'restart', 'systemd-networkd'])
284 subprocess
.check_call([NETWORKD_WAIT_ONLINE
, '--interface',
285 'port2', '--timeout=5'])
287 self
.assertEqual(self
.read_attr('port2', 'brport/priority'), '23')
288 self
.assertEqual(self
.read_attr('port2', 'brport/hairpin_mode'), '1')
289 self
.assertEqual(self
.read_attr('port2', 'brport/path_cost'), '555')
290 self
.assertEqual(self
.read_attr('port2', 'brport/multicast_fast_leave'), '1')
291 self
.assertEqual(self
.read_attr('port2', 'brport/unicast_flood'), '1')
292 self
.assertEqual(self
.read_attr('port2', 'brport/bpdu_guard'), '1')
293 self
.assertEqual(self
.read_attr('port2', 'brport/root_block'), '1')
295 class ClientTestBase(NetworkdTestingUtilities
):
296 """Provide common methods for testing networkd against servers."""
299 def setUpClass(klass
):
300 klass
.orig_log_level
= subprocess
.check_output(
301 ['systemctl', 'show', '--value', '--property', 'LogLevel'],
302 universal_newlines
=True).strip()
303 subprocess
.check_call(['systemd-analyze', 'log-level', 'debug'])
306 def tearDownClass(klass
):
307 subprocess
.check_call(['systemd-analyze', 'log-level', klass
.orig_log_level
])
310 self
.iface
= 'test_eth42'
311 self
.if_router
= 'router_eth42'
312 self
.workdir_obj
= tempfile
.TemporaryDirectory()
313 self
.workdir
= self
.workdir_obj
.name
314 self
.config
= 'test_eth42.network'
316 # get current journal cursor
317 subprocess
.check_output(['journalctl', '--sync'])
318 out
= subprocess
.check_output(['journalctl', '-b', '--quiet',
319 '--no-pager', '-n0', '--show-cursor'],
320 universal_newlines
=True)
321 self
.assertTrue(out
.startswith('-- cursor:'))
322 self
.journal_cursor
= out
.split()[-1]
324 subprocess
.call(['systemctl', 'reset-failed', 'systemd-networkd', 'systemd-resolved'])
327 self
.shutdown_iface()
328 subprocess
.call(['systemctl', 'stop', 'systemd-networkd.socket'])
329 subprocess
.call(['systemctl', 'stop', 'systemd-networkd.service'])
330 subprocess
.call(['ip', 'link', 'del', 'dummy0'],
331 stderr
=subprocess
.DEVNULL
)
333 def show_journal(self
, unit
):
334 '''Show journal of given unit since start of the test'''
336 print('---- {} ----'.format(unit
))
337 subprocess
.check_output(['journalctl', '--sync'])
339 subprocess
.call(['journalctl', '-b', '--no-pager', '--quiet',
340 '--cursor', self
.journal_cursor
, '-u', unit
])
342 def create_iface(self
, ipv6
=False):
343 '''Create test interface with DHCP server behind it'''
345 raise NotImplementedError('must be implemented by a subclass')
347 def shutdown_iface(self
):
348 '''Remove test interface and stop DHCP server'''
350 raise NotImplementedError('must be implemented by a subclass')
352 def print_server_log(self
):
353 '''Print DHCP server log for debugging failures'''
355 raise NotImplementedError('must be implemented by a subclass')
357 def start_unit(self
, unit
):
359 subprocess
.check_call(['systemctl', 'start', unit
])
360 except subprocess
.CalledProcessError
:
361 self
.show_journal(unit
)
364 def do_test(self
, coldplug
=True, ipv6
=False, extra_opts
='',
365 online_timeout
=10, dhcp_mode
='yes'):
366 self
.start_unit('systemd-resolved')
367 self
.write_network(self
.config
, '''\
373 '''.format(iface
=self
.iface
, dhcp_mode
=dhcp_mode
, extra_opts
=extra_opts
))
376 # create interface first, then start networkd
377 self
.create_iface(ipv6
=ipv6
)
378 self
.start_unit('systemd-networkd')
379 elif coldplug
is not None:
380 # start networkd first, then create interface
381 self
.start_unit('systemd-networkd')
382 self
.create_iface(ipv6
=ipv6
)
384 # "None" means test sets up interface by itself
385 self
.start_unit('systemd-networkd')
388 subprocess
.check_call([NETWORKD_WAIT_ONLINE
, '--interface',
389 self
.iface
, '--timeout=%i' % online_timeout
])
392 # check iface state and IP 6 address; FIXME: we need to wait a bit
393 # longer, as the iface is "configured" already with IPv4 *or*
394 # IPv6, but we want to wait for both
396 out
= subprocess
.check_output(['ip', 'a', 'show', 'dev', self
.iface
])
397 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
:
401 self
.fail('timed out waiting for IPv6 configuration')
403 self
.assertRegex(out
, b
'inet6 2600::.* scope global .*dynamic')
404 self
.assertRegex(out
, b
'inet6 fe80::.* scope link')
406 # should have link-local address on IPv6 only
407 out
= subprocess
.check_output(['ip', '-6', 'a', 'show', 'dev', self
.iface
])
408 self
.assertRegex(out
, br
'inet6 fe80::.* scope link')
409 self
.assertNotIn(b
'scope global', out
)
411 # should have IPv4 address
412 out
= subprocess
.check_output(['ip', '-4', 'a', 'show', 'dev', self
.iface
])
413 self
.assertIn(b
'state UP', out
)
414 self
.assertRegex(out
, br
'inet 192.168.5.\d+/.* scope global dynamic')
416 # check networkctl state
417 out
= subprocess
.check_output(['networkctl'])
418 self
.assertRegex(out
, (r
'{}\s+ether\s+[a-z-]+\s+unmanaged'.format(self
.if_router
)).encode())
419 self
.assertRegex(out
, (r
'{}\s+ether\s+routable\s+configured'.format(self
.iface
)).encode())
421 out
= subprocess
.check_output(['networkctl', '-n', '0', 'status', self
.iface
])
422 self
.assertRegex(out
, br
'Type:\s+ether')
423 self
.assertRegex(out
, br
'State:\s+routable.*configured')
424 self
.assertRegex(out
, br
'Online state:\s+online')
425 self
.assertRegex(out
, br
'Address:\s+192.168.5.\d+')
427 self
.assertRegex(out
, br
'2600::')
429 self
.assertNotIn(br
'2600::', out
)
430 self
.assertRegex(out
, br
'fe80::')
431 self
.assertRegex(out
, br
'Gateway:\s+192.168.5.1')
432 self
.assertRegex(out
, br
'DNS:\s+192.168.5.1')
433 except (AssertionError, subprocess
.CalledProcessError
):
434 # show networkd status, journal, and DHCP server log on failure
435 with
open(os
.path
.join(NETWORK_UNITDIR
, self
.config
)) as f
:
436 print('\n---- {} ----\n{}'.format(self
.config
, f
.read()))
437 print('---- interface status ----')
439 subprocess
.call(['ip', 'a', 'show', 'dev', self
.iface
])
440 print('---- networkctl status {} ----'.format(self
.iface
))
442 rc
= subprocess
.call(['networkctl', '-n', '0', 'status', self
.iface
])
444 print("'networkctl status' exited with an unexpected code {}".format(rc
))
445 self
.show_journal('systemd-networkd.service')
446 self
.print_server_log()
449 for timeout
in range(50):
450 with
open(RESOLV_CONF
) as f
:
452 if 'nameserver 192.168.5.1\n' in contents
:
456 self
.fail('nameserver 192.168.5.1 not found in ' + RESOLV_CONF
)
458 if coldplug
is False:
459 # check post-down.d hook
460 self
.shutdown_iface()
462 def test_coldplug_dhcp_yes_ip4(self
):
463 # we have a 12s timeout on RA, so we need to wait longer
464 self
.do_test(coldplug
=True, ipv6
=False, online_timeout
=15)
466 def test_coldplug_dhcp_yes_ip4_no_ra(self
):
467 # with disabling RA explicitly things should be fast
468 self
.do_test(coldplug
=True, ipv6
=False,
469 extra_opts
='IPv6AcceptRA=False')
471 def test_coldplug_dhcp_ip4_only(self
):
472 # we have a 12s timeout on RA, so we need to wait longer
473 self
.do_test(coldplug
=True, ipv6
=False, dhcp_mode
='ipv4',
476 def test_coldplug_dhcp_ip4_only_no_ra(self
):
477 # with disabling RA explicitly things should be fast
478 self
.do_test(coldplug
=True, ipv6
=False, dhcp_mode
='ipv4',
479 extra_opts
='IPv6AcceptRA=False')
481 def test_coldplug_dhcp_ip6(self
):
482 self
.do_test(coldplug
=True, ipv6
=True)
484 def test_hotplug_dhcp_ip4(self
):
485 # With IPv4 only we have a 12s timeout on RA, so we need to wait longer
486 self
.do_test(coldplug
=False, ipv6
=False, online_timeout
=15)
488 def test_hotplug_dhcp_ip6(self
):
489 self
.do_test(coldplug
=False, ipv6
=True)
491 def test_route_only_dns(self
):
492 self
.write_network('myvpn.netdev', '''\
496 MACAddress=12:34:56:78:9a:bc
498 self
.write_network('myvpn.network', '''\
502 Address=192.168.42.100/24
508 self
.do_test(coldplug
=True, ipv6
=False,
509 extra_opts
='IPv6AcceptRouterAdvertisements=False')
510 except subprocess
.CalledProcessError
as e
:
511 # networkd often fails to start in LXC: https://github.com/systemd/systemd/issues/11848
512 if IS_CONTAINER
and e
.cmd
== ['systemctl', 'start', 'systemd-networkd']:
513 raise unittest
.SkipTest('https://github.com/systemd/systemd/issues/11848')
517 with
open(RESOLV_CONF
) as f
:
519 # ~company is not a search domain, only a routing domain
520 self
.assertNotRegex(contents
, 'search.*company')
521 # our global server should appear
522 self
.assertIn('nameserver 192.168.5.1\n', contents
)
523 # should not have domain-restricted server as global server
524 self
.assertNotIn('nameserver 192.168.42.1\n', contents
)
526 def test_route_only_dns_all_domains(self
):
527 self
.write_network('myvpn.netdev', '''[NetDev]
530 MACAddress=12:34:56:78:9a:bc
532 self
.write_network('myvpn.network', '''[Match]
535 Address=192.168.42.100/24
541 self
.do_test(coldplug
=True, ipv6
=False,
542 extra_opts
='IPv6AcceptRouterAdvertisements=False')
543 except subprocess
.CalledProcessError
as e
:
544 # networkd often fails to start in LXC: https://github.com/systemd/systemd/issues/11848
545 if IS_CONTAINER
and e
.cmd
== ['systemctl', 'start', 'systemd-networkd']:
546 raise unittest
.SkipTest('https://github.com/systemd/systemd/issues/11848')
550 with
open(RESOLV_CONF
) as f
:
553 # ~company is not a search domain, only a routing domain
554 self
.assertNotRegex(contents
, 'search.*company')
556 # our global server should appear
557 self
.assertIn('nameserver 192.168.5.1\n', contents
)
558 # should have company server as global server due to ~.
559 self
.assertIn('nameserver 192.168.42.1\n', contents
)
562 @unittest.skipUnless(HAVE_DNSMASQ
, 'dnsmasq not installed')
563 class DnsmasqClientTest(ClientTestBase
, unittest
.TestCase
):
564 '''Test networkd client against dnsmasq'''
569 self
.iface_mac
= 'de:ad:be:ef:47:11'
571 def create_iface(self
, ipv6
=False, dnsmasq_opts
=None):
572 '''Create test interface with DHCP server behind it'''
575 subprocess
.check_call(['ip', 'link', 'add', 'name', self
.iface
,
576 'address', self
.iface_mac
,
577 'type', 'veth', 'peer', 'name', self
.if_router
])
579 # give our router an IP
580 subprocess
.check_call(['ip', 'a', 'flush', 'dev', self
.if_router
])
581 subprocess
.check_call(['ip', 'a', 'add', '192.168.5.1/24', 'dev', self
.if_router
])
583 subprocess
.check_call(['ip', 'a', 'add', '2600::1/64', 'dev', self
.if_router
])
584 subprocess
.check_call(['ip', 'link', 'set', self
.if_router
, 'up'])
587 self
.dnsmasq_log
= os
.path
.join(self
.workdir
, 'dnsmasq.log')
588 lease_file
= os
.path
.join(self
.workdir
, 'dnsmasq.leases')
590 extra_opts
= ['--enable-ra', '--dhcp-range=2600::10,2600::20']
594 extra_opts
+= dnsmasq_opts
595 self
.dnsmasq
= subprocess
.Popen(
596 ['dnsmasq', '--keep-in-foreground', '--log-queries',
597 '--log-facility=' + self
.dnsmasq_log
, '--conf-file=/dev/null',
598 '--dhcp-leasefile=' + lease_file
, '--bind-interfaces',
599 '--interface=' + self
.if_router
, '--except-interface=lo',
600 '--dhcp-range=192.168.5.10,192.168.5.200'] + extra_opts
)
602 def shutdown_iface(self
):
603 '''Remove test interface and stop DHCP server'''
606 subprocess
.check_call(['ip', 'link', 'del', 'dev', self
.if_router
])
607 self
.if_router
= None
613 def print_server_log(self
):
614 '''Print DHCP server log for debugging failures'''
616 with
open(self
.dnsmasq_log
) as f
:
617 sys
.stdout
.write('\n\n---- dnsmasq log ----\n{}\n------\n\n'.format(f
.read()))
619 def test_resolved_domain_restricted_dns(self
):
620 '''resolved: domain-restricted DNS servers'''
622 # enable DNSSEC in allow downgrade mode, and turn off stuff we don't want to test to make looking at logs easier
623 conf
= '/run/systemd/resolved.conf.d/test-enable-dnssec.conf'
624 os
.makedirs(os
.path
.dirname(conf
), exist_ok
=True)
625 with
open(conf
, 'w') as f
:
626 f
.write('[Resolve]\nDNSSEC=allow-downgrade\nLLMNR=no\nMulticastDNS=no\nDNSOverTLS=no\n')
627 self
.addCleanup(os
.remove
, conf
)
629 # create interface for generic connections; this will map all DNS names
631 self
.create_iface(dnsmasq_opts
=['--address=/#/192.168.42.1'])
632 self
.write_network('general.network', '''\
638 DNSSECNegativeTrustAnchors=search.example.com
639 '''.format(self
.iface
))
641 # create second device/dnsmasq for a .company/.lab VPN interface
642 # static IPs for simplicity
643 self
.add_veth_pair('testvpnclient', 'testvpnrouter')
644 subprocess
.check_call(['ip', 'a', 'flush', 'dev', 'testvpnrouter'])
645 subprocess
.check_call(['ip', 'a', 'add', '10.241.3.1/24', 'dev', 'testvpnrouter'])
646 subprocess
.check_call(['ip', 'link', 'set', 'testvpnrouter', 'up'])
648 vpn_dnsmasq_log
= os
.path
.join(self
.workdir
, 'dnsmasq-vpn.log')
649 vpn_dnsmasq
= subprocess
.Popen(
650 ['dnsmasq', '--keep-in-foreground', '--log-queries',
651 '--log-facility=' + vpn_dnsmasq_log
, '--conf-file=/dev/null',
652 '--dhcp-leasefile=/dev/null', '--bind-interfaces',
653 '--interface=testvpnrouter', '--except-interface=lo',
654 '--address=/math.lab/10.241.3.3', '--address=/cantina.company/10.241.4.4'])
655 self
.addCleanup(vpn_dnsmasq
.wait
)
656 self
.addCleanup(vpn_dnsmasq
.kill
)
658 self
.write_network('vpn.network', '''\
663 Address=10.241.3.2/24
665 Domains=~company ~lab
666 DNSSECNegativeTrustAnchors=company lab
669 self
.start_unit('systemd-networkd')
670 subprocess
.check_call([NETWORKD_WAIT_ONLINE
, '--interface', self
.iface
,
671 '--interface=testvpnclient', '--timeout=20'])
673 # ensure we start fresh with every test
674 subprocess
.check_call(['systemctl', 'restart', 'systemd-resolved'])
676 # test vpnclient specific domains; these should *not* be answered by
678 out
= subprocess
.check_output(['resolvectl', 'query', 'math.lab'])
679 self
.assertIn(b
'math.lab: 10.241.3.3', out
)
680 out
= subprocess
.check_output(['resolvectl', 'query', 'kettle.cantina.company'])
681 self
.assertIn(b
'kettle.cantina.company: 10.241.4.4', out
)
683 # test general domains
684 out
= subprocess
.check_output(['resolvectl', 'query', 'search.example.com'])
685 self
.assertIn(b
'search.example.com: 192.168.42.1', out
)
687 with
open(self
.dnsmasq_log
) as f
:
688 general_log
= f
.read()
689 with
open(vpn_dnsmasq_log
) as f
:
692 # VPN domains should only be sent to VPN DNS
693 self
.assertRegex(vpn_log
, 'query.*math.lab')
694 self
.assertRegex(vpn_log
, 'query.*cantina.company')
695 self
.assertNotIn('.lab', general_log
)
696 self
.assertNotIn('.company', general_log
)
698 # general domains should not be sent to the VPN DNS
699 self
.assertRegex(general_log
, 'query.*search.example.com')
700 self
.assertNotIn('search.example.com', vpn_log
)
702 def test_resolved_etc_hosts(self
):
703 '''resolved queries to /etc/hosts'''
705 # enabled DNSSEC in allow-downgrade mode
706 conf
= '/run/systemd/resolved.conf.d/test-enable-dnssec.conf'
707 os
.makedirs(os
.path
.dirname(conf
), exist_ok
=True)
708 with
open(conf
, 'w') as f
:
709 f
.write('[Resolve]\nDNSSEC=allow-downgrade\nLLMNR=no\nMulticastDNS=no\nDNSOverTLS=no\n')
710 self
.addCleanup(os
.remove
, conf
)
712 # Add example.com to NTA list for this test
713 negative
= '/run/dnssec-trust-anchors.d/example.com.negative'
714 os
.makedirs(os
.path
.dirname(negative
), exist_ok
=True)
715 with
open(negative
, 'w') as f
:
716 f
.write('example.com\n16.172.in-addr.arpa\n')
717 self
.addCleanup(os
.remove
, negative
)
719 # create /etc/hosts bind mount which resolves my.example.com for IPv4
720 hosts
= os
.path
.join(self
.workdir
, 'hosts')
721 with
open(hosts
, 'w') as f
:
722 f
.write('172.16.99.99 my.example.com\n')
723 subprocess
.check_call(['mount', '--bind', hosts
, '/etc/hosts'])
724 self
.addCleanup(subprocess
.call
, ['umount', '/etc/hosts'])
725 subprocess
.check_call(['systemctl', 'restart', 'systemd-resolved.service'])
727 # note: different IPv4 address here, so that it's easy to tell apart
728 # what resolved the query
729 self
.create_iface(dnsmasq_opts
=['--host-record=my.example.com,172.16.99.1,2600::99:99',
730 '--host-record=other.example.com,172.16.0.42,2600::42',
731 '--mx-host=example.com,mail.example.com'],
733 self
.do_test(coldplug
=None, ipv6
=True)
736 # family specific queries
737 out
= subprocess
.check_output(['resolvectl', 'query', '-4', 'my.example.com'])
738 self
.assertIn(b
'my.example.com: 172.16.99.99', out
)
739 # we don't expect an IPv6 answer; if /etc/hosts has any IP address,
740 # it's considered a sufficient source
741 self
.assertNotEqual(subprocess
.call(['resolvectl', 'query', '-6', 'my.example.com']), 0)
742 # "any family" query; IPv4 should come from /etc/hosts
743 out
= subprocess
.check_output(['resolvectl', 'query', 'my.example.com'])
744 self
.assertIn(b
'my.example.com: 172.16.99.99', out
)
745 # IP → name lookup; again, takes the /etc/hosts one
746 out
= subprocess
.check_output(['resolvectl', 'query', '172.16.99.99'])
747 self
.assertIn(b
'172.16.99.99: my.example.com', out
)
749 # non-address RRs should fall back to DNS
750 out
= subprocess
.check_output(['resolvectl', 'query', '--type=MX', 'example.com'])
751 self
.assertIn(b
'example.com IN MX 1 mail.example.com', out
)
753 # other domains query DNS
754 out
= subprocess
.check_output(['resolvectl', 'query', 'other.example.com'])
755 self
.assertIn(b
'172.16.0.42', out
)
756 out
= subprocess
.check_output(['resolvectl', 'query', '172.16.0.42'])
757 self
.assertIn(b
'172.16.0.42: other.example.com', out
)
758 except (AssertionError, subprocess
.CalledProcessError
):
759 self
.show_journal('systemd-resolved.service')
760 self
.print_server_log()
763 def test_transient_hostname(self
):
764 '''networkd sets transient hostname from DHCP'''
766 orig_hostname
= socket
.gethostname()
767 self
.addCleanup(socket
.sethostname
, orig_hostname
)
768 # temporarily move /etc/hostname away; restart hostnamed to pick it up
769 if os
.path
.exists('/etc/hostname'):
770 subprocess
.check_call(['mount', '--bind', '/dev/null', '/etc/hostname'])
771 self
.addCleanup(subprocess
.call
, ['umount', '/etc/hostname'])
772 subprocess
.check_call(['systemctl', 'stop', 'systemd-hostnamed.service'])
773 self
.addCleanup(subprocess
.call
, ['systemctl', 'stop', 'systemd-hostnamed.service'])
775 self
.create_iface(dnsmasq_opts
=['--dhcp-host={},192.168.5.210,testgreen'.format(self
.iface_mac
)])
776 self
.do_test(coldplug
=None, extra_opts
='IPv6AcceptRA=False', dhcp_mode
='ipv4')
779 # should have received the fixed IP above
780 out
= subprocess
.check_output(['ip', '-4', 'a', 'show', 'dev', self
.iface
])
781 self
.assertRegex(out
, b
'inet 192.168.5.210/24 .* scope global dynamic')
782 # should have set transient hostname in hostnamed; this is
783 # sometimes a bit lagging (issue #4753), so retry a few times
784 for retry
in range(1, 6):
785 out
= subprocess
.check_output(['hostnamectl'])
786 if b
'testgreen' in out
:
789 sys
.stdout
.write('[retry %i] ' % retry
)
792 self
.fail('Transient hostname not found in hostnamectl:\n{}'.format(out
.decode()))
793 # and also applied to the system
794 self
.assertEqual(socket
.gethostname(), 'testgreen')
795 except AssertionError:
796 self
.show_journal('systemd-networkd.service')
797 self
.show_journal('systemd-hostnamed.service')
798 self
.print_server_log()
801 def test_transient_hostname_with_static(self
):
802 '''transient hostname is not applied if static hostname exists'''
804 orig_hostname
= socket
.gethostname()
805 self
.addCleanup(socket
.sethostname
, orig_hostname
)
807 if not os
.path
.exists('/etc/hostname'):
808 self
.write_config('/etc/hostname', "foobarqux")
810 self
.write_config('/run/hostname.tmp', "foobarqux")
811 subprocess
.check_call(['mount', '--bind', '/run/hostname.tmp', '/etc/hostname'])
812 self
.addCleanup(subprocess
.call
, ['umount', '/etc/hostname'])
814 socket
.sethostname("foobarqux");
816 subprocess
.check_call(['systemctl', 'stop', 'systemd-hostnamed.service'])
817 self
.addCleanup(subprocess
.call
, ['systemctl', 'stop', 'systemd-hostnamed.service'])
819 self
.create_iface(dnsmasq_opts
=['--dhcp-host={},192.168.5.210,testgreen'.format(self
.iface_mac
)])
820 self
.do_test(coldplug
=None, extra_opts
='IPv6AcceptRA=False', dhcp_mode
='ipv4')
823 # should have received the fixed IP above
824 out
= subprocess
.check_output(['ip', '-4', 'a', 'show', 'dev', self
.iface
])
825 self
.assertRegex(out
, b
'inet 192.168.5.210/24 .* scope global dynamic')
826 # static hostname wins over transient one, thus *not* applied
827 self
.assertEqual(socket
.gethostname(), "foobarqux")
828 except AssertionError:
829 self
.show_journal('systemd-networkd.service')
830 self
.show_journal('systemd-hostnamed.service')
831 self
.print_server_log()
835 class NetworkdClientTest(ClientTestBase
, unittest
.TestCase
):
836 '''Test networkd client against networkd server'''
842 def create_iface(self
, ipv6
=False, dhcpserver_opts
=None):
843 '''Create test interface with DHCP server behind it'''
845 # run "router-side" networkd in own mount namespace to shield it from
846 # "client-side" configuration and networkd
847 (fd
, script
) = tempfile
.mkstemp(prefix
='networkd-router.sh')
848 self
.addCleanup(os
.remove
, script
)
849 with os
.fdopen(fd
, 'w+') as f
:
853 mkdir -p /run/systemd/network
854 mkdir -p /run/systemd/netif
855 mount -t tmpfs none /run/systemd/network
856 mount -t tmpfs none /run/systemd/netif
857 [ ! -e /run/dbus ] || mount -t tmpfs none /run/dbus
858 # create router/client veth pair
859 cat <<EOF >/run/systemd/network/test.netdev
868 cat <<EOF >/run/systemd/network/test.network
873 Address=192.168.5.1/24
884 # run networkd as in systemd-networkd.service
885 exec $(systemctl cat systemd-networkd.service | sed -n '/^ExecStart=/ {{ s/^.*=//; s/^[@+-]//; s/^!*//; p}}')
886 '''.format(ifr
=self
.if_router
,
888 addr6
=('Address=2600::1/64' if ipv6
else ''),
889 dhopts
=(dhcpserver_opts
or '')))
893 subprocess
.check_call(['systemd-run', '--unit=networkd-test-router.service',
894 '-p', 'InaccessibleDirectories=-/etc/systemd/network',
895 '-p', 'InaccessibleDirectories=-/run/systemd/network',
896 '-p', 'InaccessibleDirectories=-/run/systemd/netif',
897 '--service-type=notify', script
])
899 # wait until devices got created
901 out
= subprocess
.check_output(['ip', 'a', 'show', 'dev', self
.if_router
])
902 if b
'state UP' in out
and b
'scope global' in out
:
906 def shutdown_iface(self
):
907 '''Remove test interface and stop DHCP server'''
910 subprocess
.check_call(['systemctl', 'stop', 'networkd-test-router.service'])
911 # ensure failed transient unit does not stay around
912 subprocess
.call(['systemctl', 'reset-failed', 'networkd-test-router.service'])
913 subprocess
.call(['ip', 'link', 'del', 'dev', self
.if_router
])
914 self
.if_router
= None
916 def print_server_log(self
):
917 '''Print DHCP server log for debugging failures'''
919 self
.show_journal('networkd-test-router.service')
921 @unittest.skip('networkd does not have DHCPv6 server support')
922 def test_hotplug_dhcp_ip6(self
):
925 @unittest.skip('networkd does not have DHCPv6 server support')
926 def test_coldplug_dhcp_ip6(self
):
929 def test_search_domains(self
):
931 # we don't use this interface for this test
932 self
.if_router
= None
934 self
.write_network('test.netdev', '''\
938 MACAddress=12:34:56:78:9a:bc
940 self
.write_network('test.network', '''\
944 Address=192.168.42.100/24
946 Domains= one two three four five six seven eight nine ten
949 self
.start_unit('systemd-networkd')
951 for timeout
in range(50):
952 with
open(RESOLV_CONF
) as f
:
954 if ' one' in contents
:
957 self
.assertRegex(contents
, 'search .*one two three four five six seven eight nine ten')
959 def test_dropin(self
):
960 # we don't use this interface for this test
961 self
.if_router
= None
963 self
.write_network('test.netdev', '''\
967 MACAddress=12:34:56:78:9a:bc
969 self
.write_network('test.network', '''\
973 Address=192.168.42.100/24
976 self
.write_network_dropin('test.network', 'dns', '''\
981 self
.start_unit('systemd-resolved')
982 self
.start_unit('systemd-networkd')
984 for timeout
in range(50):
985 with
open(RESOLV_CONF
) as f
:
987 if ' 127.0.0.1' in contents
and '192.168.42.1' in contents
:
990 self
.assertIn('nameserver 192.168.42.1\n', contents
)
991 self
.assertIn('nameserver 127.0.0.1\n', contents
)
993 def test_dhcp_timezone(self
):
994 '''networkd sets time zone from DHCP'''
997 out
= subprocess
.check_output(['busctl', 'get-property', 'org.freedesktop.timedate1',
998 '/org/freedesktop/timedate1', 'org.freedesktop.timedate1', 'Timezone'])
999 assert out
.startswith(b
's "')
1001 assert out
.endswith(b
'"')
1002 return out
[3:-1].decode()
1004 orig_timezone
= get_tz()
1005 self
.addCleanup(subprocess
.call
, ['timedatectl', 'set-timezone', orig_timezone
])
1007 self
.create_iface(dhcpserver_opts
='EmitTimezone=yes\nTimezone=Pacific/Honolulu')
1008 self
.do_test(coldplug
=None, extra_opts
='IPv6AcceptRA=false\n[DHCP]\nUseTimezone=true', dhcp_mode
='ipv4')
1010 # should have applied the received timezone
1012 self
.assertEqual(get_tz(), 'Pacific/Honolulu')
1013 except AssertionError:
1014 self
.show_journal('systemd-networkd.service')
1015 self
.show_journal('systemd-hostnamed.service')
1019 class MatchClientTest(unittest
.TestCase
, NetworkdTestingUtilities
):
1020 """Test [Match] sections in .network files.
1022 Be aware that matching the test host's interfaces will wipe their
1023 configuration, so as a precaution, all network files should have a
1024 restrictive [Match] section to only ever interfere with the
1025 temporary veth interfaces created here.
1029 """Stop networkd."""
1030 subprocess
.call(['systemctl', 'stop', 'systemd-networkd.socket'])
1031 subprocess
.call(['systemctl', 'stop', 'systemd-networkd.service'])
1033 def test_basic_matching(self
):
1034 """Verify the Name= line works throughout this class."""
1035 self
.add_veth_pair('test_if1', 'fake_if2')
1036 self
.write_network('test.network', "[Match]\nName=test_*\n[Network]")
1037 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
1038 self
.assert_link_states(test_if1
='managed', fake_if2
='unmanaged')
1040 def test_inverted_matching(self
):
1041 """Verify that a '!'-prefixed value inverts the match."""
1042 # Use a MAC address as the interfaces' common matching attribute
1043 # to avoid depending on udev, to support testing in containers.
1044 mac
= '00:01:02:03:98:99'
1045 self
.add_veth_pair('test_veth', 'test_peer',
1046 ['addr', mac
], ['addr', mac
])
1047 self
.write_network('no-veth.network', """\
1050 Name=!nonexistent *peer*
1051 [Network]""".format(mac
))
1052 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
1053 self
.assert_link_states(test_veth
='managed', test_peer
='unmanaged')
1056 class UnmanagedClientTest(unittest
.TestCase
, NetworkdTestingUtilities
):
1057 """Test if networkd manages the correct interfaces."""
1060 """Write .network files to match the named veth devices."""
1061 # Define the veth+peer pairs to be created.
1062 # Their pairing doesn't actually matter, only their names do.
1068 # Define the contents of .network files to be read in order.
1070 "[Match]\nName=m1def\n",
1071 "[Match]\nName=m1unm\n[Link]\nUnmanaged=yes\n",
1072 "[Match]\nName=m1*\n[Link]\nUnmanaged=no\n",
1075 # Write out the .network files to be cleaned up automatically.
1076 for i
, config
in enumerate(self
.configs
):
1077 self
.write_network("%02d-test.network" % i
, config
)
1080 """Stop networkd."""
1081 subprocess
.call(['systemctl', 'stop', 'systemd-networkd.socket'])
1082 subprocess
.call(['systemctl', 'stop', 'systemd-networkd.service'])
1084 def create_iface(self
):
1085 """Create temporary veth pairs for interface matching."""
1086 for veth
, peer
in self
.veths
.items():
1087 self
.add_veth_pair(veth
, peer
)
1089 def test_unmanaged_setting(self
):
1090 """Verify link states with Unmanaged= settings, hot-plug."""
1091 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
1093 self
.assert_link_states(m1def
='managed',
1098 def test_unmanaged_setting_coldplug(self
):
1099 """Verify link states with Unmanaged= settings, cold-plug."""
1101 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
1102 self
.assert_link_states(m1def
='managed',
1107 def test_catchall_config(self
):
1108 """Verify link states with a catch-all config, hot-plug."""
1109 # Don't actually catch ALL interfaces. It messes up the host.
1110 self
.write_network('all.network', "[Match]\nName=m[01]???\n")
1111 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
1113 self
.assert_link_states(m1def
='managed',
1118 def test_catchall_config_coldplug(self
):
1119 """Verify link states with a catch-all config, cold-plug."""
1120 # Don't actually catch ALL interfaces. It messes up the host.
1121 self
.write_network('all.network', "[Match]\nName=m[01]???\n")
1123 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
1124 self
.assert_link_states(m1def
='managed',
1130 if __name__
== '__main__':
1131 unittest
.main(testRunner
=unittest
.TextTestRunner(stream
=sys
.stdout
,