resolvectl_bin = shutil.which('resolvectl', path=which_paths)
timedatectl_bin = shutil.which('timedatectl', path=which_paths)
udevadm_bin = shutil.which('udevadm', path=which_paths)
+test_ndisc_send = None
build_dir = None
source_dir = None
self.assertRegex(output, route_regex)
+ def wait_route_dropped(self, link, route_regex, table='main', ipv='', timeout_sec=100):
+ for i in range(timeout_sec):
+ if i > 0:
+ time.sleep(1)
+ output = check_output(f'ip {ipv} route show dev {link} table {table}')
+ if not re.search(route_regex, output):
+ break
+
+ self.assertNotRegex(output, route_regex)
+
def check_netlabel(self, interface, address, label='system_u:object_r:root_t:s0'):
if not shutil.which('selinuxenabled'):
print('## Checking NetLabel skipped: selinuxenabled command not found.')
self.check_ipv6_token_static()
+ def test_ndisc_redirect(self):
+ if not os.path.exists(test_ndisc_send):
+ self.skipTest(f"{test_ndisc_send} does not exist.")
+
+ copy_network_unit('25-veth.netdev', '25-ipv6-prefix.network', '25-ipv6-prefix-veth-token-static.network')
+ start_networkd()
+
+ self.check_ipv6_token_static()
+
+ # Introduce two redirect routes.
+ check_output(f'{test_ndisc_send} --interface veth-peer --type redirect --target-address 2002:da8:1:1:1a:2b:3c:4d --redirect-destination 2002:da8:1:1:1a:2b:3c:4d')
+ check_output(f'{test_ndisc_send} --interface veth-peer --type redirect --target-address 2002:da8:1::1 --redirect-destination 2002:da8:1:2:1a:2b:3c:4d')
+ self.wait_route('veth99', r'2002:da8:1:1:1a:2b:3c:4d proto redirect', ipv='-6', timeout_sec=10)
+ self.wait_route('veth99', r'2002:da8:1:2:1a:2b:3c:4d via 2002:da8:1::1 proto redirect', ipv='-6', timeout_sec=10)
+
+ # Change the target address of the redirects.
+ check_output(f'{test_ndisc_send} --interface veth-peer --type redirect --target-address 2002:da8:1::2 --redirect-destination 2002:da8:1:1:1a:2b:3c:4d')
+ check_output(f'{test_ndisc_send} --interface veth-peer --type redirect --target-address 2002:da8:1::3 --redirect-destination 2002:da8:1:2:1a:2b:3c:4d')
+ self.wait_route_dropped('veth99', r'2002:da8:1:1:1a:2b:3c:4d proto redirect', ipv='-6', timeout_sec=10)
+ self.wait_route_dropped('veth99', r'2002:da8:1:2:1a:2b:3c:4d via 2002:da8:1::1 proto redirect', ipv='-6', timeout_sec=10)
+ self.wait_route('veth99', r'2002:da8:1:1:1a:2b:3c:4d via 2002:da8:1::2 proto redirect', ipv='-6', timeout_sec=10)
+ self.wait_route('veth99', r'2002:da8:1:2:1a:2b:3c:4d via 2002:da8:1::3 proto redirect', ipv='-6', timeout_sec=10)
+
+ # Send Neighbor Advertisement without the router flag to announce the default router is not available anymore.
+ # Then, verify that all redirect routes and the default route are dropped.
+ output = check_output('ip -6 address show dev veth-peer scope link')
+ veth_peer_ipv6ll = re.search('fe80:[:0-9a-f]*', output).group()
+ print(f'veth-peer IPv6LL address: {veth_peer_ipv6ll}')
+ check_output(f'{test_ndisc_send} --interface veth-peer --type neighbor-advertisement --target-address {veth_peer_ipv6ll} --is-router no')
+ self.wait_route_dropped('veth99', 'proto redirect', ipv='-6', timeout_sec=10)
+ self.wait_route_dropped('veth99', 'proto ra', ipv='-6', timeout_sec=10)
+
def test_ipv6_token_prefixstable(self):
copy_network_unit('25-veth.netdev', '25-ipv6-prefix.network', '25-ipv6-prefix-veth-token-prefixstable.network')
start_networkd()
def tearDown(self):
tear_down_common()
- def test_dhcp_server(self):
- copy_network_unit('25-veth.netdev', '25-dhcp-client.network', '25-dhcp-server.network')
- start_networkd()
- self.wait_online('veth99:routable', 'veth-peer:routable')
-
+ def check_dhcp_server(self, persist_leases=True):
output = networkctl_status('veth99')
print(output)
self.assertRegex(output, r'Address: 192.168.5.[0-9]* \(DHCP4 via 192.168.5.1\)')
print(output)
self.assertRegex(output, "Offered DHCP leases: 192.168.5.[0-9]*")
+ if persist_leases:
+ with open('/var/lib/systemd/network/dhcp-server-lease/veth-peer', encoding='utf-8') as f:
+ check_json(f.read())
+ else:
+ self.assertFalse(os.path.exists('/var/lib/systemd/network/dhcp-server-lease/veth-peer'))
+
+ def test_dhcp_server(self):
+ copy_network_unit('25-veth.netdev', '25-dhcp-client.network', '25-dhcp-server.network')
+ start_networkd()
+ self.wait_online('veth99:routable', 'veth-peer:routable')
+
+ self.check_dhcp_server()
+
networkctl_reconfigure('veth-peer')
self.wait_online('veth-peer:routable')
else:
self.fail()
+ def test_dhcp_server_persist_leases_no(self):
+ copy_networkd_conf_dropin('persist-leases-no.conf')
+ copy_network_unit('25-veth.netdev', '25-dhcp-client.network', '25-dhcp-server.network')
+ start_networkd()
+ self.wait_online('veth99:routable', 'veth-peer:routable')
+
+ self.check_dhcp_server(persist_leases=False)
+
+ remove_networkd_conf_dropin('persist-leases-no.conf')
+ with open(os.path.join(network_unit_dir, '25-dhcp-server.network'), mode='a', encoding='utf-8') as f:
+ f.write('[DHCPServer]\nPersistLeases=no')
+ restart_networkd()
+ self.wait_online('veth99:routable', 'veth-peer:routable')
+
+ self.check_dhcp_server(persist_leases=False)
+
def test_dhcp_server_null_server_address(self):
copy_network_unit('25-veth.netdev', '25-dhcp-client.network', '25-dhcp-server-null-server-address.network')
start_networkd()
output = networkctl_status('veth-peer')
self.assertIn(f'Offered DHCP leases: {client_address}', output)
+ # Check if the same addresses are used even if the service is restarted.
+ restart_networkd()
+ self.wait_online('veth99:routable', 'veth-peer:routable')
+
+ output = check_output('ip -4 address show dev veth-peer')
+ print(output)
+ self.assertIn(f'{server_address}', output)
+
+ output = check_output('ip -4 address show dev veth99')
+ print(output)
+ self.assertIn(f'{client_address}', output)
+
+ output = networkctl_status('veth99')
+ print(output)
+ self.assertRegex(output, rf'Address: {client_address} \(DHCP4 via {server_address}\)')
+ self.assertIn(f'Gateway: {server_address}', output)
+ self.assertIn(f'DNS: {server_address}', output)
+ self.assertIn(f'NTP: {server_address}', output)
+
+ output = networkctl_status('veth-peer')
+ self.assertIn(f'Offered DHCP leases: {client_address}', output)
+
def test_dhcp_server_with_uplink(self):
copy_network_unit('25-veth.netdev', '25-dhcp-client.network', '25-dhcp-server-downstream.network',
'12-dummy.netdev', '25-dhcp-server-uplink.network')
check(self, True, False)
check(self, False, True)
check(self, False, False)
+
+ def test_dhcp_client_default_use_domains(self):
+ def check(self, ipv4, ipv6):
+ mkdir_p(networkd_conf_dropin_dir)
+ with open(os.path.join(networkd_conf_dropin_dir, 'default_use_domains.conf'), mode='w', encoding='utf-8') as f:
+ f.write('[DHCPv4]\nUseDomains=')
+ f.write('yes\n' if ipv4 else 'no\n')
+ f.write('[DHCPv6]\nUseDomains=')
+ f.write('yes\n' if ipv6 else 'no\n')
+
+ restart_networkd()
+ self.wait_online('veth-peer:carrier')
+ start_dnsmasq('--dhcp-option=option:dns-server,192.168.5.1',
+ '--dhcp-option=option6:dns-server,[2600::1]',
+ '--dhcp-option=option:domain-search,example.com',
+ '--dhcp-option=option6:domain-search,example.com')
+
+ self.wait_online('veth99:routable')
+
+ # link becomes 'routable' when at least one protocol provide an valid address. Hence, we need to explicitly wait for both addresses.
+ self.wait_address('veth99', r'inet 192.168.5.[0-9]*/24 metric 1024 brd 192.168.5.255 scope global dynamic', ipv='-4')
+ self.wait_address('veth99', r'inet6 2600::[0-9a-f]*/128 scope global (dynamic noprefixroute|noprefixroute dynamic)', ipv='-6')
+
+ for _ in range(20):
+ output = resolvectl('domain', 'veth99')
+ if ipv4 or ipv6:
+ if 'example.com' in output:
+ break
+ else:
+ if 'example.com' not in output:
+ break
+ time.sleep(0.5)
+ else:
+ print(output)
+ self.fail('unexpected domain setting in resolved...')
+
+ stop_dnsmasq()
+ remove_networkd_conf_dropin('default_use_domains.conf')
+
+ copy_network_unit('25-veth.netdev', '25-dhcp-server-veth-peer.network', '25-dhcp-client.network', copy_dropins=False)
+ check(self, True, True)
+ check(self, True, False)
+ check(self, False, True)
+ check(self, False, False)
def test_dhcp_client_use_captive_portal(self):
def check(self, ipv4, ipv6):
udevadm_cmd = valgrind_cmd.split() + [udevadm_bin]
wait_online_cmd = valgrind_cmd.split() + [wait_online_bin]
+ if build_dir:
+ test_ndisc_send = os.path.normpath(os.path.join(build_dir, 'test-ndisc-send'))
+ else:
+ test_ndisc_send = '/usr/lib/tests/test-ndisc-send'
+
if asan_options:
env.update({'ASAN_OPTIONS': asan_options})
if lsan_options: