#!/usr/bin/env python3 # # Copyright VyOS maintainers and contributors # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License version 2 or later as # published by the Free Software Foundation. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see . import unittest from psutil import process_iter from ipaddress import IPv4Address from ipaddress import IPv6Address from ipaddress import IPv4Network from ipaddress import IPv6Network from base_vyostest_shim import VyOSUnitTestSHIM from vyos.configsession import ConfigSessionError from vyos.utils.dict import dict_search_recursive from vyos.utils.network import get_interface_address from vyos.xml_ref import default_value config_file: str = '/etc/ppp/peers/{}' base_path: list = ['interfaces', 'pppoe'] veth_path: list = ['interfaces', 'virtual-ethernet'] pppoe_server_path = ['service', 'pppoe-server'] connect_timeout: int = 20 name_servers: list = ['1.1.1.1', '2.2.2.2'] ipv4_pool: str = '100.64.0.0/18' ipv6_pool: str = '2001:db8:8000::/48' ipv6_pool_pd: str = '2001:db8:9000::/48' def calculate_ipv6_interface_address(prefix: IPv6Network, sla_id: int, interface_id: int): # Ensure SLA-ID is 8 bits if not (0 <= sla_id <= 0xFF): raise ValueError('SLA-ID must be an 8-bit integer (0-255)') # Ensure Interface ID is 64 bits if not (0 <= interface_id <= 0xFFFFFFFFFFFFFFFF): raise ValueError('Interface ID must be a 64-bit integer') # Build the /64 subnet from the PD prefix len + SLA-ID subnet_int = int(prefix.network_address) | (sla_id << 64) # Calculate full interface address return IPv6Address(subnet_int | interface_id) def get_config_value(interface, key): with open(config_file.format(interface), 'r') as f: for line in f: if line.startswith(key): return list(line.split()) return [] def wait_for_interface(interface: str, timeout=connect_timeout) -> bool: """ Wait until PPPoE interface has been connected to the BRAS """ from time import time from time import sleep from vyos.utils.network import get_interface_config start_time = time() while not get_interface_config(interface): sleep(0.250) if time() - start_time >= timeout: return False return True # add a classmethod to setup a temporaray PPPoE server for "proper" validation class PPPoEInterfaceTest(VyOSUnitTestSHIM.TestCase): @classmethod def setUpClass(cls): super(PPPoEInterfaceTest, cls).setUpClass() # ensure we can also run this test on a live system - so lets clean # out the current configuration :) cls.cli_delete(cls, base_path) cls.cli_delete(cls, veth_path) cls.cli_delete(cls, pppoe_server_path) cls._interfaces = ['pppoe10', 'pppoe20', 'pppoe30'] cls._source_interface = 'veth102' pppoe_server_interface = 'veth101' cls.cli_set(cls, veth_path + [pppoe_server_interface, 'peer-name', cls._source_interface]) cls.cli_set(cls, veth_path + [cls._source_interface, 'peer-name', pppoe_server_interface]) cls.cli_set(cls, pppoe_server_path + ['authentication', 'mode', 'local']) cls.cli_set(cls, pppoe_server_path + ['client-ip-pool', 'IPv4-POOL', 'range', ipv4_pool]) cls.cli_set(cls, pppoe_server_path + ['client-ipv6-pool', 'IPv6-POOL', 'prefix', ipv6_pool, 'mask', '64']) cls.cli_set(cls, pppoe_server_path + ['client-ipv6-pool', 'IPv6-POOL', 'delegate', ipv6_pool_pd, 'delegation-prefix', '56']) cls.cli_set(cls, pppoe_server_path + ['default-ipv6-pool', 'IPv6-POOL']) cls.cli_set(cls, pppoe_server_path + ['default-pool', 'IPv4-POOL']) cls.cli_set(cls, pppoe_server_path + ['gateway-address', '100.64.0.1']) cls.cli_set(cls, pppoe_server_path + ['interface', pppoe_server_interface]) for ns in name_servers: cls.cli_set(cls, pppoe_server_path + ['name-server', ns]) cls.cli_set(cls, pppoe_server_path + ['ppp-options', 'disable-ccp']) cls.cli_set(cls, pppoe_server_path + ['ppp-options', 'ipv6', 'allow']) cls.cli_set(cls, pppoe_server_path + ['session-control', 'disable']) cls.u_p_dict = {} for interface in cls._interfaces: username = f'VyOS-user-{interface}' password = f'VyOS-passwd-{interface}' cls.cli_set(cls, pppoe_server_path + ['authentication', 'local-users', 'username', username, 'password', password]) cls.u_p_dict[interface] = (username, password) # Start PPPoE server cls.cli_commit(cls) @classmethod def tearDownClass(cls): cls.cli_delete(cls, base_path) cls.cli_delete(cls, veth_path) cls.cli_delete(cls, pppoe_server_path) # Stop PPPoE server cls.cli_commit(cls) super(PPPoEInterfaceTest, cls).tearDownClass() def tearDown(self): self.cli_delete(base_path) self.cli_commit() # always forward to base class super().tearDown() def _verify_interface_address(self, interface): # Verify that the assigned IPv4/IPv6 addresses from the BRAS (PPPoE # server) are from the assigned pools for address in get_interface_address(interface): if 'family' in address and address['family'] == 'inet': # The PPPoE assigned IPv4 address must be from our pool self.assertIn(IPv4Address(address['address']), IPv4Network(ipv4_pool)) elif 'family' in address and address['family'] == 'inet6': # The PPPoE assigned IPv6 address must be from our pool ipv6 = IPv6Address(address['address']) if not ipv6.is_link_local: self.assertIn(ipv6, IPv6Network(ipv6_pool)) def test_pppoe_client(self): # Check if PPPoE dialer can be configured and runs mtu = '1400' for interface in self._interfaces: (user, passwd) = self.u_p_dict[interface] self.cli_set(base_path + [interface, 'authentication', 'username', user]) self.cli_set(base_path + [interface, 'authentication', 'password', passwd]) self.cli_set(base_path + [interface, 'mtu', mtu]) self.cli_set(base_path + [interface, 'no-peer-dns']) # check validate() - a source-interface is required with self.assertRaises(ConfigSessionError): self.cli_commit() self.cli_set(base_path + [interface, 'source-interface', self._source_interface]) # commit changes self.cli_commit() # verify configuration file(s) for interface in self._interfaces: self.assertTrue(wait_for_interface(interface), msg=f'Interface {interface} not found after {connect_timeout} seconds!') (user, passwd) = self.u_p_dict[interface] tmp = get_config_value(interface, 'mtu')[1] self.assertEqual(tmp, mtu) # MRU must default to MTU if not specified on CLI tmp = get_config_value(interface, 'mru')[1] self.assertEqual(tmp, mtu) tmp = get_config_value(interface, 'user')[1].replace('"', '') self.assertEqual(tmp, user) tmp = get_config_value(interface, 'password')[1].replace('"', '') self.assertEqual(tmp, passwd) tmp = get_config_value(interface, 'ifname')[1] self.assertEqual(tmp, interface) # Validate and verify assigned IP addresses self._verify_interface_address(interface) # validate that we have learned a default route tmp = self.getFRRopmode('show ip route 0.0.0.0/0', json=True) # Test if we have a default route 0.0.0.0/0 pointing to our PPPoE interface tmp = dict_search_recursive(tmp, 'interfaceName') #self.assertTrue(any(iface == interface for (iface, _) in tmp)) self.skipTest('Bug in FRR 10.2 - PPPoE interfaces sometimes carry ifIndex 0 which is invalid') def test_pppoe_client_disabled_interface(self): # Check if PPPoE Client can be disabled for interface in self._interfaces: (user, passwd) = self.u_p_dict[interface] self.cli_set(base_path + [interface, 'authentication', 'username', user]) self.cli_set(base_path + [interface, 'authentication', 'password', passwd]) self.cli_set(base_path + [interface, 'source-interface', self._source_interface]) self.cli_set(base_path + [interface, 'disable']) self.cli_commit() # Validate PPPoE client process - must not run as interfaces are disabled for interface in self._interfaces: running = False for proc in process_iter(): if interface in proc.cmdline(): running = True break self.assertFalse(running) # enable PPPoE interfaces for interface in self._interfaces: self.cli_delete(base_path + [interface, 'disable']) self.cli_commit() def test_pppoe_authentication(self): # When username or password is set - so must be the other for interface in self._interfaces: (user, passwd) = self.u_p_dict[interface] self.cli_set(base_path + [interface, 'address', 'dhcpv6']) self.cli_set(base_path + [interface, 'source-interface', self._source_interface]) self.cli_set(base_path + [interface, 'ipv6', 'address', 'autoconf']) self.cli_set(base_path + [interface, 'authentication', 'username', user]) # check validate() - if user is set, so must be the password with self.assertRaises(ConfigSessionError): self.cli_commit() self.cli_set(base_path + [interface, 'authentication', 'password', passwd]) self.cli_commit() for interface in self._interfaces: self.assertTrue(wait_for_interface(interface), msg=f'Interface {interface} not found after {connect_timeout} seconds!') # Validate and verify assigned IP addresses self._verify_interface_address(interface) def test_pppoe_dhcpv6pd(self): # Check if PPPoE dialer can be configured with DHCPv6-PD address = 1 sla_id = 0xff for interface in self._interfaces: (user, passwd) = self.u_p_dict[interface] interface_id = ''.join(c for c in interface if c.isdigit()) self.cli_set(base_path + [interface, 'authentication', 'username', user]) self.cli_set(base_path + [interface, 'authentication', 'password', passwd]) self.cli_set(base_path + [interface, 'no-default-route']) self.cli_set(base_path + [interface, 'no-peer-dns']) self.cli_set(base_path + [interface, 'source-interface', self._source_interface]) self.cli_set(base_path + [interface, 'ipv6', 'address', 'autoconf']) # interface we will delegate to delegate_if = f'dum{interface_id}' self.cli_set(['interfaces', 'dummy', delegate_if]) # prefix delegation stuff dhcpv6_pd_base = base_path + [interface, 'dhcpv6-options', 'pd', '0'] self.cli_set(dhcpv6_pd_base + ['length', '56']) self.cli_set(dhcpv6_pd_base + ['interface', delegate_if, 'address'], value=str(address)) self.cli_set(dhcpv6_pd_base + ['interface', delegate_if, 'sla-id'], value=str(sla_id)) # commit changes self.cli_commit() for interface in self._interfaces: self.assertTrue(wait_for_interface(interface), msg=f'Interface {interface} not found after {connect_timeout} seconds!') (user, passwd) = self.u_p_dict[interface] mtu_default = default_value(base_path + [interface, 'mtu']) tmp = get_config_value(interface, 'mtu')[1] self.assertEqual(tmp, mtu_default) tmp = get_config_value(interface, 'user')[1].replace('"', '') self.assertEqual(tmp, user) tmp = get_config_value(interface, 'password')[1].replace('"', '') self.assertEqual(tmp, passwd) tmp = get_config_value(interface, '+ipv6 ipv6cp-use-ipaddr') self.assertListEqual(tmp, ['+ipv6', 'ipv6cp-use-ipaddr']) # Validate and verify assigned IP addresses self._verify_interface_address(interface) # interface we delegated to delegate_if = f'dum{interface_id}' tmp = get_interface_address(delegate_if) self.assertIn('addr_info', tmp) # Verify IPv6 address received from out DHCPv6-PD for addr_info in tmp['addr_info']: if 'family' not in addr_info or addr_info['family'] != 'inet6': continue # Skip link-local interface address ipv6 = IPv6Address(addr_info['local']) if ipv6.is_link_local: continue # DHCPv6-PD assigned interface addres is of length /64 self.assertEqual(addr_info['prefixlen'], 64) # Interface IP address must be within the PD pool self.assertIn(ipv6, IPv6Network(ipv6_pool_pd)) # Get corresponding PD assigned prefix for this site/connection pd_prefix = IPv6Network(f"{ipv6}/56", strict=False) # Prefix must be within the PD pool self.assertTrue(pd_prefix.subnet_of(IPv6Network(ipv6_pool_pd))) gen_addr = calculate_ipv6_interface_address(pd_prefix, sla_id, address) self.assertEqual(gen_addr, ipv6) self.cli_delete(['interfaces', 'dummy', delegate_if]) def test_pppoe_options(self): # Verify access-concentrator and service-name CLI options ac_name: str = 'ACN123' service_name: str = 'VyOS' self.cli_set(pppoe_server_path + ['access-concentrator', ac_name]) self.cli_set(pppoe_server_path + ['service-name', service_name]) self.cli_commit() # as this tests uniqueness - we only use one interface in this test interface = self._interfaces[0] (user, passwd) = self.u_p_dict[interface] host_uniq = 'cafe010203' self.cli_set(base_path + [interface, 'authentication', 'username', user]) self.cli_set(base_path + [interface, 'authentication', 'password', passwd]) self.cli_set(base_path + [interface, 'source-interface', self._source_interface]) self.cli_set(base_path + [interface, 'access-concentrator', ac_name]) self.cli_set(base_path + [interface, 'service-name', service_name]) self.cli_set(base_path + [interface, 'host-uniq', host_uniq]) # commit changes self.cli_commit() self.assertTrue(wait_for_interface(interface), msg=f'Interface {interface} not found after {connect_timeout} seconds!') tmp = get_config_value(interface, 'pppoe-ac')[1] self.assertEqual(tmp, f'"{ac_name}"') tmp = get_config_value(interface, 'pppoe-service')[1] self.assertEqual(tmp, f'"{service_name}"') tmp = get_config_value(interface, 'pppoe-host-uniq')[1] self.assertEqual(tmp, f'"{host_uniq}"') # Validate and verify assigned IP addresses self._verify_interface_address(interface) self.cli_delete(pppoe_server_path + ['access-concentrator']) self.cli_delete(pppoe_server_path + ['service-name']) def test_pppoe_mtu_mru(self): # Check if PPPoE dialer can be configured and runs for interface in self._interfaces: (user, passwd) = self.u_p_dict[interface] mtu = '1400' mru = '1300' self.cli_set(base_path + [interface, 'authentication', 'username', user]) self.cli_set(base_path + [interface, 'authentication', 'password', passwd]) self.cli_set(base_path + [interface, 'mtu', mtu]) self.cli_set(base_path + [interface, 'mru', '9000']) # check validate() - a source-interface is required with self.assertRaises(ConfigSessionError): self.cli_commit() self.cli_set(base_path + [interface, 'source-interface', self._source_interface]) # check validate() - MRU needs to be less or equal then MTU with self.assertRaises(ConfigSessionError): self.cli_commit() self.cli_set(base_path + [interface, 'mru', mru]) # commit changes self.cli_commit() # verify configuration file(s) for interface in self._interfaces: self.assertTrue(wait_for_interface(interface), msg=f'Interface {interface} not found after {connect_timeout} seconds!') (user, passwd) = self.u_p_dict[interface] tmp = get_config_value(interface, 'mtu')[1] self.assertEqual(tmp, mtu) tmp = get_config_value(interface, 'mru')[1] self.assertEqual(tmp, mru) tmp = get_config_value(interface, 'user')[1].replace('"', '') self.assertEqual(tmp, user) tmp = get_config_value(interface, 'password')[1].replace('"', '') self.assertEqual(tmp, passwd) tmp = get_config_value(interface, 'ifname')[1] self.assertEqual(tmp, interface) # Validate and verify assigned IP addresses self._verify_interface_address(interface) def test_pppoe_default_route_survives_static_deletion(self): # T6991/T9054: The PPPoE default route must not be withdrawn from FRR # when "protocols static" is deleted - only the statically configured # routes must disappear, the PPPoE-sourced default route must stay. interface = self._interfaces[0] (user, passwd) = self.u_p_dict[interface] static_base_path = ['protocols', 'static'] self.cli_set(base_path + [interface, 'authentication', 'username', user]) self.cli_set(base_path + [interface, 'authentication', 'password', passwd]) self.cli_set(base_path + [interface, 'source-interface', self._source_interface]) self.cli_commit() self.assertTrue(wait_for_interface(interface), msg=f'Interface {interface} not found after {connect_timeout} seconds!') default_route = rf'ip route 0.0.0.0/0 {interface} tag 210' frrconfig = self.getFRRconfig('') self.assertIn(default_route, frrconfig) # Add an unrelated static route - this is what triggers "protocols # static" to exist on the CLI in the first place self.cli_set(static_base_path + ['route', '10.0.0.0/8', 'blackhole']) self.cli_commit() frrconfig = self.getFRRconfig('') self.assertIn(default_route, frrconfig) self.assertIn(r'ip route 10.0.0.0/8 blackhole', frrconfig) # Now delete "protocols static" entirely - the PPPoE default route # must remain in the FRR configuration (and thus in the RIB/FIB) self.cli_delete(static_base_path) self.cli_commit() frrconfig = self.getFRRconfig('') self.assertNotIn(r'ip route 10.0.0.0/8 blackhole', frrconfig) self.assertIn(default_route, frrconfig) if __name__ == '__main__': unittest.main(verbosity=2, failfast=VyOSUnitTestSHIM.TestCase.debug_on())