]>
git.ipfire.org Git - thirdparty/systemd.git/blob - test/networkd-test.py
2 # SPDX-License-Identifier: LGPL-2.1+
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''')
196 self
.write_network('port2.netdev', '''\
200 MACAddress=12:34:56:78:9a:bd''')
201 self
.write_network('mybridge.netdev', '''\
205 self
.write_network('port1.network', '''\
210 self
.write_network('port2.network', '''\
215 self
.write_network('mybridge.network', '''\
220 Address=192.168.250.33/24
221 Gateway=192.168.250.1''')
222 subprocess
.call(['systemctl', 'reset-failed', 'systemd-networkd', 'systemd-resolved'])
223 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
226 subprocess
.check_call(['systemctl', 'stop', 'systemd-networkd'])
227 subprocess
.check_call(['ip', 'link', 'del', 'mybridge'])
228 subprocess
.check_call(['ip', 'link', 'del', 'port1'])
229 subprocess
.check_call(['ip', 'link', 'del', 'port2'])
231 def test_bridge_init(self
):
232 self
.assert_link_states(
237 def test_bridge_port_priority(self
):
238 self
.assertEqual(self
.read_attr('port1', 'brport/priority'), '32')
239 self
.write_network_dropin('port1.network', 'priority', '''\
243 subprocess
.check_call(['systemctl', 'restart', 'systemd-networkd'])
244 self
.assertEqual(self
.read_attr('port1', 'brport/priority'), '28')
246 def test_bridge_port_priority_set_zero(self
):
247 """It should be possible to set the bridge port priority to 0"""
248 self
.assertEqual(self
.read_attr('port2', 'brport/priority'), '32')
249 self
.write_network_dropin('port2.network', 'priority', '''\
253 subprocess
.check_call(['systemctl', 'restart', 'systemd-networkd'])
254 self
.assertEqual(self
.read_attr('port2', 'brport/priority'), '0')
256 def test_bridge_port_property(self
):
257 """Test the "[Bridge]" section keys"""
258 self
.assertEqual(self
.read_attr('port2', 'brport/priority'), '32')
259 self
.write_network_dropin('port2.network', 'property', '''\
265 AllowPortToBeRoot=true
269 subprocess
.check_call(['systemctl', 'restart', 'systemd-networkd'])
271 self
.assertEqual(self
.read_attr('port2', 'brport/priority'), '23')
272 self
.assertEqual(self
.read_attr('port2', 'brport/hairpin_mode'), '1')
273 self
.assertEqual(self
.read_attr('port2', 'brport/path_cost'), '555')
274 self
.assertEqual(self
.read_attr('port2', 'brport/multicast_fast_leave'), '1')
275 self
.assertEqual(self
.read_attr('port2', 'brport/unicast_flood'), '1')
276 self
.assertEqual(self
.read_attr('port2', 'brport/bpdu_guard'), '1')
277 self
.assertEqual(self
.read_attr('port2', 'brport/root_block'), '1')
279 class ClientTestBase(NetworkdTestingUtilities
):
280 """Provide common methods for testing networkd against servers."""
283 def setUpClass(klass
):
284 klass
.orig_log_level
= subprocess
.check_output(
285 ['systemctl', 'show', '--value', '--property', 'LogLevel'],
286 universal_newlines
=True).strip()
287 subprocess
.check_call(['systemd-analyze', 'log-level', 'debug'])
290 def tearDownClass(klass
):
291 subprocess
.check_call(['systemd-analyze', 'log-level', klass
.orig_log_level
])
294 self
.iface
= 'test_eth42'
295 self
.if_router
= 'router_eth42'
296 self
.workdir_obj
= tempfile
.TemporaryDirectory()
297 self
.workdir
= self
.workdir_obj
.name
298 self
.config
= 'test_eth42.network'
300 # get current journal cursor
301 subprocess
.check_output(['journalctl', '--sync'])
302 out
= subprocess
.check_output(['journalctl', '-b', '--quiet',
303 '--no-pager', '-n0', '--show-cursor'],
304 universal_newlines
=True)
305 self
.assertTrue(out
.startswith('-- cursor:'))
306 self
.journal_cursor
= out
.split()[-1]
308 subprocess
.call(['systemctl', 'reset-failed', 'systemd-networkd', 'systemd-resolved'])
311 self
.shutdown_iface()
312 subprocess
.call(['systemctl', 'stop', 'systemd-networkd'])
313 subprocess
.call(['ip', 'link', 'del', 'dummy0'],
314 stderr
=subprocess
.DEVNULL
)
316 def show_journal(self
, unit
):
317 '''Show journal of given unit since start of the test'''
319 print('---- {} ----'.format(unit
))
320 subprocess
.check_output(['journalctl', '--sync'])
322 subprocess
.call(['journalctl', '-b', '--no-pager', '--quiet',
323 '--cursor', self
.journal_cursor
, '-u', unit
])
325 def create_iface(self
, ipv6
=False):
326 '''Create test interface with DHCP server behind it'''
328 raise NotImplementedError('must be implemented by a subclass')
330 def shutdown_iface(self
):
331 '''Remove test interface and stop DHCP server'''
333 raise NotImplementedError('must be implemented by a subclass')
335 def print_server_log(self
):
336 '''Print DHCP server log for debugging failures'''
338 raise NotImplementedError('must be implemented by a subclass')
340 def start_unit(self
, unit
):
342 subprocess
.check_call(['systemctl', 'start', unit
])
343 except subprocess
.CalledProcessError
:
344 self
.show_journal(unit
)
347 def do_test(self
, coldplug
=True, ipv6
=False, extra_opts
='',
348 online_timeout
=10, dhcp_mode
='yes'):
349 self
.start_unit('systemd-resolved')
350 self
.write_network(self
.config
, '''\
355 {}'''.format(self
.iface
, dhcp_mode
, extra_opts
))
358 # create interface first, then start networkd
359 self
.create_iface(ipv6
=ipv6
)
360 self
.start_unit('systemd-networkd')
361 elif coldplug
is not None:
362 # start networkd first, then create interface
363 self
.start_unit('systemd-networkd')
364 self
.create_iface(ipv6
=ipv6
)
366 # "None" means test sets up interface by itself
367 self
.start_unit('systemd-networkd')
370 subprocess
.check_call([NETWORKD_WAIT_ONLINE
, '--interface',
371 self
.iface
, '--timeout=%i' % online_timeout
])
374 # check iface state and IP 6 address; FIXME: we need to wait a bit
375 # longer, as the iface is "configured" already with IPv4 *or*
376 # IPv6, but we want to wait for both
378 out
= subprocess
.check_output(['ip', 'a', 'show', 'dev', self
.iface
])
379 if b
'state UP' in out
and b
'inet6 2600' in out
and b
'inet 192.168' in out
:
383 self
.fail('timed out waiting for IPv6 configuration')
385 self
.assertRegex(out
, b
'inet6 2600::.* scope global .*dynamic')
386 self
.assertRegex(out
, b
'inet6 fe80::.* scope link')
388 # should have link-local address on IPv6 only
389 out
= subprocess
.check_output(['ip', '-6', 'a', 'show', 'dev', self
.iface
])
390 self
.assertRegex(out
, br
'inet6 fe80::.* scope link')
391 self
.assertNotIn(b
'scope global', out
)
393 # should have IPv4 address
394 out
= subprocess
.check_output(['ip', '-4', 'a', 'show', 'dev', self
.iface
])
395 self
.assertIn(b
'state UP', out
)
396 self
.assertRegex(out
, br
'inet 192.168.5.\d+/.* scope global dynamic')
398 # check networkctl state
399 out
= subprocess
.check_output(['networkctl'])
400 self
.assertRegex(out
, (r
'{}\s+ether\s+[a-z-]+\s+unmanaged'.format(self
.if_router
)).encode())
401 self
.assertRegex(out
, (r
'{}\s+ether\s+routable\s+configured'.format(self
.iface
)).encode())
403 out
= subprocess
.check_output(['networkctl', '-n', '0', 'status', self
.iface
])
404 self
.assertRegex(out
, br
'Type:\s+ether')
405 self
.assertRegex(out
, br
'State:\s+routable.*configured')
406 self
.assertRegex(out
, br
'Address:\s+192.168.5.\d+')
408 self
.assertRegex(out
, br
'2600::')
410 self
.assertNotIn(br
'2600::', out
)
411 self
.assertRegex(out
, br
'fe80::')
412 self
.assertRegex(out
, br
'Gateway:\s+192.168.5.1')
413 self
.assertRegex(out
, br
'DNS:\s+192.168.5.1')
414 except (AssertionError, subprocess
.CalledProcessError
):
415 # show networkd status, journal, and DHCP server log on failure
416 with
open(os
.path
.join(NETWORK_UNITDIR
, self
.config
)) as f
:
417 print('\n---- {} ----\n{}'.format(self
.config
, f
.read()))
418 print('---- interface status ----')
420 subprocess
.call(['ip', 'a', 'show', 'dev', self
.iface
])
421 print('---- networkctl status {} ----'.format(self
.iface
))
423 rc
= subprocess
.call(['networkctl', '-n', '0', 'status', self
.iface
])
425 print("'networkctl status' exited with an unexpected code {}".format(rc
))
426 self
.show_journal('systemd-networkd.service')
427 self
.print_server_log()
430 for timeout
in range(50):
431 with
open(RESOLV_CONF
) as f
:
433 if 'nameserver 192.168.5.1\n' in contents
:
437 self
.fail('nameserver 192.168.5.1 not found in ' + RESOLV_CONF
)
439 if coldplug
is False:
440 # check post-down.d hook
441 self
.shutdown_iface()
443 def test_coldplug_dhcp_yes_ip4(self
):
444 # we have a 12s timeout on RA, so we need to wait longer
445 self
.do_test(coldplug
=True, ipv6
=False, online_timeout
=15)
447 def test_coldplug_dhcp_yes_ip4_no_ra(self
):
448 # with disabling RA explicitly things should be fast
449 self
.do_test(coldplug
=True, ipv6
=False,
450 extra_opts
='IPv6AcceptRA=False')
452 def test_coldplug_dhcp_ip4_only(self
):
453 # we have a 12s timeout on RA, so we need to wait longer
454 self
.do_test(coldplug
=True, ipv6
=False, dhcp_mode
='ipv4',
457 def test_coldplug_dhcp_ip4_only_no_ra(self
):
458 # with disabling RA explicitly things should be fast
459 self
.do_test(coldplug
=True, ipv6
=False, dhcp_mode
='ipv4',
460 extra_opts
='IPv6AcceptRA=False')
462 def test_coldplug_dhcp_ip6(self
):
463 self
.do_test(coldplug
=True, ipv6
=True)
465 def test_hotplug_dhcp_ip4(self
):
466 # With IPv4 only we have a 12s timeout on RA, so we need to wait longer
467 self
.do_test(coldplug
=False, ipv6
=False, online_timeout
=15)
469 def test_hotplug_dhcp_ip6(self
):
470 self
.do_test(coldplug
=False, ipv6
=True)
472 def test_route_only_dns(self
):
473 self
.write_network('myvpn.netdev', '''\
477 MACAddress=12:34:56:78:9a:bc''')
478 self
.write_network('myvpn.network', '''\
482 Address=192.168.42.100/24
484 Domains= ~company''')
487 self
.do_test(coldplug
=True, ipv6
=False,
488 extra_opts
='IPv6AcceptRouterAdvertisements=False')
489 except subprocess
.CalledProcessError
as e
:
490 # networkd often fails to start in LXC: https://github.com/systemd/systemd/issues/11848
491 if IS_CONTAINER
and e
.cmd
== ['systemctl', 'start', 'systemd-networkd']:
492 raise unittest
.SkipTest('https://github.com/systemd/systemd/issues/11848')
496 with
open(RESOLV_CONF
) as f
:
498 # ~company is not a search domain, only a routing domain
499 self
.assertNotRegex(contents
, 'search.*company')
500 # our global server should appear
501 self
.assertIn('nameserver 192.168.5.1\n', contents
)
502 # should not have domain-restricted server as global server
503 self
.assertNotIn('nameserver 192.168.42.1\n', contents
)
505 def test_route_only_dns_all_domains(self
):
506 self
.write_network('myvpn.netdev', '''[NetDev]
509 MACAddress=12:34:56:78:9a:bc''')
510 self
.write_network('myvpn.network', '''[Match]
513 Address=192.168.42.100/24
515 Domains= ~company ~.''')
518 self
.do_test(coldplug
=True, ipv6
=False,
519 extra_opts
='IPv6AcceptRouterAdvertisements=False')
520 except subprocess
.CalledProcessError
as e
:
521 # networkd often fails to start in LXC: https://github.com/systemd/systemd/issues/11848
522 if IS_CONTAINER
and e
.cmd
== ['systemctl', 'start', 'systemd-networkd']:
523 raise unittest
.SkipTest('https://github.com/systemd/systemd/issues/11848')
527 with
open(RESOLV_CONF
) as f
:
530 # ~company is not a search domain, only a routing domain
531 self
.assertNotRegex(contents
, 'search.*company')
533 # our global server should appear
534 self
.assertIn('nameserver 192.168.5.1\n', contents
)
535 # should have company server as global server due to ~.
536 self
.assertIn('nameserver 192.168.42.1\n', contents
)
539 @unittest.skipUnless(HAVE_DNSMASQ
, 'dnsmasq not installed')
540 class DnsmasqClientTest(ClientTestBase
, unittest
.TestCase
):
541 '''Test networkd client against dnsmasq'''
546 self
.iface_mac
= 'de:ad:be:ef:47:11'
548 def create_iface(self
, ipv6
=False, dnsmasq_opts
=None):
549 '''Create test interface with DHCP server behind it'''
552 subprocess
.check_call(['ip', 'link', 'add', 'name', self
.iface
,
553 'address', self
.iface_mac
,
554 'type', 'veth', 'peer', 'name', self
.if_router
])
556 # give our router an IP
557 subprocess
.check_call(['ip', 'a', 'flush', 'dev', self
.if_router
])
558 subprocess
.check_call(['ip', 'a', 'add', '192.168.5.1/24', 'dev', self
.if_router
])
560 subprocess
.check_call(['ip', 'a', 'add', '2600::1/64', 'dev', self
.if_router
])
561 subprocess
.check_call(['ip', 'link', 'set', self
.if_router
, 'up'])
564 self
.dnsmasq_log
= os
.path
.join(self
.workdir
, 'dnsmasq.log')
565 lease_file
= os
.path
.join(self
.workdir
, 'dnsmasq.leases')
567 extra_opts
= ['--enable-ra', '--dhcp-range=2600::10,2600::20']
571 extra_opts
+= dnsmasq_opts
572 self
.dnsmasq
= subprocess
.Popen(
573 ['dnsmasq', '--keep-in-foreground', '--log-queries',
574 '--log-facility=' + self
.dnsmasq_log
, '--conf-file=/dev/null',
575 '--dhcp-leasefile=' + lease_file
, '--bind-interfaces',
576 '--interface=' + self
.if_router
, '--except-interface=lo',
577 '--dhcp-range=192.168.5.10,192.168.5.200'] + extra_opts
)
579 def shutdown_iface(self
):
580 '''Remove test interface and stop DHCP server'''
583 subprocess
.check_call(['ip', 'link', 'del', 'dev', self
.if_router
])
584 self
.if_router
= None
590 def print_server_log(self
):
591 '''Print DHCP server log for debugging failures'''
593 with
open(self
.dnsmasq_log
) as f
:
594 sys
.stdout
.write('\n\n---- dnsmasq log ----\n{}\n------\n\n'.format(f
.read()))
596 def test_resolved_domain_restricted_dns(self
):
597 '''resolved: domain-restricted DNS servers'''
599 # FIXME: resolvectl query fails with enabled DNSSEC against our dnsmasq
600 conf
= '/run/systemd/resolved.conf.d/test-disable-dnssec.conf'
601 os
.makedirs(os
.path
.dirname(conf
), exist_ok
=True)
602 with
open(conf
, 'w') as f
:
603 f
.write('[Resolve]\nDNSSEC=no\n')
604 self
.addCleanup(os
.remove
, conf
)
606 # create interface for generic connections; this will map all DNS names
608 self
.create_iface(dnsmasq_opts
=['--address=/#/192.168.42.1'])
609 self
.write_network('general.network', '''\
614 IPv6AcceptRA=False'''.format(self
.iface
))
616 # create second device/dnsmasq for a .company/.lab VPN interface
617 # static IPs for simplicity
618 self
.add_veth_pair('testvpnclient', 'testvpnrouter')
619 subprocess
.check_call(['ip', 'a', 'flush', 'dev', 'testvpnrouter'])
620 subprocess
.check_call(['ip', 'a', 'add', '10.241.3.1/24', 'dev', 'testvpnrouter'])
621 subprocess
.check_call(['ip', 'link', 'set', 'testvpnrouter', 'up'])
623 vpn_dnsmasq_log
= os
.path
.join(self
.workdir
, 'dnsmasq-vpn.log')
624 vpn_dnsmasq
= subprocess
.Popen(
625 ['dnsmasq', '--keep-in-foreground', '--log-queries',
626 '--log-facility=' + vpn_dnsmasq_log
, '--conf-file=/dev/null',
627 '--dhcp-leasefile=/dev/null', '--bind-interfaces',
628 '--interface=testvpnrouter', '--except-interface=lo',
629 '--address=/math.lab/10.241.3.3', '--address=/cantina.company/10.241.4.4'])
630 self
.addCleanup(vpn_dnsmasq
.wait
)
631 self
.addCleanup(vpn_dnsmasq
.kill
)
633 self
.write_network('vpn.network', '''\
638 Address=10.241.3.2/24
640 Domains= ~company ~lab''')
642 self
.start_unit('systemd-networkd')
643 subprocess
.check_call([NETWORKD_WAIT_ONLINE
, '--interface', self
.iface
,
644 '--interface=testvpnclient', '--timeout=20'])
646 # ensure we start fresh with every test
647 subprocess
.check_call(['systemctl', 'restart', 'systemd-resolved'])
649 # test vpnclient specific domains; these should *not* be answered by
651 out
= subprocess
.check_output(['resolvectl', 'query', 'math.lab'])
652 self
.assertIn(b
'math.lab: 10.241.3.3', out
)
653 out
= subprocess
.check_output(['resolvectl', 'query', 'kettle.cantina.company'])
654 self
.assertIn(b
'kettle.cantina.company: 10.241.4.4', out
)
656 # test general domains
657 out
= subprocess
.check_output(['resolvectl', 'query', 'megasearch.net'])
658 self
.assertIn(b
'megasearch.net: 192.168.42.1', out
)
660 with
open(self
.dnsmasq_log
) as f
:
661 general_log
= f
.read()
662 with
open(vpn_dnsmasq_log
) as f
:
665 # VPN domains should only be sent to VPN DNS
666 self
.assertRegex(vpn_log
, 'query.*math.lab')
667 self
.assertRegex(vpn_log
, 'query.*cantina.company')
668 self
.assertNotIn('.lab', general_log
)
669 self
.assertNotIn('.company', general_log
)
671 # general domains should not be sent to the VPN DNS
672 self
.assertRegex(general_log
, 'query.*megasearch.net')
673 self
.assertNotIn('megasearch.net', vpn_log
)
675 def test_resolved_etc_hosts(self
):
676 '''resolved queries to /etc/hosts'''
678 # FIXME: -t MX query fails with enabled DNSSEC (even when using
679 # the known negative trust anchor .internal instead of .example.com)
680 conf
= '/run/systemd/resolved.conf.d/test-disable-dnssec.conf'
681 os
.makedirs(os
.path
.dirname(conf
), exist_ok
=True)
682 with
open(conf
, 'w') as f
:
683 f
.write('[Resolve]\nDNSSEC=no\nLLMNR=no\nMulticastDNS=no\n')
684 self
.addCleanup(os
.remove
, conf
)
686 # create /etc/hosts bind mount which resolves my.example.com for IPv4
687 hosts
= os
.path
.join(self
.workdir
, 'hosts')
688 with
open(hosts
, 'w') as f
:
689 f
.write('172.16.99.99 my.example.com\n')
690 subprocess
.check_call(['mount', '--bind', hosts
, '/etc/hosts'])
691 self
.addCleanup(subprocess
.call
, ['umount', '/etc/hosts'])
692 subprocess
.check_call(['systemctl', 'stop', 'systemd-resolved.service'])
694 # note: different IPv4 address here, so that it's easy to tell apart
695 # what resolved the query
696 self
.create_iface(dnsmasq_opts
=['--host-record=my.example.com,172.16.99.1,2600::99:99',
697 '--host-record=other.example.com,172.16.0.42,2600::42',
698 '--mx-host=example.com,mail.example.com'],
700 self
.do_test(coldplug
=None, ipv6
=True)
703 # family specific queries
704 out
= subprocess
.check_output(['resolvectl', 'query', '-4', 'my.example.com'])
705 self
.assertIn(b
'my.example.com: 172.16.99.99', out
)
706 # we don't expect an IPv6 answer; if /etc/hosts has any IP address,
707 # it's considered a sufficient source
708 self
.assertNotEqual(subprocess
.call(['resolvectl', 'query', '-6', 'my.example.com']), 0)
709 # "any family" query; IPv4 should come from /etc/hosts
710 out
= subprocess
.check_output(['resolvectl', 'query', 'my.example.com'])
711 self
.assertIn(b
'my.example.com: 172.16.99.99', out
)
712 # IP → name lookup; again, takes the /etc/hosts one
713 out
= subprocess
.check_output(['resolvectl', 'query', '172.16.99.99'])
714 self
.assertIn(b
'172.16.99.99: my.example.com', out
)
716 # non-address RRs should fall back to DNS
717 out
= subprocess
.check_output(['resolvectl', 'query', '--type=MX', 'example.com'])
718 self
.assertIn(b
'example.com IN MX 1 mail.example.com', out
)
720 # other domains query DNS
721 out
= subprocess
.check_output(['resolvectl', 'query', 'other.example.com'])
722 self
.assertIn(b
'172.16.0.42', out
)
723 out
= subprocess
.check_output(['resolvectl', 'query', '172.16.0.42'])
724 self
.assertIn(b
'172.16.0.42: other.example.com', out
)
725 except (AssertionError, subprocess
.CalledProcessError
):
726 self
.show_journal('systemd-resolved.service')
727 self
.print_server_log()
730 def test_transient_hostname(self
):
731 '''networkd sets transient hostname from DHCP'''
733 orig_hostname
= socket
.gethostname()
734 self
.addCleanup(socket
.sethostname
, orig_hostname
)
735 # temporarily move /etc/hostname away; restart hostnamed to pick it up
736 if os
.path
.exists('/etc/hostname'):
737 subprocess
.check_call(['mount', '--bind', '/dev/null', '/etc/hostname'])
738 self
.addCleanup(subprocess
.call
, ['umount', '/etc/hostname'])
739 subprocess
.check_call(['systemctl', 'stop', 'systemd-hostnamed.service'])
740 self
.addCleanup(subprocess
.call
, ['systemctl', 'stop', 'systemd-hostnamed.service'])
742 self
.create_iface(dnsmasq_opts
=['--dhcp-host={},192.168.5.210,testgreen'.format(self
.iface_mac
)])
743 self
.do_test(coldplug
=None, extra_opts
='IPv6AcceptRA=False', dhcp_mode
='ipv4')
746 # should have received the fixed IP above
747 out
= subprocess
.check_output(['ip', '-4', 'a', 'show', 'dev', self
.iface
])
748 self
.assertRegex(out
, b
'inet 192.168.5.210/24 .* scope global dynamic')
749 # should have set transient hostname in hostnamed; this is
750 # sometimes a bit lagging (issue #4753), so retry a few times
751 for retry
in range(1, 6):
752 out
= subprocess
.check_output(['hostnamectl'])
753 if b
'testgreen' in out
:
756 sys
.stdout
.write('[retry %i] ' % retry
)
759 self
.fail('Transient hostname not found in hostnamectl:\n{}'.format(out
.decode()))
760 # and also applied to the system
761 self
.assertEqual(socket
.gethostname(), 'testgreen')
762 except AssertionError:
763 self
.show_journal('systemd-networkd.service')
764 self
.show_journal('systemd-hostnamed.service')
765 self
.print_server_log()
768 def test_transient_hostname_with_static(self
):
769 '''transient hostname is not applied if static hostname exists'''
771 orig_hostname
= socket
.gethostname()
772 self
.addCleanup(socket
.sethostname
, orig_hostname
)
774 if not os
.path
.exists('/etc/hostname'):
775 self
.write_config('/etc/hostname', "foobarqux")
777 self
.write_config('/run/hostname.tmp', "foobarqux")
778 subprocess
.check_call(['mount', '--bind', '/run/hostname.tmp', '/etc/hostname'])
779 self
.addCleanup(subprocess
.call
, ['umount', '/etc/hostname'])
781 socket
.sethostname("foobarqux");
783 subprocess
.check_call(['systemctl', 'stop', 'systemd-hostnamed.service'])
784 self
.addCleanup(subprocess
.call
, ['systemctl', 'stop', 'systemd-hostnamed.service'])
786 self
.create_iface(dnsmasq_opts
=['--dhcp-host={},192.168.5.210,testgreen'.format(self
.iface_mac
)])
787 self
.do_test(coldplug
=None, extra_opts
='IPv6AcceptRA=False', dhcp_mode
='ipv4')
790 # should have received the fixed IP above
791 out
= subprocess
.check_output(['ip', '-4', 'a', 'show', 'dev', self
.iface
])
792 self
.assertRegex(out
, b
'inet 192.168.5.210/24 .* scope global dynamic')
793 # static hostname wins over transient one, thus *not* applied
794 self
.assertEqual(socket
.gethostname(), "foobarqux")
795 except AssertionError:
796 self
.show_journal('systemd-networkd.service')
797 self
.show_journal('systemd-hostnamed.service')
798 self
.print_server_log()
802 class NetworkdClientTest(ClientTestBase
, unittest
.TestCase
):
803 '''Test networkd client against networkd server'''
809 def create_iface(self
, ipv6
=False, dhcpserver_opts
=None):
810 '''Create test interface with DHCP server behind it'''
812 # run "router-side" networkd in own mount namespace to shield it from
813 # "client-side" configuration and networkd
814 (fd
, script
) = tempfile
.mkstemp(prefix
='networkd-router.sh')
815 self
.addCleanup(os
.remove
, script
)
816 with os
.fdopen(fd
, 'w+') as f
:
820 mkdir -p /run/systemd/network
821 mkdir -p /run/systemd/netif
822 mount -t tmpfs none /run/systemd/network
823 mount -t tmpfs none /run/systemd/netif
824 [ ! -e /run/dbus ] || mount -t tmpfs none /run/dbus
825 # create router/client veth pair
826 cat << EOF > /run/systemd/network/test.netdev
835 cat << EOF > /run/systemd/network/test.network
840 Address=192.168.5.1/24
851 # run networkd as in systemd-networkd.service
852 exec $(systemctl cat systemd-networkd.service | sed -n '/^ExecStart=/ { s/^.*=//; s/^[@+-]//; s/^!*//; p}')
853 ''' % {'ifr': self
.if_router
, 'ifc': self
.iface
, 'addr6': ipv6
and 'Address=2600::1/64' or '',
854 'dhopts': dhcpserver_opts
or ''})
858 subprocess
.check_call(['systemd-run', '--unit=networkd-test-router.service',
859 '-p', 'InaccessibleDirectories=-/etc/systemd/network',
860 '-p', 'InaccessibleDirectories=-/run/systemd/network',
861 '-p', 'InaccessibleDirectories=-/run/systemd/netif',
862 '--service-type=notify', script
])
864 # wait until devices got created
866 out
= subprocess
.check_output(['ip', 'a', 'show', 'dev', self
.if_router
])
867 if b
'state UP' in out
and b
'scope global' in out
:
871 def shutdown_iface(self
):
872 '''Remove test interface and stop DHCP server'''
875 subprocess
.check_call(['systemctl', 'stop', 'networkd-test-router.service'])
876 # ensure failed transient unit does not stay around
877 subprocess
.call(['systemctl', 'reset-failed', 'networkd-test-router.service'])
878 subprocess
.call(['ip', 'link', 'del', 'dev', self
.if_router
])
879 self
.if_router
= None
881 def print_server_log(self
):
882 '''Print DHCP server log for debugging failures'''
884 self
.show_journal('networkd-test-router.service')
886 @unittest.skip('networkd does not have DHCPv6 server support')
887 def test_hotplug_dhcp_ip6(self
):
890 @unittest.skip('networkd does not have DHCPv6 server support')
891 def test_coldplug_dhcp_ip6(self
):
894 def test_search_domains(self
):
896 # we don't use this interface for this test
897 self
.if_router
= None
899 self
.write_network('test.netdev', '''\
903 MACAddress=12:34:56:78:9a:bc''')
904 self
.write_network('test.network', '''\
908 Address=192.168.42.100/24
910 Domains= one two three four five six seven eight nine ten''')
912 self
.start_unit('systemd-networkd')
914 for timeout
in range(50):
915 with
open(RESOLV_CONF
) as f
:
917 if ' one' in contents
:
920 self
.assertRegex(contents
, 'search .*one two three four')
921 self
.assertNotIn('seven\n', contents
)
922 self
.assertIn('# Too many search domains configured, remaining ones ignored.\n', contents
)
924 def test_search_domains_too_long(self
):
926 # we don't use this interface for this test
927 self
.if_router
= None
929 name_prefix
= 'a' * 60
931 self
.write_network('test.netdev', '''\
935 MACAddress=12:34:56:78:9a:bc''')
936 self
.write_network('test.network', '''\
940 Address=192.168.42.100/24
942 Domains={p}0 {p}1 {p}2 {p}3 {p}4'''.format(p
=name_prefix
))
944 self
.start_unit('systemd-networkd')
946 for timeout
in range(50):
947 with
open(RESOLV_CONF
) as f
:
949 if ' one' in contents
:
952 self
.assertRegex(contents
, 'search .*{p}0 {p}1 {p}2'.format(p
=name_prefix
))
953 self
.assertIn('# Total length of all search domains is too long, remaining ones ignored.', contents
)
955 def test_dropin(self
):
956 # we don't use this interface for this test
957 self
.if_router
= None
959 self
.write_network('test.netdev', '''\
963 MACAddress=12:34:56:78:9a:bc''')
964 self
.write_network('test.network', '''\
968 Address=192.168.42.100/24
970 self
.write_network_dropin('test.network', 'dns', '''\
974 self
.start_unit('systemd-resolved')
975 self
.start_unit('systemd-networkd')
977 for timeout
in range(50):
978 with
open(RESOLV_CONF
) as f
:
980 if ' 127.0.0.1' in contents
and '192.168.42.1' in contents
:
983 self
.assertIn('nameserver 192.168.42.1\n', contents
)
984 self
.assertIn('nameserver 127.0.0.1\n', contents
)
986 def test_dhcp_timezone(self
):
987 '''networkd sets time zone from DHCP'''
990 out
= subprocess
.check_output(['busctl', 'get-property', 'org.freedesktop.timedate1',
991 '/org/freedesktop/timedate1', 'org.freedesktop.timedate1', 'Timezone'])
992 assert out
.startswith(b
's "')
994 assert out
.endswith(b
'"')
995 return out
[3:-1].decode()
997 orig_timezone
= get_tz()
998 self
.addCleanup(subprocess
.call
, ['timedatectl', 'set-timezone', orig_timezone
])
1000 self
.create_iface(dhcpserver_opts
='EmitTimezone=yes\nTimezone=Pacific/Honolulu')
1001 self
.do_test(coldplug
=None, extra_opts
='IPv6AcceptRA=false\n[DHCP]\nUseTimezone=true', dhcp_mode
='ipv4')
1003 # should have applied the received timezone
1005 self
.assertEqual(get_tz(), 'Pacific/Honolulu')
1006 except AssertionError:
1007 self
.show_journal('systemd-networkd.service')
1008 self
.show_journal('systemd-hostnamed.service')
1012 class MatchClientTest(unittest
.TestCase
, NetworkdTestingUtilities
):
1013 """Test [Match] sections in .network files.
1015 Be aware that matching the test host's interfaces will wipe their
1016 configuration, so as a precaution, all network files should have a
1017 restrictive [Match] section to only ever interfere with the
1018 temporary veth interfaces created here.
1022 """Stop networkd."""
1023 subprocess
.call(['systemctl', 'stop', 'systemd-networkd'])
1025 def test_basic_matching(self
):
1026 """Verify the Name= line works throughout this class."""
1027 self
.add_veth_pair('test_if1', 'fake_if2')
1028 self
.write_network('test.network', "[Match]\nName=test_*\n[Network]")
1029 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
1030 self
.assert_link_states(test_if1
='managed', fake_if2
='unmanaged')
1032 def test_inverted_matching(self
):
1033 """Verify that a '!'-prefixed value inverts the match."""
1034 # Use a MAC address as the interfaces' common matching attribute
1035 # to avoid depending on udev, to support testing in containers.
1036 mac
= '00:01:02:03:98:99'
1037 self
.add_veth_pair('test_veth', 'test_peer',
1038 ['addr', mac
], ['addr', mac
])
1039 self
.write_network('no-veth.network', """\
1042 Name=!nonexistent *peer*
1043 [Network]""".format(mac
))
1044 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
1045 self
.assert_link_states(test_veth
='managed', test_peer
='unmanaged')
1048 class UnmanagedClientTest(unittest
.TestCase
, NetworkdTestingUtilities
):
1049 """Test if networkd manages the correct interfaces."""
1052 """Write .network files to match the named veth devices."""
1053 # Define the veth+peer pairs to be created.
1054 # Their pairing doesn't actually matter, only their names do.
1060 # Define the contents of .network files to be read in order.
1062 "[Match]\nName=m1def\n",
1063 "[Match]\nName=m1unm\n[Link]\nUnmanaged=yes\n",
1064 "[Match]\nName=m1*\n[Link]\nUnmanaged=no\n",
1067 # Write out the .network files to be cleaned up automatically.
1068 for i
, config
in enumerate(self
.configs
):
1069 self
.write_network("%02d-test.network" % i
, config
)
1072 """Stop networkd."""
1073 subprocess
.call(['systemctl', 'stop', 'systemd-networkd'])
1075 def create_iface(self
):
1076 """Create temporary veth pairs for interface matching."""
1077 for veth
, peer
in self
.veths
.items():
1078 self
.add_veth_pair(veth
, peer
)
1080 def test_unmanaged_setting(self
):
1081 """Verify link states with Unmanaged= settings, hot-plug."""
1082 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
1084 self
.assert_link_states(m1def
='managed',
1089 def test_unmanaged_setting_coldplug(self
):
1090 """Verify link states with Unmanaged= settings, cold-plug."""
1092 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
1093 self
.assert_link_states(m1def
='managed',
1098 def test_catchall_config(self
):
1099 """Verify link states with a catch-all config, hot-plug."""
1100 # Don't actually catch ALL interfaces. It messes up the host.
1101 self
.write_network('all.network', "[Match]\nName=m[01]???\n")
1102 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
1104 self
.assert_link_states(m1def
='managed',
1109 def test_catchall_config_coldplug(self
):
1110 """Verify link states with a catch-all config, cold-plug."""
1111 # Don't actually catch ALL interfaces. It messes up the host.
1112 self
.write_network('all.network', "[Match]\nName=m[01]???\n")
1114 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
1115 self
.assert_link_states(m1def
='managed',
1121 if __name__
== '__main__':
1122 unittest
.main(testRunner
=unittest
.TextTestRunner(stream
=sys
.stdout
,