]>
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')
56 # Ensure we don't mess with an existing networkd config
57 for u
in ['systemd-networkd.socket', 'systemd-networkd', 'systemd-resolved']:
58 if subprocess
.call(['systemctl', 'is-active', '--quiet', u
]) == 0:
59 subprocess
.call(['systemctl', 'stop', u
])
60 running_units
.append(u
)
62 stopped_units
.append(u
)
63 for d
in ['/etc/systemd/network', '/run/systemd/network',
64 '/run/systemd/netif', '/run/systemd/resolve']:
66 subprocess
.check_call(["mount", "-t", "tmpfs", "none", d
])
68 if os
.path
.isdir('/run/systemd/resolve'):
69 os
.chmod('/run/systemd/resolve', 0o755)
70 shutil
.chown('/run/systemd/resolve', 'systemd-resolve', 'systemd-resolve')
72 # Avoid "Failed to open /dev/tty" errors in containers.
73 os
.environ
['SYSTEMD_LOG_TARGET'] = 'journal'
75 # Ensure the unit directory exists so tests can dump files into it.
76 os
.makedirs(NETWORK_UNITDIR
, exist_ok
=True)
78 # create static systemd-network user for networkd-test-router.service (it
79 # needs to do some stuff as root and can't start as user; but networkd
80 # still insists on the user)
81 subprocess
.check_call(['adduser', '--system', '--no-create-home', 'systemd-network'])
87 subprocess
.check_call(["umount", d
])
88 for u
in stopped_units
:
89 subprocess
.call(["systemctl", "stop", u
])
90 for u
in running_units
:
91 subprocess
.call(["systemctl", "restart", u
])
94 class NetworkdTestingUtilities
:
95 """Provide a set of utility functions to facilitate networkd tests.
97 This class must be inherited along with unittest.TestCase to define
98 some required methods.
101 def add_veth_pair(self
, veth
, peer
, veth_options
=(), peer_options
=()):
102 """Add a veth interface pair, and queue them to be removed."""
103 subprocess
.check_call(['ip', 'link', 'add', 'name', veth
] +
105 ['type', 'veth', 'peer', 'name', peer
] +
107 self
.addCleanup(subprocess
.call
, ['ip', 'link', 'del', 'dev', peer
])
109 def write_network(self
, unit_name
, contents
):
110 """Write a network unit file, and queue it to be removed."""
111 unit_path
= os
.path
.join(NETWORK_UNITDIR
, unit_name
)
113 with
open(unit_path
, 'w') as unit
:
115 self
.addCleanup(os
.remove
, unit_path
)
117 def write_network_dropin(self
, unit_name
, dropin_name
, contents
):
118 """Write a network unit drop-in, and queue it to be removed."""
119 dropin_dir
= os
.path
.join(NETWORK_UNITDIR
, "{}.d".format(unit_name
))
120 dropin_path
= os
.path
.join(dropin_dir
, "{}.conf".format(dropin_name
))
122 os
.makedirs(dropin_dir
, exist_ok
=True)
123 self
.addCleanup(os
.rmdir
, dropin_dir
)
124 with
open(dropin_path
, 'w') as dropin
:
125 dropin
.write(contents
)
126 self
.addCleanup(os
.remove
, dropin_path
)
128 def read_attr(self
, link
, attribute
):
129 """Read a link attributed from the sysfs."""
130 # Note we we don't want to check if interface `link' is managed, we
131 # want to evaluate link variable and pass the value of the link to
132 # assert_link_states e.g. eth0=managed.
133 self
.assert_link_states(**{link
:'managed'})
134 with
open(os
.path
.join('/sys/class/net', link
, attribute
)) as f
:
135 return f
.readline().strip()
137 def assert_link_states(self
, **kwargs
):
138 """Match networkctl link states to the given ones.
140 Each keyword argument should be the name of a network interface
141 with its expected value of the "SETUP" column in output from
142 networkctl. The interfaces have five seconds to come online
143 before the check is performed. Every specified interface must
144 be present in the output, and any other interfaces found in the
147 A special interface state "managed" is supported, which matches
148 any value in the "SETUP" column other than "unmanaged".
152 interfaces
= set(kwargs
)
154 # Wait for the requested interfaces, but don't fail for them.
155 subprocess
.call([NETWORKD_WAIT_ONLINE
, '--timeout=5'] +
156 ['--interface={}'.format(iface
) for iface
in kwargs
])
158 # Validate each link state found in the networkctl output.
159 out
= subprocess
.check_output(['networkctl', '--no-legend']).rstrip()
160 for line
in out
.decode('utf-8').split('\n'):
161 fields
= line
.split()
162 if len(fields
) >= 5 and fields
[1] in kwargs
:
164 expected
= kwargs
[iface
]
166 if (actual
!= expected
and
167 not (expected
== 'managed' and actual
!= 'unmanaged')):
168 self
.fail("Link {} expects state {}, found {}".format(iface
, expected
, actual
))
169 interfaces
.remove(iface
)
171 # Ensure that all requested interfaces have been covered.
173 self
.fail("Missing links in status output: {}".format(interfaces
))
176 class BridgeTest(NetworkdTestingUtilities
, unittest
.TestCase
):
177 """Provide common methods for testing networkd against servers."""
180 self
.write_network('port1.netdev', '''\
184 MACAddress=12:34:56:78:9a:bc''')
185 self
.write_network('port2.netdev', '''\
189 MACAddress=12:34:56:78:9a:bd''')
190 self
.write_network('mybridge.netdev', '''\
194 self
.write_network('port1.network', '''\
199 self
.write_network('port2.network', '''\
204 self
.write_network('mybridge.network', '''\
209 Address=192.168.250.33/24
210 Gateway=192.168.250.1''')
211 subprocess
.call(['systemctl', 'reset-failed', 'systemd-networkd', 'systemd-resolved'])
212 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
215 subprocess
.check_call(['systemctl', 'stop', 'systemd-networkd'])
216 subprocess
.check_call(['ip', 'link', 'del', 'mybridge'])
217 subprocess
.check_call(['ip', 'link', 'del', 'port1'])
218 subprocess
.check_call(['ip', 'link', 'del', 'port2'])
220 def test_bridge_init(self
):
221 self
.assert_link_states(
226 def test_bridge_port_priority(self
):
227 self
.assertEqual(self
.read_attr('port1', 'brport/priority'), '32')
228 self
.write_network_dropin('port1.network', 'priority', '''\
232 subprocess
.check_call(['systemctl', 'restart', 'systemd-networkd'])
233 self
.assertEqual(self
.read_attr('port1', 'brport/priority'), '28')
235 def test_bridge_port_priority_set_zero(self
):
236 """It should be possible to set the bridge port priority to 0"""
237 self
.assertEqual(self
.read_attr('port2', 'brport/priority'), '32')
238 self
.write_network_dropin('port2.network', 'priority', '''\
242 subprocess
.check_call(['systemctl', 'restart', 'systemd-networkd'])
243 self
.assertEqual(self
.read_attr('port2', 'brport/priority'), '0')
245 def test_bridge_port_property(self
):
246 """Test the "[Bridge]" section keys"""
247 self
.assertEqual(self
.read_attr('port2', 'brport/priority'), '32')
248 self
.write_network_dropin('port2.network', 'property', '''\
254 AllowPortToBeRoot=true
258 subprocess
.check_call(['systemctl', 'restart', 'systemd-networkd'])
260 self
.assertEqual(self
.read_attr('port2', 'brport/priority'), '23')
261 self
.assertEqual(self
.read_attr('port2', 'brport/hairpin_mode'), '1')
262 self
.assertEqual(self
.read_attr('port2', 'brport/path_cost'), '555')
263 self
.assertEqual(self
.read_attr('port2', 'brport/multicast_fast_leave'), '1')
264 self
.assertEqual(self
.read_attr('port2', 'brport/unicast_flood'), '1')
265 self
.assertEqual(self
.read_attr('port2', 'brport/bpdu_guard'), '1')
266 self
.assertEqual(self
.read_attr('port2', 'brport/root_block'), '1')
268 class ClientTestBase(NetworkdTestingUtilities
):
269 """Provide common methods for testing networkd against servers."""
272 def setUpClass(klass
):
273 klass
.orig_log_level
= subprocess
.check_output(
274 ['systemctl', 'show', '--value', '--property', 'LogLevel'],
275 universal_newlines
=True).strip()
276 subprocess
.check_call(['systemd-analyze', 'set-log-level', 'debug'])
279 def tearDownClass(klass
):
280 subprocess
.check_call(['systemd-analyze', 'set-log-level', klass
.orig_log_level
])
283 self
.iface
= 'test_eth42'
284 self
.if_router
= 'router_eth42'
285 self
.workdir_obj
= tempfile
.TemporaryDirectory()
286 self
.workdir
= self
.workdir_obj
.name
287 self
.config
= 'test_eth42.network'
289 # get current journal cursor
290 subprocess
.check_output(['journalctl', '--sync'])
291 out
= subprocess
.check_output(['journalctl', '-b', '--quiet',
292 '--no-pager', '-n0', '--show-cursor'],
293 universal_newlines
=True)
294 self
.assertTrue(out
.startswith('-- cursor:'))
295 self
.journal_cursor
= out
.split()[-1]
297 subprocess
.call(['systemctl', 'reset-failed', 'systemd-networkd', 'systemd-resolved'])
300 self
.shutdown_iface()
301 subprocess
.call(['systemctl', 'stop', 'systemd-networkd'])
302 subprocess
.call(['ip', 'link', 'del', 'dummy0'],
303 stderr
=subprocess
.DEVNULL
)
305 def show_journal(self
, unit
):
306 '''Show journal of given unit since start of the test'''
308 print('---- {} ----'.format(unit
))
309 subprocess
.check_output(['journalctl', '--sync'])
311 subprocess
.call(['journalctl', '-b', '--no-pager', '--quiet',
312 '--cursor', self
.journal_cursor
, '-u', unit
])
314 def create_iface(self
, ipv6
=False):
315 '''Create test interface with DHCP server behind it'''
317 raise NotImplementedError('must be implemented by a subclass')
319 def shutdown_iface(self
):
320 '''Remove test interface and stop DHCP server'''
322 raise NotImplementedError('must be implemented by a subclass')
324 def print_server_log(self
):
325 '''Print DHCP server log for debugging failures'''
327 raise NotImplementedError('must be implemented by a subclass')
329 def do_test(self
, coldplug
=True, ipv6
=False, extra_opts
='',
330 online_timeout
=10, dhcp_mode
='yes'):
332 subprocess
.check_call(['systemctl', 'start', 'systemd-resolved'])
333 except subprocess
.CalledProcessError
:
334 self
.show_journal('systemd-resolved.service')
336 self
.write_network(self
.config
, '''\
341 {}'''.format(self
.iface
, dhcp_mode
, extra_opts
))
344 # create interface first, then start networkd
345 self
.create_iface(ipv6
=ipv6
)
346 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
347 elif coldplug
is not None:
348 # start networkd first, then create interface
349 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
350 self
.create_iface(ipv6
=ipv6
)
352 # "None" means test sets up interface by itself
353 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
356 subprocess
.check_call([NETWORKD_WAIT_ONLINE
, '--interface',
357 self
.iface
, '--timeout=%i' % online_timeout
])
360 # check iface state and IP 6 address; FIXME: we need to wait a bit
361 # longer, as the iface is "configured" already with IPv4 *or*
362 # IPv6, but we want to wait for both
364 out
= subprocess
.check_output(['ip', 'a', 'show', 'dev', self
.iface
])
365 if b
'state UP' in out
and b
'inet6 2600' in out
and b
'inet 192.168' in out
:
369 self
.fail('timed out waiting for IPv6 configuration')
371 self
.assertRegex(out
, b
'inet6 2600::.* scope global .*dynamic')
372 self
.assertRegex(out
, b
'inet6 fe80::.* scope link')
374 # should have link-local address on IPv6 only
375 out
= subprocess
.check_output(['ip', '-6', 'a', 'show', 'dev', self
.iface
])
376 self
.assertRegex(out
, br
'inet6 fe80::.* scope link')
377 self
.assertNotIn(b
'scope global', out
)
379 # should have IPv4 address
380 out
= subprocess
.check_output(['ip', '-4', 'a', 'show', 'dev', self
.iface
])
381 self
.assertIn(b
'state UP', out
)
382 self
.assertRegex(out
, br
'inet 192.168.5.\d+/.* scope global dynamic')
384 # check networkctl state
385 out
= subprocess
.check_output(['networkctl'])
386 self
.assertRegex(out
, (r
'{}\s+ether\s+[a-z-]+\s+unmanaged'.format(self
.if_router
)).encode())
387 self
.assertRegex(out
, (r
'{}\s+ether\s+routable\s+configured'.format(self
.iface
)).encode())
389 out
= subprocess
.check_output(['networkctl', 'status', self
.iface
])
390 self
.assertRegex(out
, br
'Type:\s+ether')
391 self
.assertRegex(out
, br
'State:\s+routable.*configured')
392 self
.assertRegex(out
, br
'Address:\s+192.168.5.\d+')
394 self
.assertRegex(out
, br
'2600::')
396 self
.assertNotIn(br
'2600::', out
)
397 self
.assertRegex(out
, br
'fe80::')
398 self
.assertRegex(out
, br
'Gateway:\s+192.168.5.1')
399 self
.assertRegex(out
, br
'DNS:\s+192.168.5.1')
400 except (AssertionError, subprocess
.CalledProcessError
):
401 # show networkd status, journal, and DHCP server log on failure
402 with
open(os
.path
.join(NETWORK_UNITDIR
, self
.config
)) as f
:
403 print('\n---- {} ----\n{}'.format(self
.config
, f
.read()))
404 print('---- interface status ----')
406 subprocess
.call(['ip', 'a', 'show', 'dev', self
.iface
])
407 print('---- networkctl status {} ----'.format(self
.iface
))
409 subprocess
.call(['networkctl', 'status', self
.iface
])
410 self
.show_journal('systemd-networkd.service')
411 self
.print_server_log()
414 for timeout
in range(50):
415 with
open(RESOLV_CONF
) as f
:
417 if 'nameserver 192.168.5.1\n' in contents
:
421 self
.fail('nameserver 192.168.5.1 not found in ' + RESOLV_CONF
)
423 if coldplug
is False:
424 # check post-down.d hook
425 self
.shutdown_iface()
427 def test_coldplug_dhcp_yes_ip4(self
):
428 # we have a 12s timeout on RA, so we need to wait longer
429 self
.do_test(coldplug
=True, ipv6
=False, online_timeout
=15)
431 def test_coldplug_dhcp_yes_ip4_no_ra(self
):
432 # with disabling RA explicitly things should be fast
433 self
.do_test(coldplug
=True, ipv6
=False,
434 extra_opts
='IPv6AcceptRA=False')
436 def test_coldplug_dhcp_ip4_only(self
):
437 # we have a 12s timeout on RA, so we need to wait longer
438 self
.do_test(coldplug
=True, ipv6
=False, dhcp_mode
='ipv4',
441 def test_coldplug_dhcp_ip4_only_no_ra(self
):
442 # with disabling RA explicitly things should be fast
443 self
.do_test(coldplug
=True, ipv6
=False, dhcp_mode
='ipv4',
444 extra_opts
='IPv6AcceptRA=False')
446 def test_coldplug_dhcp_ip6(self
):
447 self
.do_test(coldplug
=True, ipv6
=True)
449 def test_hotplug_dhcp_ip4(self
):
450 # With IPv4 only we have a 12s timeout on RA, so we need to wait longer
451 self
.do_test(coldplug
=False, ipv6
=False, online_timeout
=15)
453 def test_hotplug_dhcp_ip6(self
):
454 self
.do_test(coldplug
=False, ipv6
=True)
456 def test_route_only_dns(self
):
457 self
.write_network('myvpn.netdev', '''\
461 MACAddress=12:34:56:78:9a:bc''')
462 self
.write_network('myvpn.network', '''\
466 Address=192.168.42.100
468 Domains= ~company''')
470 self
.do_test(coldplug
=True, ipv6
=False,
471 extra_opts
='IPv6AcceptRouterAdvertisements=False')
473 with
open(RESOLV_CONF
) as f
:
475 # ~company is not a search domain, only a routing domain
476 self
.assertNotRegex(contents
, 'search.*company')
477 # our global server should appear
478 self
.assertIn('nameserver 192.168.5.1\n', contents
)
479 # should not have domain-restricted server as global server
480 self
.assertNotIn('nameserver 192.168.42.1\n', contents
)
482 def test_route_only_dns_all_domains(self
):
483 self
.write_network('myvpn.netdev', '''[NetDev]
486 MACAddress=12:34:56:78:9a:bc''')
487 self
.write_network('myvpn.network', '''[Match]
490 Address=192.168.42.100
492 Domains= ~company ~.''')
494 self
.do_test(coldplug
=True, ipv6
=False,
495 extra_opts
='IPv6AcceptRouterAdvertisements=False')
497 with
open(RESOLV_CONF
) as f
:
500 # ~company is not a search domain, only a routing domain
501 self
.assertNotRegex(contents
, 'search.*company')
503 # our global server should appear
504 self
.assertIn('nameserver 192.168.5.1\n', contents
)
505 # should have company server as global server due to ~.
506 self
.assertIn('nameserver 192.168.42.1\n', contents
)
509 @unittest.skipUnless(HAVE_DNSMASQ
, 'dnsmasq not installed')
510 class DnsmasqClientTest(ClientTestBase
, unittest
.TestCase
):
511 '''Test networkd client against dnsmasq'''
516 self
.iface_mac
= 'de:ad:be:ef:47:11'
518 def create_iface(self
, ipv6
=False, dnsmasq_opts
=None):
519 '''Create test interface with DHCP server behind it'''
522 subprocess
.check_call(['ip', 'link', 'add', 'name', self
.iface
,
523 'address', self
.iface_mac
,
524 'type', 'veth', 'peer', 'name', self
.if_router
])
526 # give our router an IP
527 subprocess
.check_call(['ip', 'a', 'flush', 'dev', self
.if_router
])
528 subprocess
.check_call(['ip', 'a', 'add', '192.168.5.1/24', 'dev', self
.if_router
])
530 subprocess
.check_call(['ip', 'a', 'add', '2600::1/64', 'dev', self
.if_router
])
531 subprocess
.check_call(['ip', 'link', 'set', self
.if_router
, 'up'])
534 self
.dnsmasq_log
= os
.path
.join(self
.workdir
, 'dnsmasq.log')
535 lease_file
= os
.path
.join(self
.workdir
, 'dnsmasq.leases')
537 extra_opts
= ['--enable-ra', '--dhcp-range=2600::10,2600::20']
541 extra_opts
+= dnsmasq_opts
542 self
.dnsmasq
= subprocess
.Popen(
543 ['dnsmasq', '--keep-in-foreground', '--log-queries',
544 '--log-facility=' + self
.dnsmasq_log
, '--conf-file=/dev/null',
545 '--dhcp-leasefile=' + lease_file
, '--bind-interfaces',
546 '--interface=' + self
.if_router
, '--except-interface=lo',
547 '--dhcp-range=192.168.5.10,192.168.5.200'] + extra_opts
)
549 def shutdown_iface(self
):
550 '''Remove test interface and stop DHCP server'''
553 subprocess
.check_call(['ip', 'link', 'del', 'dev', self
.if_router
])
554 self
.if_router
= None
560 def print_server_log(self
):
561 '''Print DHCP server log for debugging failures'''
563 with
open(self
.dnsmasq_log
) as f
:
564 sys
.stdout
.write('\n\n---- dnsmasq log ----\n{}\n------\n\n'.format(f
.read()))
566 def test_resolved_domain_restricted_dns(self
):
567 '''resolved: domain-restricted DNS servers'''
569 # create interface for generic connections; this will map all DNS names
571 self
.create_iface(dnsmasq_opts
=['--address=/#/192.168.42.1'])
572 self
.write_network('general.network', '''\
577 IPv6AcceptRA=False'''.format(self
.iface
))
579 # create second device/dnsmasq for a .company/.lab VPN interface
580 # static IPs for simplicity
581 self
.add_veth_pair('testvpnclient', 'testvpnrouter')
582 subprocess
.check_call(['ip', 'a', 'flush', 'dev', 'testvpnrouter'])
583 subprocess
.check_call(['ip', 'a', 'add', '10.241.3.1/24', 'dev', 'testvpnrouter'])
584 subprocess
.check_call(['ip', 'link', 'set', 'testvpnrouter', 'up'])
586 vpn_dnsmasq_log
= os
.path
.join(self
.workdir
, 'dnsmasq-vpn.log')
587 vpn_dnsmasq
= subprocess
.Popen(
588 ['dnsmasq', '--keep-in-foreground', '--log-queries',
589 '--log-facility=' + vpn_dnsmasq_log
, '--conf-file=/dev/null',
590 '--dhcp-leasefile=/dev/null', '--bind-interfaces',
591 '--interface=testvpnrouter', '--except-interface=lo',
592 '--address=/math.lab/10.241.3.3', '--address=/cantina.company/10.241.4.4'])
593 self
.addCleanup(vpn_dnsmasq
.wait
)
594 self
.addCleanup(vpn_dnsmasq
.kill
)
596 self
.write_network('vpn.network', '''\
601 Address=10.241.3.2/24
603 Domains= ~company ~lab''')
605 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
606 subprocess
.check_call([NETWORKD_WAIT_ONLINE
, '--interface', self
.iface
,
607 '--interface=testvpnclient', '--timeout=20'])
609 # ensure we start fresh with every test
610 subprocess
.check_call(['systemctl', 'restart', 'systemd-resolved'])
612 # test vpnclient specific domains; these should *not* be answered by
614 out
= subprocess
.check_output(['systemd-resolve', 'math.lab'])
615 self
.assertIn(b
'math.lab: 10.241.3.3', out
)
616 out
= subprocess
.check_output(['systemd-resolve', 'kettle.cantina.company'])
617 self
.assertIn(b
'kettle.cantina.company: 10.241.4.4', out
)
619 # test general domains
620 out
= subprocess
.check_output(['systemd-resolve', 'megasearch.net'])
621 self
.assertIn(b
'megasearch.net: 192.168.42.1', out
)
623 with
open(self
.dnsmasq_log
) as f
:
624 general_log
= f
.read()
625 with
open(vpn_dnsmasq_log
) as f
:
628 # VPN domains should only be sent to VPN DNS
629 self
.assertRegex(vpn_log
, 'query.*math.lab')
630 self
.assertRegex(vpn_log
, 'query.*cantina.company')
631 self
.assertNotIn('.lab', general_log
)
632 self
.assertNotIn('.company', general_log
)
634 # general domains should not be sent to the VPN DNS
635 self
.assertRegex(general_log
, 'query.*megasearch.net')
636 self
.assertNotIn('megasearch.net', vpn_log
)
638 def test_resolved_etc_hosts(self
):
639 '''resolved queries to /etc/hosts'''
641 # FIXME: -t MX query fails with enabled DNSSEC (even when using
642 # the known negative trust anchor .internal instead of .example)
643 conf
= '/run/systemd/resolved.conf.d/test-disable-dnssec.conf'
644 os
.makedirs(os
.path
.dirname(conf
), exist_ok
=True)
645 with
open(conf
, 'w') as f
:
646 f
.write('[Resolve]\nDNSSEC=no')
647 self
.addCleanup(os
.remove
, conf
)
649 # create /etc/hosts bind mount which resolves my.example for IPv4
650 hosts
= os
.path
.join(self
.workdir
, 'hosts')
651 with
open(hosts
, 'w') as f
:
652 f
.write('172.16.99.99 my.example\n')
653 subprocess
.check_call(['mount', '--bind', hosts
, '/etc/hosts'])
654 self
.addCleanup(subprocess
.call
, ['umount', '/etc/hosts'])
655 subprocess
.check_call(['systemctl', 'stop', 'systemd-resolved.service'])
657 # note: different IPv4 address here, so that it's easy to tell apart
658 # what resolved the query
659 self
.create_iface(dnsmasq_opts
=['--host-record=my.example,172.16.99.1,2600::99:99',
660 '--host-record=other.example,172.16.0.42,2600::42',
661 '--mx-host=example,mail.example'],
663 self
.do_test(coldplug
=None, ipv6
=True)
666 # family specific queries
667 out
= subprocess
.check_output(['systemd-resolve', '-4', 'my.example'])
668 self
.assertIn(b
'my.example: 172.16.99.99', out
)
669 # we don't expect an IPv6 answer; if /etc/hosts has any IP address,
670 # it's considered a sufficient source
671 self
.assertNotEqual(subprocess
.call(['systemd-resolve', '-6', 'my.example']), 0)
672 # "any family" query; IPv4 should come from /etc/hosts
673 out
= subprocess
.check_output(['systemd-resolve', 'my.example'])
674 self
.assertIn(b
'my.example: 172.16.99.99', out
)
675 # IP → name lookup; again, takes the /etc/hosts one
676 out
= subprocess
.check_output(['systemd-resolve', '172.16.99.99'])
677 self
.assertIn(b
'172.16.99.99: my.example', out
)
679 # non-address RRs should fall back to DNS
680 out
= subprocess
.check_output(['systemd-resolve', '--type=MX', 'example'])
681 self
.assertIn(b
'example IN MX 1 mail.example', out
)
683 # other domains query DNS
684 out
= subprocess
.check_output(['systemd-resolve', 'other.example'])
685 self
.assertIn(b
'172.16.0.42', out
)
686 out
= subprocess
.check_output(['systemd-resolve', '172.16.0.42'])
687 self
.assertIn(b
'172.16.0.42: other.example', out
)
688 except (AssertionError, subprocess
.CalledProcessError
):
689 self
.show_journal('systemd-resolved.service')
690 self
.print_server_log()
693 def test_transient_hostname(self
):
694 '''networkd sets transient hostname from DHCP'''
696 orig_hostname
= socket
.gethostname()
697 self
.addCleanup(socket
.sethostname
, orig_hostname
)
698 # temporarily move /etc/hostname away; restart hostnamed to pick it up
699 if os
.path
.exists('/etc/hostname'):
700 subprocess
.check_call(['mount', '--bind', '/dev/null', '/etc/hostname'])
701 self
.addCleanup(subprocess
.call
, ['umount', '/etc/hostname'])
702 subprocess
.check_call(['systemctl', 'stop', 'systemd-hostnamed.service'])
704 self
.create_iface(dnsmasq_opts
=['--dhcp-host={},192.168.5.210,testgreen'.format(self
.iface_mac
)])
705 self
.do_test(coldplug
=None, extra_opts
='IPv6AcceptRA=False', dhcp_mode
='ipv4')
708 # should have received the fixed IP above
709 out
= subprocess
.check_output(['ip', '-4', 'a', 'show', 'dev', self
.iface
])
710 self
.assertRegex(out
, b
'inet 192.168.5.210/24 .* scope global dynamic')
711 # should have set transient hostname in hostnamed; this is
712 # sometimes a bit lagging (issue #4753), so retry a few times
713 for retry
in range(1, 6):
714 out
= subprocess
.check_output(['hostnamectl'])
715 if b
'testgreen' in out
:
718 sys
.stdout
.write('[retry %i] ' % retry
)
721 self
.fail('Transient hostname not found in hostnamectl:\n{}'.format(out
.decode()))
722 # and also applied to the system
723 self
.assertEqual(socket
.gethostname(), 'testgreen')
724 except AssertionError:
725 self
.show_journal('systemd-networkd.service')
726 self
.show_journal('systemd-hostnamed.service')
727 self
.print_server_log()
730 def test_transient_hostname_with_static(self
):
731 '''transient hostname is not applied if static hostname exists'''
733 orig_hostname
= socket
.gethostname()
734 self
.addCleanup(socket
.sethostname
, orig_hostname
)
735 if not os
.path
.exists('/etc/hostname'):
736 self
.writeConfig('/etc/hostname', orig_hostname
)
737 subprocess
.check_call(['systemctl', 'stop', 'systemd-hostnamed.service'])
739 self
.create_iface(dnsmasq_opts
=['--dhcp-host={},192.168.5.210,testgreen'.format(self
.iface_mac
)])
740 self
.do_test(coldplug
=None, extra_opts
='IPv6AcceptRA=False', dhcp_mode
='ipv4')
743 # should have received the fixed IP above
744 out
= subprocess
.check_output(['ip', '-4', 'a', 'show', 'dev', self
.iface
])
745 self
.assertRegex(out
, b
'inet 192.168.5.210/24 .* scope global dynamic')
746 # static hostname wins over transient one, thus *not* applied
747 self
.assertEqual(socket
.gethostname(), orig_hostname
)
748 except AssertionError:
749 self
.show_journal('systemd-networkd.service')
750 self
.show_journal('systemd-hostnamed.service')
751 self
.print_server_log()
755 class NetworkdClientTest(ClientTestBase
, unittest
.TestCase
):
756 '''Test networkd client against networkd server'''
762 def create_iface(self
, ipv6
=False, dhcpserver_opts
=None):
763 '''Create test interface with DHCP server behind it'''
765 # run "router-side" networkd in own mount namespace to shield it from
766 # "client-side" configuration and networkd
767 (fd
, script
) = tempfile
.mkstemp(prefix
='networkd-router.sh')
768 self
.addCleanup(os
.remove
, script
)
769 with os
.fdopen(fd
, 'w+') as f
:
773 mkdir -p /run/systemd/network
774 mkdir -p /run/systemd/netif
775 mount -t tmpfs none /run/systemd/network
776 mount -t tmpfs none /run/systemd/netif
777 [ ! -e /run/dbus ] || mount -t tmpfs none /run/dbus
778 # create router/client veth pair
779 cat << EOF > /run/systemd/network/test.netdev
788 cat << EOF > /run/systemd/network/test.network
793 Address=192.168.5.1/24
804 # run networkd as in systemd-networkd.service
805 exec $(systemctl cat systemd-networkd.service | sed -n '/^ExecStart=/ { s/^.*=//; s/^[@+-]//; s/^!*//; p}')
806 ''' % {'ifr': self
.if_router
, 'ifc': self
.iface
, 'addr6': ipv6
and 'Address=2600::1/64' or '',
807 'dhopts': dhcpserver_opts
or ''})
811 subprocess
.check_call(['systemd-run', '--unit=networkd-test-router.service',
812 '-p', 'InaccessibleDirectories=-/etc/systemd/network',
813 '-p', 'InaccessibleDirectories=-/run/systemd/network',
814 '-p', 'InaccessibleDirectories=-/run/systemd/netif',
815 '--service-type=notify', script
])
817 # wait until devices got created
819 out
= subprocess
.check_output(['ip', 'a', 'show', 'dev', self
.if_router
])
820 if b
'state UP' in out
and b
'scope global' in out
:
824 def shutdown_iface(self
):
825 '''Remove test interface and stop DHCP server'''
828 subprocess
.check_call(['systemctl', 'stop', 'networkd-test-router.service'])
829 # ensure failed transient unit does not stay around
830 subprocess
.call(['systemctl', 'reset-failed', 'networkd-test-router.service'])
831 subprocess
.call(['ip', 'link', 'del', 'dev', self
.if_router
])
832 self
.if_router
= None
834 def print_server_log(self
):
835 '''Print DHCP server log for debugging failures'''
837 self
.show_journal('networkd-test-router.service')
839 @unittest.skip('networkd does not have DHCPv6 server support')
840 def test_hotplug_dhcp_ip6(self
):
843 @unittest.skip('networkd does not have DHCPv6 server support')
844 def test_coldplug_dhcp_ip6(self
):
847 def test_search_domains(self
):
849 # we don't use this interface for this test
850 self
.if_router
= None
852 self
.write_network('test.netdev', '''\
856 MACAddress=12:34:56:78:9a:bc''')
857 self
.write_network('test.network', '''\
861 Address=192.168.42.100
863 Domains= one two three four five six seven eight nine ten''')
865 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
867 for timeout
in range(50):
868 with
open(RESOLV_CONF
) as f
:
870 if ' one' in contents
:
873 self
.assertRegex(contents
, 'search .*one two three four')
874 self
.assertNotIn('seven\n', contents
)
875 self
.assertIn('# Too many search domains configured, remaining ones ignored.\n', contents
)
877 def test_search_domains_too_long(self
):
879 # we don't use this interface for this test
880 self
.if_router
= None
882 name_prefix
= 'a' * 60
884 self
.write_network('test.netdev', '''\
888 MACAddress=12:34:56:78:9a:bc''')
889 self
.write_network('test.network', '''\
893 Address=192.168.42.100
895 Domains={p}0 {p}1 {p}2 {p}3 {p}4'''.format(p
=name_prefix
))
897 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
899 for timeout
in range(50):
900 with
open(RESOLV_CONF
) as f
:
902 if ' one' in contents
:
905 self
.assertRegex(contents
, 'search .*{p}0 {p}1 {p}2'.format(p
=name_prefix
))
906 self
.assertIn('# Total length of all search domains is too long, remaining ones ignored.', contents
)
908 def test_dropin(self
):
909 # we don't use this interface for this test
910 self
.if_router
= None
912 self
.write_network('test.netdev', '''\
916 MACAddress=12:34:56:78:9a:bc''')
917 self
.write_network('test.network', '''\
921 Address=192.168.42.100
923 self
.write_network_dropin('test.network', 'dns', '''\
927 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
929 for timeout
in range(50):
930 with
open(RESOLV_CONF
) as f
:
932 if ' 127.0.0.1' in contents
:
935 self
.assertIn('nameserver 192.168.42.1\n', contents
)
936 self
.assertIn('nameserver 127.0.0.1\n', contents
)
938 def test_dhcp_timezone(self
):
939 '''networkd sets time zone from DHCP'''
942 out
= subprocess
.check_output(['busctl', 'get-property', 'org.freedesktop.timedate1',
943 '/org/freedesktop/timedate1', 'org.freedesktop.timedate1', 'Timezone'])
944 assert out
.startswith(b
's "')
946 assert out
.endswith(b
'"')
947 return out
[3:-1].decode()
949 orig_timezone
= get_tz()
950 self
.addCleanup(subprocess
.call
, ['timedatectl', 'set-timezone', orig_timezone
])
952 self
.create_iface(dhcpserver_opts
='EmitTimezone=yes\nTimezone=Pacific/Honolulu')
953 self
.do_test(coldplug
=None, extra_opts
='IPv6AcceptRA=false\n[DHCP]\nUseTimezone=true', dhcp_mode
='ipv4')
955 # should have applied the received timezone
957 self
.assertEqual(get_tz(), 'Pacific/Honolulu')
958 except AssertionError:
959 self
.show_journal('systemd-networkd.service')
960 self
.show_journal('systemd-hostnamed.service')
964 class MatchClientTest(unittest
.TestCase
, NetworkdTestingUtilities
):
965 """Test [Match] sections in .network files.
967 Be aware that matching the test host's interfaces will wipe their
968 configuration, so as a precaution, all network files should have a
969 restrictive [Match] section to only ever interfere with the
970 temporary veth interfaces created here.
975 subprocess
.call(['systemctl', 'stop', 'systemd-networkd'])
977 def test_basic_matching(self
):
978 """Verify the Name= line works throughout this class."""
979 self
.add_veth_pair('test_if1', 'fake_if2')
980 self
.write_network('test.network', "[Match]\nName=test_*\n[Network]")
981 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
982 self
.assert_link_states(test_if1
='managed', fake_if2
='unmanaged')
984 def test_inverted_matching(self
):
985 """Verify that a '!'-prefixed value inverts the match."""
986 # Use a MAC address as the interfaces' common matching attribute
987 # to avoid depending on udev, to support testing in containers.
988 mac
= '00:01:02:03:98:99'
989 self
.add_veth_pair('test_veth', 'test_peer',
990 ['addr', mac
], ['addr', mac
])
991 self
.write_network('no-veth.network', """\
994 Name=!nonexistent *peer*
995 [Network]""".format(mac
))
996 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
997 self
.assert_link_states(test_veth
='managed', test_peer
='unmanaged')
1000 class UnmanagedClientTest(unittest
.TestCase
, NetworkdTestingUtilities
):
1001 """Test if networkd manages the correct interfaces."""
1004 """Write .network files to match the named veth devices."""
1005 # Define the veth+peer pairs to be created.
1006 # Their pairing doesn't actually matter, only their names do.
1012 # Define the contents of .network files to be read in order.
1014 "[Match]\nName=m1def\n",
1015 "[Match]\nName=m1unm\n[Link]\nUnmanaged=yes\n",
1016 "[Match]\nName=m1*\n[Link]\nUnmanaged=no\n",
1019 # Write out the .network files to be cleaned up automatically.
1020 for i
, config
in enumerate(self
.configs
):
1021 self
.write_network("%02d-test.network" % i
, config
)
1024 """Stop networkd."""
1025 subprocess
.call(['systemctl', 'stop', 'systemd-networkd'])
1027 def create_iface(self
):
1028 """Create temporary veth pairs for interface matching."""
1029 for veth
, peer
in self
.veths
.items():
1030 self
.add_veth_pair(veth
, peer
)
1032 def test_unmanaged_setting(self
):
1033 """Verify link states with Unmanaged= settings, hot-plug."""
1034 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
1036 self
.assert_link_states(m1def
='managed',
1041 def test_unmanaged_setting_coldplug(self
):
1042 """Verify link states with Unmanaged= settings, cold-plug."""
1044 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
1045 self
.assert_link_states(m1def
='managed',
1050 def test_catchall_config(self
):
1051 """Verify link states with a catch-all config, hot-plug."""
1052 # Don't actually catch ALL interfaces. It messes up the host.
1053 self
.write_network('all.network', "[Match]\nName=m[01]???\n")
1054 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
1056 self
.assert_link_states(m1def
='managed',
1061 def test_catchall_config_coldplug(self
):
1062 """Verify link states with a catch-all config, cold-plug."""
1063 # Don't actually catch ALL interfaces. It messes up the host.
1064 self
.write_network('all.network', "[Match]\nName=m[01]???\n")
1066 subprocess
.check_call(['systemctl', 'start', 'systemd-networkd'])
1067 self
.assert_link_states(m1def
='managed',
1073 if __name__
== '__main__':
1074 unittest
.main(testRunner
=unittest
.TextTestRunner(stream
=sys
.stdout
,