]>
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
33 NETWORK_UNITDIR
= '/run/systemd/network'
35 NETWORKD_WAIT_ONLINE
= shutil
.which('systemd-networkd-wait-online',
36 path
='/usr/lib/systemd:/lib/systemd')
38 RESOLV_CONF
= '/run/systemd/resolve/resolv.conf'
48 """Initialize the environment, and perform sanity checks on it."""
49 if NETWORKD_WAIT_ONLINE
is None:
50 raise OSError(errno
.ENOENT
, 'systemd-networkd-wait-online not found')
52 # Do not run any tests if the system is using networkd already and it's not virtualized
53 if (subprocess
.call(['systemctl', 'is-active', '--quiet', 'systemd-networkd.service']) == 0 and
54 subprocess
.call(['systemd-detect-virt', '--quiet']) != 0):
55 raise unittest
.SkipTest('not virtualized and networkd is already active')
57 # Ensure we don't mess with an existing networkd config
58 for u
in ['systemd-networkd.socket', 'systemd-networkd', 'systemd-resolved']:
59 if subprocess
.call(['systemctl', 'is-active', '--quiet', u
]) == 0:
60 subprocess
.call(['systemctl', 'stop', u
])
61 running_units
.append(u
)
63 stopped_units
.append(u
)
65 # create static systemd-network user for networkd-test-router.service (it
66 # needs to do some stuff as root and can't start as user; but networkd
67 # still insists on the user)
68 subprocess
.call(['adduser', '--system', '--no-create-home', 'systemd-network'])
70 for d
in ['/etc/systemd/network', '/run/systemd/network',
71 '/run/systemd/netif', '/run/systemd/resolve']:
73 subprocess
.check_call(["mount", "-t", "tmpfs", "none", d
])
75 if os
.path
.isdir('/run/systemd/resolve'):
76 os
.chmod('/run/systemd/resolve', 0o755)
77 shutil
.chown('/run/systemd/resolve', 'systemd-resolve', 'systemd-resolve')
78 if os
.path
.isdir('/run/systemd/netif'):
79 os
.chmod('/run/systemd/netif', 0o755)
80 shutil
.chown('/run/systemd/netif', 'systemd-network', 'systemd-network')
82 # Avoid "Failed to open /dev/tty" errors in containers.
83 os
.environ
['SYSTEMD_LOG_TARGET'] = 'journal'
85 # Ensure the unit directory exists so tests can dump files into it.
86 os
.makedirs(NETWORK_UNITDIR
, exist_ok
=True)
92 subprocess
.check_call(["umount", d
])
93 for u
in stopped_units
:
94 subprocess
.call(["systemctl", "stop", u
])
95 for u
in running_units
:
96 subprocess
.call(["systemctl", "restart", u
])
99 class NetworkdTestingUtilities
:
100 """Provide a set of utility functions to facilitate networkd tests.
102 This class must be inherited along with unittest.TestCase to define
103 some required methods.
106 def add_veth_pair(self
, veth
, peer
, veth_options
=(), peer_options
=()):
107 """Add a veth interface pair, and queue them to be removed."""
108 subprocess
.check_call(['ip', 'link', 'add', 'name', veth
] +
110 ['type', 'veth', 'peer', 'name', peer
] +
112 self
.addCleanup(subprocess
.call
, ['ip', 'link', 'del', 'dev', peer
])
114 def write_config(self
, path
, contents
):
115 """"Write a configuration file, and queue it to be removed."""
117 with
open(path
, 'w') as f
:
120 self
.addCleanup(os
.remove
, path
)
122 def write_network(self
, unit_name
, contents
):
123 """Write a network unit file, and queue it to be removed."""
124 self
.write_config(os
.path
.join(NETWORK_UNITDIR
, unit_name
), contents
)
126 def write_network_dropin(self
, unit_name
, dropin_name
, contents
):
127 """Write a network unit drop-in, and queue it to be removed."""
128 dropin_dir
= os
.path
.join(NETWORK_UNITDIR
, "{}.d".format(unit_name
))
129 dropin_path
= os
.path
.join(dropin_dir
, "{}.conf".format(dropin_name
))
131 os
.makedirs(dropin_dir
, exist_ok
=True)
132 self
.addCleanup(os
.rmdir
, dropin_dir
)
133 with
open(dropin_path
, 'w') as dropin
:
134 dropin
.write(contents
)
135 self
.addCleanup(os
.remove
, dropin_path
)
137 def read_attr(self
, link
, attribute
):
138 """Read a link attributed from the sysfs."""
139 # Note we we don't want to check if interface `link' is managed, we
140 # want to evaluate link variable and pass the value of the link to
141 # assert_link_states e.g. eth0=managed.
142 self
.assert_link_states(**{link
:'managed'})
143 with
open(os
.path
.join('/sys/class/net', link
, attribute
)) as f
:
144 return f
.readline().strip()
146 def assert_link_states(self
, **kwargs
):
147 """Match networkctl link states to the given ones.
149 Each keyword argument should be the name of a network interface
150 with its expected value of the "SETUP" column in output from
151 networkctl. The interfaces have five seconds to come online
152 before the check is performed. Every specified interface must
153 be present in the output, and any other interfaces found in the
156 A special interface state "managed" is supported, which matches
157 any value in the "SETUP" column other than "unmanaged".
161 interfaces
= set(kwargs
)
163 # Wait for the requested interfaces, but don't fail for them.
164 subprocess
.call([NETWORKD_WAIT_ONLINE
, '--timeout=5'] +
165 ['--interface={}'.format(iface
) for iface
in kwargs
])
167 # Validate each link state found in the networkctl output.
168 out
= subprocess
.check_output(['networkctl', '--no-legend']).rstrip()
169 for line
in out
.decode('utf-8').split('\n'):
170 fields
= line
.split()
171 if len(fields
) >= 5 and fields
[1] in kwargs
:
173 expected
= kwargs
[iface
]
175 if (actual
!= expected
and
176 not (expected
== 'managed' and actual
!= 'unmanaged')):
177 self
.fail("Link {} expects state {}, found {}".format(iface
, expected
, actual
))
178 interfaces
.remove(iface
)
180 # Ensure that all requested interfaces have been covered.
182 self
.fail("Missing links in status output: {}".format(interfaces
))
185 class BridgeTest(NetworkdTestingUtilities
, unittest
.TestCase
):
186 """Provide common methods for testing networkd against servers."""
189 self
.write_network('port1.netdev', '''\
193 MACAddress=12:34:56:78:9a:bc''')
194 self
.write_network('port2.netdev', '''\
198 MACAddress=12:34:56:78:9a:bd''')
199 self
.write_network('mybridge.netdev', '''\
203 self
.write_network('port1.network', '''\
208 self
.write_network('port2.network', '''\
213 self
.write_network('mybridge.network', '''\
218 Address=192.168.250.33/24
219 Gateway=192.168.250.1''')
220 subprocess
.call(['systemctl', 'reset-failed', 'systemd-networkd', 'systemd-resolved'])
221 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
224 subprocess
.check_call(['systemctl', 'stop', 'systemd-networkd'])
225 subprocess
.check_call(['ip', 'link', 'del', 'mybridge'])
226 subprocess
.check_call(['ip', 'link', 'del', 'port1'])
227 subprocess
.check_call(['ip', 'link', 'del', 'port2'])
229 def test_bridge_init(self
):
230 self
.assert_link_states(
235 def test_bridge_port_priority(self
):
236 self
.assertEqual(self
.read_attr('port1', 'brport/priority'), '32')
237 self
.write_network_dropin('port1.network', 'priority', '''\
241 subprocess
.check_call(['systemctl', 'restart', 'systemd-networkd'])
242 self
.assertEqual(self
.read_attr('port1', 'brport/priority'), '28')
244 def test_bridge_port_priority_set_zero(self
):
245 """It should be possible to set the bridge port priority to 0"""
246 self
.assertEqual(self
.read_attr('port2', 'brport/priority'), '32')
247 self
.write_network_dropin('port2.network', 'priority', '''\
251 subprocess
.check_call(['systemctl', 'restart', 'systemd-networkd'])
252 self
.assertEqual(self
.read_attr('port2', 'brport/priority'), '0')
254 def test_bridge_port_property(self
):
255 """Test the "[Bridge]" section keys"""
256 self
.assertEqual(self
.read_attr('port2', 'brport/priority'), '32')
257 self
.write_network_dropin('port2.network', 'property', '''\
263 AllowPortToBeRoot=true
267 subprocess
.check_call(['systemctl', 'restart', 'systemd-networkd'])
269 self
.assertEqual(self
.read_attr('port2', 'brport/priority'), '23')
270 self
.assertEqual(self
.read_attr('port2', 'brport/hairpin_mode'), '1')
271 self
.assertEqual(self
.read_attr('port2', 'brport/path_cost'), '555')
272 self
.assertEqual(self
.read_attr('port2', 'brport/multicast_fast_leave'), '1')
273 self
.assertEqual(self
.read_attr('port2', 'brport/unicast_flood'), '1')
274 self
.assertEqual(self
.read_attr('port2', 'brport/bpdu_guard'), '1')
275 self
.assertEqual(self
.read_attr('port2', 'brport/root_block'), '1')
277 class ClientTestBase(NetworkdTestingUtilities
):
278 """Provide common methods for testing networkd against servers."""
281 def setUpClass(klass
):
282 klass
.orig_log_level
= subprocess
.check_output(
283 ['systemctl', 'show', '--value', '--property', 'LogLevel'],
284 universal_newlines
=True).strip()
285 subprocess
.check_call(['systemd-analyze', 'set-log-level', 'debug'])
288 def tearDownClass(klass
):
289 subprocess
.check_call(['systemd-analyze', 'set-log-level', klass
.orig_log_level
])
292 self
.iface
= 'test_eth42'
293 self
.if_router
= 'router_eth42'
294 self
.workdir_obj
= tempfile
.TemporaryDirectory()
295 self
.workdir
= self
.workdir_obj
.name
296 self
.config
= 'test_eth42.network'
298 # get current journal cursor
299 subprocess
.check_output(['journalctl', '--sync'])
300 out
= subprocess
.check_output(['journalctl', '-b', '--quiet',
301 '--no-pager', '-n0', '--show-cursor'],
302 universal_newlines
=True)
303 self
.assertTrue(out
.startswith('-- cursor:'))
304 self
.journal_cursor
= out
.split()[-1]
306 subprocess
.call(['systemctl', 'reset-failed', 'systemd-networkd', 'systemd-resolved'])
309 self
.shutdown_iface()
310 subprocess
.call(['systemctl', 'stop', 'systemd-networkd'])
311 subprocess
.call(['ip', 'link', 'del', 'dummy0'],
312 stderr
=subprocess
.DEVNULL
)
314 def show_journal(self
, unit
):
315 '''Show journal of given unit since start of the test'''
317 print('---- {} ----'.format(unit
))
318 subprocess
.check_output(['journalctl', '--sync'])
320 subprocess
.call(['journalctl', '-b', '--no-pager', '--quiet',
321 '--cursor', self
.journal_cursor
, '-u', unit
])
323 def create_iface(self
, ipv6
=False):
324 '''Create test interface with DHCP server behind it'''
326 raise NotImplementedError('must be implemented by a subclass')
328 def shutdown_iface(self
):
329 '''Remove test interface and stop DHCP server'''
331 raise NotImplementedError('must be implemented by a subclass')
333 def print_server_log(self
):
334 '''Print DHCP server log for debugging failures'''
336 raise NotImplementedError('must be implemented by a subclass')
338 def do_test(self
, coldplug
=True, ipv6
=False, extra_opts
='',
339 online_timeout
=10, dhcp_mode
='yes'):
341 subprocess
.check_call(['systemctl', 'start', 'systemd-resolved'])
342 except subprocess
.CalledProcessError
:
343 self
.show_journal('systemd-resolved.service')
345 self
.write_network(self
.config
, '''\
350 {}'''.format(self
.iface
, dhcp_mode
, extra_opts
))
353 # create interface first, then start networkd
354 self
.create_iface(ipv6
=ipv6
)
355 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
356 elif coldplug
is not None:
357 # start networkd first, then create interface
358 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
359 self
.create_iface(ipv6
=ipv6
)
361 # "None" means test sets up interface by itself
362 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
365 subprocess
.check_call([NETWORKD_WAIT_ONLINE
, '--interface',
366 self
.iface
, '--timeout=%i' % online_timeout
])
369 # check iface state and IP 6 address; FIXME: we need to wait a bit
370 # longer, as the iface is "configured" already with IPv4 *or*
371 # IPv6, but we want to wait for both
373 out
= subprocess
.check_output(['ip', 'a', 'show', 'dev', self
.iface
])
374 if b
'state UP' in out
and b
'inet6 2600' in out
and b
'inet 192.168' in out
:
378 self
.fail('timed out waiting for IPv6 configuration')
380 self
.assertRegex(out
, b
'inet6 2600::.* scope global .*dynamic')
381 self
.assertRegex(out
, b
'inet6 fe80::.* scope link')
383 # should have link-local address on IPv6 only
384 out
= subprocess
.check_output(['ip', '-6', 'a', 'show', 'dev', self
.iface
])
385 self
.assertRegex(out
, br
'inet6 fe80::.* scope link')
386 self
.assertNotIn(b
'scope global', out
)
388 # should have IPv4 address
389 out
= subprocess
.check_output(['ip', '-4', 'a', 'show', 'dev', self
.iface
])
390 self
.assertIn(b
'state UP', out
)
391 self
.assertRegex(out
, br
'inet 192.168.5.\d+/.* scope global dynamic')
393 # check networkctl state
394 out
= subprocess
.check_output(['networkctl'])
395 self
.assertRegex(out
, (r
'{}\s+ether\s+[a-z-]+\s+unmanaged'.format(self
.if_router
)).encode())
396 self
.assertRegex(out
, (r
'{}\s+ether\s+routable\s+configured'.format(self
.iface
)).encode())
398 out
= subprocess
.check_output(['networkctl', 'status', self
.iface
])
399 self
.assertRegex(out
, br
'Type:\s+ether')
400 self
.assertRegex(out
, br
'State:\s+routable.*configured')
401 self
.assertRegex(out
, br
'Address:\s+192.168.5.\d+')
403 self
.assertRegex(out
, br
'2600::')
405 self
.assertNotIn(br
'2600::', out
)
406 self
.assertRegex(out
, br
'fe80::')
407 self
.assertRegex(out
, br
'Gateway:\s+192.168.5.1')
408 self
.assertRegex(out
, br
'DNS:\s+192.168.5.1')
409 except (AssertionError, subprocess
.CalledProcessError
):
410 # show networkd status, journal, and DHCP server log on failure
411 with
open(os
.path
.join(NETWORK_UNITDIR
, self
.config
)) as f
:
412 print('\n---- {} ----\n{}'.format(self
.config
, f
.read()))
413 print('---- interface status ----')
415 subprocess
.call(['ip', 'a', 'show', 'dev', self
.iface
])
416 print('---- networkctl status {} ----'.format(self
.iface
))
418 subprocess
.call(['networkctl', 'status', self
.iface
])
419 self
.show_journal('systemd-networkd.service')
420 self
.print_server_log()
423 for timeout
in range(50):
424 with
open(RESOLV_CONF
) as f
:
426 if 'nameserver 192.168.5.1\n' in contents
:
430 self
.fail('nameserver 192.168.5.1 not found in ' + RESOLV_CONF
)
432 if coldplug
is False:
433 # check post-down.d hook
434 self
.shutdown_iface()
436 def test_coldplug_dhcp_yes_ip4(self
):
437 # we have a 12s timeout on RA, so we need to wait longer
438 self
.do_test(coldplug
=True, ipv6
=False, online_timeout
=15)
440 def test_coldplug_dhcp_yes_ip4_no_ra(self
):
441 # with disabling RA explicitly things should be fast
442 self
.do_test(coldplug
=True, ipv6
=False,
443 extra_opts
='IPv6AcceptRA=False')
445 def test_coldplug_dhcp_ip4_only(self
):
446 # we have a 12s timeout on RA, so we need to wait longer
447 self
.do_test(coldplug
=True, ipv6
=False, dhcp_mode
='ipv4',
450 def test_coldplug_dhcp_ip4_only_no_ra(self
):
451 # with disabling RA explicitly things should be fast
452 self
.do_test(coldplug
=True, ipv6
=False, dhcp_mode
='ipv4',
453 extra_opts
='IPv6AcceptRA=False')
455 def test_coldplug_dhcp_ip6(self
):
456 self
.do_test(coldplug
=True, ipv6
=True)
458 def test_hotplug_dhcp_ip4(self
):
459 # With IPv4 only we have a 12s timeout on RA, so we need to wait longer
460 self
.do_test(coldplug
=False, ipv6
=False, online_timeout
=15)
462 def test_hotplug_dhcp_ip6(self
):
463 self
.do_test(coldplug
=False, ipv6
=True)
465 def test_route_only_dns(self
):
466 self
.write_network('myvpn.netdev', '''\
470 MACAddress=12:34:56:78:9a:bc''')
471 self
.write_network('myvpn.network', '''\
475 Address=192.168.42.100
477 Domains= ~company''')
479 self
.do_test(coldplug
=True, ipv6
=False,
480 extra_opts
='IPv6AcceptRouterAdvertisements=False')
482 with
open(RESOLV_CONF
) as f
:
484 # ~company is not a search domain, only a routing domain
485 self
.assertNotRegex(contents
, 'search.*company')
486 # our global server should appear
487 self
.assertIn('nameserver 192.168.5.1\n', contents
)
488 # should not have domain-restricted server as global server
489 self
.assertNotIn('nameserver 192.168.42.1\n', contents
)
491 def test_route_only_dns_all_domains(self
):
492 self
.write_network('myvpn.netdev', '''[NetDev]
495 MACAddress=12:34:56:78:9a:bc''')
496 self
.write_network('myvpn.network', '''[Match]
499 Address=192.168.42.100
501 Domains= ~company ~.''')
503 self
.do_test(coldplug
=True, ipv6
=False,
504 extra_opts
='IPv6AcceptRouterAdvertisements=False')
506 with
open(RESOLV_CONF
) as f
:
509 # ~company is not a search domain, only a routing domain
510 self
.assertNotRegex(contents
, 'search.*company')
512 # our global server should appear
513 self
.assertIn('nameserver 192.168.5.1\n', contents
)
514 # should have company server as global server due to ~.
515 self
.assertIn('nameserver 192.168.42.1\n', contents
)
518 @unittest.skipUnless(HAVE_DNSMASQ
, 'dnsmasq not installed')
519 class DnsmasqClientTest(ClientTestBase
, unittest
.TestCase
):
520 '''Test networkd client against dnsmasq'''
525 self
.iface_mac
= 'de:ad:be:ef:47:11'
527 def create_iface(self
, ipv6
=False, dnsmasq_opts
=None):
528 '''Create test interface with DHCP server behind it'''
531 subprocess
.check_call(['ip', 'link', 'add', 'name', self
.iface
,
532 'address', self
.iface_mac
,
533 'type', 'veth', 'peer', 'name', self
.if_router
])
535 # give our router an IP
536 subprocess
.check_call(['ip', 'a', 'flush', 'dev', self
.if_router
])
537 subprocess
.check_call(['ip', 'a', 'add', '192.168.5.1/24', 'dev', self
.if_router
])
539 subprocess
.check_call(['ip', 'a', 'add', '2600::1/64', 'dev', self
.if_router
])
540 subprocess
.check_call(['ip', 'link', 'set', self
.if_router
, 'up'])
543 self
.dnsmasq_log
= os
.path
.join(self
.workdir
, 'dnsmasq.log')
544 lease_file
= os
.path
.join(self
.workdir
, 'dnsmasq.leases')
546 extra_opts
= ['--enable-ra', '--dhcp-range=2600::10,2600::20']
550 extra_opts
+= dnsmasq_opts
551 self
.dnsmasq
= subprocess
.Popen(
552 ['dnsmasq', '--keep-in-foreground', '--log-queries',
553 '--log-facility=' + self
.dnsmasq_log
, '--conf-file=/dev/null',
554 '--dhcp-leasefile=' + lease_file
, '--bind-interfaces',
555 '--interface=' + self
.if_router
, '--except-interface=lo',
556 '--dhcp-range=192.168.5.10,192.168.5.200'] + extra_opts
)
558 def shutdown_iface(self
):
559 '''Remove test interface and stop DHCP server'''
562 subprocess
.check_call(['ip', 'link', 'del', 'dev', self
.if_router
])
563 self
.if_router
= None
569 def print_server_log(self
):
570 '''Print DHCP server log for debugging failures'''
572 with
open(self
.dnsmasq_log
) as f
:
573 sys
.stdout
.write('\n\n---- dnsmasq log ----\n{}\n------\n\n'.format(f
.read()))
575 def test_resolved_domain_restricted_dns(self
):
576 '''resolved: domain-restricted DNS servers'''
578 # create interface for generic connections; this will map all DNS names
580 self
.create_iface(dnsmasq_opts
=['--address=/#/192.168.42.1'])
581 self
.write_network('general.network', '''\
586 IPv6AcceptRA=False'''.format(self
.iface
))
588 # create second device/dnsmasq for a .company/.lab VPN interface
589 # static IPs for simplicity
590 self
.add_veth_pair('testvpnclient', 'testvpnrouter')
591 subprocess
.check_call(['ip', 'a', 'flush', 'dev', 'testvpnrouter'])
592 subprocess
.check_call(['ip', 'a', 'add', '10.241.3.1/24', 'dev', 'testvpnrouter'])
593 subprocess
.check_call(['ip', 'link', 'set', 'testvpnrouter', 'up'])
595 vpn_dnsmasq_log
= os
.path
.join(self
.workdir
, 'dnsmasq-vpn.log')
596 vpn_dnsmasq
= subprocess
.Popen(
597 ['dnsmasq', '--keep-in-foreground', '--log-queries',
598 '--log-facility=' + vpn_dnsmasq_log
, '--conf-file=/dev/null',
599 '--dhcp-leasefile=/dev/null', '--bind-interfaces',
600 '--interface=testvpnrouter', '--except-interface=lo',
601 '--address=/math.lab/10.241.3.3', '--address=/cantina.company/10.241.4.4'])
602 self
.addCleanup(vpn_dnsmasq
.wait
)
603 self
.addCleanup(vpn_dnsmasq
.kill
)
605 self
.write_network('vpn.network', '''\
610 Address=10.241.3.2/24
612 Domains= ~company ~lab''')
614 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
615 subprocess
.check_call([NETWORKD_WAIT_ONLINE
, '--interface', self
.iface
,
616 '--interface=testvpnclient', '--timeout=20'])
618 # ensure we start fresh with every test
619 subprocess
.check_call(['systemctl', 'restart', 'systemd-resolved'])
621 # test vpnclient specific domains; these should *not* be answered by
623 out
= subprocess
.check_output(['resolvectl', 'query', 'math.lab'])
624 self
.assertIn(b
'math.lab: 10.241.3.3', out
)
625 out
= subprocess
.check_output(['resolvectl', 'query', 'kettle.cantina.company'])
626 self
.assertIn(b
'kettle.cantina.company: 10.241.4.4', out
)
628 # test general domains
629 out
= subprocess
.check_output(['resolvectl', 'query', 'megasearch.net'])
630 self
.assertIn(b
'megasearch.net: 192.168.42.1', out
)
632 with
open(self
.dnsmasq_log
) as f
:
633 general_log
= f
.read()
634 with
open(vpn_dnsmasq_log
) as f
:
637 # VPN domains should only be sent to VPN DNS
638 self
.assertRegex(vpn_log
, 'query.*math.lab')
639 self
.assertRegex(vpn_log
, 'query.*cantina.company')
640 self
.assertNotIn('.lab', general_log
)
641 self
.assertNotIn('.company', general_log
)
643 # general domains should not be sent to the VPN DNS
644 self
.assertRegex(general_log
, 'query.*megasearch.net')
645 self
.assertNotIn('megasearch.net', vpn_log
)
647 def test_resolved_etc_hosts(self
):
648 '''resolved queries to /etc/hosts'''
650 # FIXME: -t MX query fails with enabled DNSSEC (even when using
651 # the known negative trust anchor .internal instead of .example)
652 conf
= '/run/systemd/resolved.conf.d/test-disable-dnssec.conf'
653 os
.makedirs(os
.path
.dirname(conf
), exist_ok
=True)
654 with
open(conf
, 'w') as f
:
655 f
.write('[Resolve]\nDNSSEC=no\nLLMNR=no\nMulticastDNS=no\n')
656 self
.addCleanup(os
.remove
, conf
)
658 # create /etc/hosts bind mount which resolves my.example for IPv4
659 hosts
= os
.path
.join(self
.workdir
, 'hosts')
660 with
open(hosts
, 'w') as f
:
661 f
.write('172.16.99.99 my.example\n')
662 subprocess
.check_call(['mount', '--bind', hosts
, '/etc/hosts'])
663 self
.addCleanup(subprocess
.call
, ['umount', '/etc/hosts'])
664 subprocess
.check_call(['systemctl', 'stop', 'systemd-resolved.service'])
666 # note: different IPv4 address here, so that it's easy to tell apart
667 # what resolved the query
668 self
.create_iface(dnsmasq_opts
=['--host-record=my.example,172.16.99.1,2600::99:99',
669 '--host-record=other.example,172.16.0.42,2600::42',
670 '--mx-host=example,mail.example'],
672 self
.do_test(coldplug
=None, ipv6
=True)
675 # family specific queries
676 out
= subprocess
.check_output(['resolvectl', 'query', '-4', 'my.example'])
677 self
.assertIn(b
'my.example: 172.16.99.99', out
)
678 # we don't expect an IPv6 answer; if /etc/hosts has any IP address,
679 # it's considered a sufficient source
680 self
.assertNotEqual(subprocess
.call(['resolvectl', 'query', '-6', 'my.example']), 0)
681 # "any family" query; IPv4 should come from /etc/hosts
682 out
= subprocess
.check_output(['resolvectl', 'query', 'my.example'])
683 self
.assertIn(b
'my.example: 172.16.99.99', out
)
684 # IP → name lookup; again, takes the /etc/hosts one
685 out
= subprocess
.check_output(['resolvectl', 'query', '172.16.99.99'])
686 self
.assertIn(b
'172.16.99.99: my.example', out
)
688 # non-address RRs should fall back to DNS
689 out
= subprocess
.check_output(['resolvectl', 'query', '--type=MX', 'example'])
690 self
.assertIn(b
'example IN MX 1 mail.example', out
)
692 # other domains query DNS
693 out
= subprocess
.check_output(['resolvectl', 'query', 'other.example'])
694 self
.assertIn(b
'172.16.0.42', out
)
695 out
= subprocess
.check_output(['resolvectl', 'query', '172.16.0.42'])
696 self
.assertIn(b
'172.16.0.42: other.example', out
)
697 except (AssertionError, subprocess
.CalledProcessError
):
698 self
.show_journal('systemd-resolved.service')
699 self
.print_server_log()
702 def test_transient_hostname(self
):
703 '''networkd sets transient hostname from DHCP'''
705 orig_hostname
= socket
.gethostname()
706 self
.addCleanup(socket
.sethostname
, orig_hostname
)
707 # temporarily move /etc/hostname away; restart hostnamed to pick it up
708 if os
.path
.exists('/etc/hostname'):
709 subprocess
.check_call(['mount', '--bind', '/dev/null', '/etc/hostname'])
710 self
.addCleanup(subprocess
.call
, ['umount', '/etc/hostname'])
711 subprocess
.check_call(['systemctl', 'stop', 'systemd-hostnamed.service'])
712 self
.addCleanup(subprocess
.call
, ['systemctl', 'stop', 'systemd-hostnamed.service'])
714 self
.create_iface(dnsmasq_opts
=['--dhcp-host={},192.168.5.210,testgreen'.format(self
.iface_mac
)])
715 self
.do_test(coldplug
=None, extra_opts
='IPv6AcceptRA=False', dhcp_mode
='ipv4')
718 # should have received the fixed IP above
719 out
= subprocess
.check_output(['ip', '-4', 'a', 'show', 'dev', self
.iface
])
720 self
.assertRegex(out
, b
'inet 192.168.5.210/24 .* scope global dynamic')
721 # should have set transient hostname in hostnamed; this is
722 # sometimes a bit lagging (issue #4753), so retry a few times
723 for retry
in range(1, 6):
724 out
= subprocess
.check_output(['hostnamectl'])
725 if b
'testgreen' in out
:
728 sys
.stdout
.write('[retry %i] ' % retry
)
731 self
.fail('Transient hostname not found in hostnamectl:\n{}'.format(out
.decode()))
732 # and also applied to the system
733 self
.assertEqual(socket
.gethostname(), 'testgreen')
734 except AssertionError:
735 self
.show_journal('systemd-networkd.service')
736 self
.show_journal('systemd-hostnamed.service')
737 self
.print_server_log()
740 def test_transient_hostname_with_static(self
):
741 '''transient hostname is not applied if static hostname exists'''
743 orig_hostname
= socket
.gethostname()
744 self
.addCleanup(socket
.sethostname
, orig_hostname
)
746 if not os
.path
.exists('/etc/hostname'):
747 self
.write_config('/etc/hostname', "foobarqux")
749 self
.write_config('/run/hostname.tmp', "foobarqux")
750 subprocess
.check_call(['mount', '--bind', '/run/hostname.tmp', '/etc/hostname'])
751 self
.addCleanup(subprocess
.call
, ['umount', '/etc/hostname'])
753 socket
.sethostname("foobarqux");
755 subprocess
.check_call(['systemctl', 'stop', 'systemd-hostnamed.service'])
756 self
.addCleanup(subprocess
.call
, ['systemctl', 'stop', 'systemd-hostnamed.service'])
758 self
.create_iface(dnsmasq_opts
=['--dhcp-host={},192.168.5.210,testgreen'.format(self
.iface_mac
)])
759 self
.do_test(coldplug
=None, extra_opts
='IPv6AcceptRA=False', dhcp_mode
='ipv4')
762 # should have received the fixed IP above
763 out
= subprocess
.check_output(['ip', '-4', 'a', 'show', 'dev', self
.iface
])
764 self
.assertRegex(out
, b
'inet 192.168.5.210/24 .* scope global dynamic')
765 # static hostname wins over transient one, thus *not* applied
766 self
.assertEqual(socket
.gethostname(), "foobarqux")
767 except AssertionError:
768 self
.show_journal('systemd-networkd.service')
769 self
.show_journal('systemd-hostnamed.service')
770 self
.print_server_log()
774 class NetworkdClientTest(ClientTestBase
, unittest
.TestCase
):
775 '''Test networkd client against networkd server'''
781 def create_iface(self
, ipv6
=False, dhcpserver_opts
=None):
782 '''Create test interface with DHCP server behind it'''
784 # run "router-side" networkd in own mount namespace to shield it from
785 # "client-side" configuration and networkd
786 (fd
, script
) = tempfile
.mkstemp(prefix
='networkd-router.sh')
787 self
.addCleanup(os
.remove
, script
)
788 with os
.fdopen(fd
, 'w+') as f
:
792 mkdir -p /run/systemd/network
793 mkdir -p /run/systemd/netif
794 mount -t tmpfs none /run/systemd/network
795 mount -t tmpfs none /run/systemd/netif
796 [ ! -e /run/dbus ] || mount -t tmpfs none /run/dbus
797 # create router/client veth pair
798 cat << EOF > /run/systemd/network/test.netdev
807 cat << EOF > /run/systemd/network/test.network
812 Address=192.168.5.1/24
823 # run networkd as in systemd-networkd.service
824 exec $(systemctl cat systemd-networkd.service | sed -n '/^ExecStart=/ { s/^.*=//; s/^[@+-]//; s/^!*//; p}')
825 ''' % {'ifr': self
.if_router
, 'ifc': self
.iface
, 'addr6': ipv6
and 'Address=2600::1/64' or '',
826 'dhopts': dhcpserver_opts
or ''})
830 subprocess
.check_call(['systemd-run', '--unit=networkd-test-router.service',
831 '-p', 'InaccessibleDirectories=-/etc/systemd/network',
832 '-p', 'InaccessibleDirectories=-/run/systemd/network',
833 '-p', 'InaccessibleDirectories=-/run/systemd/netif',
834 '--service-type=notify', script
])
836 # wait until devices got created
838 out
= subprocess
.check_output(['ip', 'a', 'show', 'dev', self
.if_router
])
839 if b
'state UP' in out
and b
'scope global' in out
:
843 def shutdown_iface(self
):
844 '''Remove test interface and stop DHCP server'''
847 subprocess
.check_call(['systemctl', 'stop', 'networkd-test-router.service'])
848 # ensure failed transient unit does not stay around
849 subprocess
.call(['systemctl', 'reset-failed', 'networkd-test-router.service'])
850 subprocess
.call(['ip', 'link', 'del', 'dev', self
.if_router
])
851 self
.if_router
= None
853 def print_server_log(self
):
854 '''Print DHCP server log for debugging failures'''
856 self
.show_journal('networkd-test-router.service')
858 @unittest.skip('networkd does not have DHCPv6 server support')
859 def test_hotplug_dhcp_ip6(self
):
862 @unittest.skip('networkd does not have DHCPv6 server support')
863 def test_coldplug_dhcp_ip6(self
):
866 def test_search_domains(self
):
868 # we don't use this interface for this test
869 self
.if_router
= None
871 self
.write_network('test.netdev', '''\
875 MACAddress=12:34:56:78:9a:bc''')
876 self
.write_network('test.network', '''\
880 Address=192.168.42.100
882 Domains= one two three four five six seven eight nine ten''')
884 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
886 for timeout
in range(50):
887 with
open(RESOLV_CONF
) as f
:
889 if ' one' in contents
:
892 self
.assertRegex(contents
, 'search .*one two three four')
893 self
.assertNotIn('seven\n', contents
)
894 self
.assertIn('# Too many search domains configured, remaining ones ignored.\n', contents
)
896 def test_search_domains_too_long(self
):
898 # we don't use this interface for this test
899 self
.if_router
= None
901 name_prefix
= 'a' * 60
903 self
.write_network('test.netdev', '''\
907 MACAddress=12:34:56:78:9a:bc''')
908 self
.write_network('test.network', '''\
912 Address=192.168.42.100
914 Domains={p}0 {p}1 {p}2 {p}3 {p}4'''.format(p
=name_prefix
))
916 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
918 for timeout
in range(50):
919 with
open(RESOLV_CONF
) as f
:
921 if ' one' in contents
:
924 self
.assertRegex(contents
, 'search .*{p}0 {p}1 {p}2'.format(p
=name_prefix
))
925 self
.assertIn('# Total length of all search domains is too long, remaining ones ignored.', contents
)
927 def test_dropin(self
):
928 # we don't use this interface for this test
929 self
.if_router
= None
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
942 self
.write_network_dropin('test.network', 'dns', '''\
946 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
948 for timeout
in range(50):
949 with
open(RESOLV_CONF
) as f
:
951 if ' 127.0.0.1' in contents
:
954 self
.assertIn('nameserver 192.168.42.1\n', contents
)
955 self
.assertIn('nameserver 127.0.0.1\n', contents
)
957 def test_dhcp_timezone(self
):
958 '''networkd sets time zone from DHCP'''
961 out
= subprocess
.check_output(['busctl', 'get-property', 'org.freedesktop.timedate1',
962 '/org/freedesktop/timedate1', 'org.freedesktop.timedate1', 'Timezone'])
963 assert out
.startswith(b
's "')
965 assert out
.endswith(b
'"')
966 return out
[3:-1].decode()
968 orig_timezone
= get_tz()
969 self
.addCleanup(subprocess
.call
, ['timedatectl', 'set-timezone', orig_timezone
])
971 self
.create_iface(dhcpserver_opts
='EmitTimezone=yes\nTimezone=Pacific/Honolulu')
972 self
.do_test(coldplug
=None, extra_opts
='IPv6AcceptRA=false\n[DHCP]\nUseTimezone=true', dhcp_mode
='ipv4')
974 # should have applied the received timezone
976 self
.assertEqual(get_tz(), 'Pacific/Honolulu')
977 except AssertionError:
978 self
.show_journal('systemd-networkd.service')
979 self
.show_journal('systemd-hostnamed.service')
983 class MatchClientTest(unittest
.TestCase
, NetworkdTestingUtilities
):
984 """Test [Match] sections in .network files.
986 Be aware that matching the test host's interfaces will wipe their
987 configuration, so as a precaution, all network files should have a
988 restrictive [Match] section to only ever interfere with the
989 temporary veth interfaces created here.
994 subprocess
.call(['systemctl', 'stop', 'systemd-networkd'])
996 def test_basic_matching(self
):
997 """Verify the Name= line works throughout this class."""
998 self
.add_veth_pair('test_if1', 'fake_if2')
999 self
.write_network('test.network', "[Match]\nName=test_*\n[Network]")
1000 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
1001 self
.assert_link_states(test_if1
='managed', fake_if2
='unmanaged')
1003 def test_inverted_matching(self
):
1004 """Verify that a '!'-prefixed value inverts the match."""
1005 # Use a MAC address as the interfaces' common matching attribute
1006 # to avoid depending on udev, to support testing in containers.
1007 mac
= '00:01:02:03:98:99'
1008 self
.add_veth_pair('test_veth', 'test_peer',
1009 ['addr', mac
], ['addr', mac
])
1010 self
.write_network('no-veth.network', """\
1013 Name=!nonexistent *peer*
1014 [Network]""".format(mac
))
1015 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
1016 self
.assert_link_states(test_veth
='managed', test_peer
='unmanaged')
1019 class UnmanagedClientTest(unittest
.TestCase
, NetworkdTestingUtilities
):
1020 """Test if networkd manages the correct interfaces."""
1023 """Write .network files to match the named veth devices."""
1024 # Define the veth+peer pairs to be created.
1025 # Their pairing doesn't actually matter, only their names do.
1031 # Define the contents of .network files to be read in order.
1033 "[Match]\nName=m1def\n",
1034 "[Match]\nName=m1unm\n[Link]\nUnmanaged=yes\n",
1035 "[Match]\nName=m1*\n[Link]\nUnmanaged=no\n",
1038 # Write out the .network files to be cleaned up automatically.
1039 for i
, config
in enumerate(self
.configs
):
1040 self
.write_network("%02d-test.network" % i
, config
)
1043 """Stop networkd."""
1044 subprocess
.call(['systemctl', 'stop', 'systemd-networkd'])
1046 def create_iface(self
):
1047 """Create temporary veth pairs for interface matching."""
1048 for veth
, peer
in self
.veths
.items():
1049 self
.add_veth_pair(veth
, peer
)
1051 def test_unmanaged_setting(self
):
1052 """Verify link states with Unmanaged= settings, hot-plug."""
1053 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
1055 self
.assert_link_states(m1def
='managed',
1060 def test_unmanaged_setting_coldplug(self
):
1061 """Verify link states with Unmanaged= settings, cold-plug."""
1063 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
1064 self
.assert_link_states(m1def
='managed',
1069 def test_catchall_config(self
):
1070 """Verify link states with a catch-all config, hot-plug."""
1071 # Don't actually catch ALL interfaces. It messes up the host.
1072 self
.write_network('all.network', "[Match]\nName=m[01]???\n")
1073 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
1075 self
.assert_link_states(m1def
='managed',
1080 def test_catchall_config_coldplug(self
):
1081 """Verify link states with a catch-all config, cold-plug."""
1082 # Don't actually catch ALL interfaces. It messes up the host.
1083 self
.write_network('all.network', "[Match]\nName=m[01]???\n")
1085 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
1086 self
.assert_link_states(m1def
='managed',
1092 if __name__
== '__main__':
1093 unittest
.main(testRunner
=unittest
.TextTestRunner(stream
=sys
.stdout
,