smoketest: T7858: add PPPoE client tests with IPv4, IPv6 and DHCPv6-PD

VyOS includes a full-featured PPPoE server (BRAS), but it was previously not
exercised during embedded platform smoketests. This commit extends the
smoketest suite to include a basic PPPoE server configuration.

The test starts a local PPPoE server instance that provides both IPv4 and IPv6
addresses, including DHCPv6-PD for prefix delegation. The client side attempts
to establish a PPPoE session and verifies that the assigned addresses and
prefixes are within the expected configured pools. Connection is established
through virtual-ethernet interface pairs.

This helps ensure that core PPPoE functionality works correctly in the base
system image and catches regressions early.

2 files changed, 218 insertions, 57 deletions

diff --git a/smoketest/scripts/cli/base_vyostest_shim.py b/smoketest/scripts/cli/base_vyostest_shim.py
index ab372d361..9f4656dfc 100644
--- a/smoketest/scripts/cli/base_vyostest_shim.py
+++ b/smoketest/scripts/cli/base_vyostest_shim.py
@@ -160,6 +160,17 @@ class VyOSUnitTestSHIM:
                     print(f'FRR configuration still empty after {empty_retry} retires!')
             return out
 
+        def getFRRopmode(self, command : str, json : bool=False):
+            from json import loads
+            if json: command += f' json'
+            out = cmd(f'vtysh -c "{command}"')
+            if json:
+                out = loads(out)
+            if self.debug:
+                print(f'\n\ncommand "{command}" returned:\n')
+                pprint.pprint(out)
+            return out
+
         @staticmethod
         def ssh_send_cmd(command, username, password, key_filename=None,
                          hostname='localhost'):
diff --git a/smoketest/scripts/cli/test_interfaces_pppoe.py b/smoketest/scripts/cli/test_interfaces_pppoe.py
index cce08a29a..227a01032 100755
--- a/smoketest/scripts/cli/test_interfaces_pppoe.py
+++ b/smoketest/scripts/cli/test_interfaces_pppoe.py
@@ -17,13 +17,26 @@
 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 = '/etc/ppp/peers/{}'
-base_path = ['interfaces', 'pppoe']
+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 get_config_value(interface, key):
     with open(config_file.format(interface), 'r') as f:
@@ -32,6 +45,19 @@ def get_config_value(interface, 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
@@ -40,29 +66,76 @@ class PPPoEInterfaceTest(VyOSUnitTestSHIM.TestCase):
         # 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 = 'eth0'
+        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)
 
-    def tearDown(self):
-        # Validate PPPoE client process
-        for interface in self._interfaces:
-            running = False
-            for proc in process_iter():
-                if interface in proc.cmdline():
-                    running = True
-                    break
-            self.assertTrue(running)
+    @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()
 
+    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 = f'VyOS-user-{interface}'
-            passwd = f'VyOS-passwd-{interface}'
-            mtu = '1400'
+            (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])
@@ -79,8 +152,10 @@ class PPPoEInterfaceTest(VyOSUnitTestSHIM.TestCase):
 
         # verify configuration file(s)
         for interface in self._interfaces:
-            user = f'VyOS-user-{interface}'
-            passwd = f'VyOS-passwd-{interface}'
+            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)
@@ -94,11 +169,19 @@ class PPPoEInterfaceTest(VyOSUnitTestSHIM.TestCase):
             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))
+
     def test_pppoe_client_disabled_interface(self):
         # Check if PPPoE Client can be disabled
         for interface in self._interfaces:
-            user = f'VyOS-user-{interface}'
-            passwd = f'VyOS-passwd-{interface}'
+            (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])
@@ -122,17 +205,15 @@ class PPPoEInterfaceTest(VyOSUnitTestSHIM.TestCase):
 
         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 = f'VyOS-user-{interface}'
-            passwd = f'VyOS-passwd-{interface}'
+            (user, passwd) = self.u_p_dict[interface]
 
             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()
@@ -141,15 +222,23 @@ class PPPoEInterfaceTest(VyOSUnitTestSHIM.TestCase):
 
         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 = '0'
-        sla_len = '8'
+
+        # verify IPv6 DHCPv6-PD assignment on one dialer interface only - to not mix things up
 
         for interface in self._interfaces:
-            user = f'VyOS-user-{interface}'
-            passwd = f'VyOS-passwd-{interface}'
+            (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])
@@ -158,18 +247,29 @@ class PPPoEInterfaceTest(VyOSUnitTestSHIM.TestCase):
             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', self._source_interface, 'address', address])
-            self.cli_set(dhcpv6_pd_base + ['interface', self._source_interface, 'sla-id',  sla_id])
+            self.cli_set(dhcpv6_pd_base + ['interface', delegate_if, 'address', address])
+            self.cli_set(dhcpv6_pd_base + ['interface', delegate_if, 'sla-id',  sla_id])
+
+            address = str(int(address) + 1)
+            sla_id = str(int(sla_id) + 2)
 
         # commit changes
         self.cli_commit()
 
+        address = '1'
+        sla_id = '0'
         for interface in self._interfaces:
-            user = f'VyOS-user-{interface}'
-            passwd = f'VyOS-passwd-{interface}'
+            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]
@@ -181,43 +281,88 @@ class PPPoEInterfaceTest(VyOSUnitTestSHIM.TestCase):
             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.assertIn(pd_prefix, IPv6Network(ipv6_pool_pd))
+                # Calculate the assigned IP address from the prefix delegation
+                network_addr = str(pd_prefix.network_address) # 2001:db8:8003::
+                generated_sla_addr = network_addr[:-1] + f'{sla_id}::{address}'
+                self.assertEqual(generated_sla_addr, ipv6)
+
+                address = str(int(address) + 1)
+                sla_id = str(int(sla_id) + 2)
+
+            self.cli_delete(['interfaces', 'dummy', delegate_if])
+
     def test_pppoe_options(self):
-        # Check if PPPoE dialer can be configured with DHCPv6-PD
-        for interface in self._interfaces:
-            user = f'VyOS-user-{interface}'
-            passwd = f'VyOS-passwd-{interface}'
-            ac_name = f'AC{interface}'
-            service_name = f'SRV{interface}'
-            host_uniq = 'cafebeefBABE123456'
+        # Verify access-concentrator and service-name CLI options
 
-            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])
+        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])
+        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()
 
-        for interface in self._interfaces:
-            ac_name = f'AC{interface}'
-            service_name = f'SRV{interface}'
-            host_uniq = 'cafebeefBABE123456'
+        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)
 
-            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}"')
+        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 = f'VyOS-user-{interface}'
-            passwd = f'VyOS-passwd-{interface}'
+            (user, passwd) = self.u_p_dict[interface]
             mtu = '1400'
             mru = '1300'
 
@@ -241,8 +386,10 @@ class PPPoEInterfaceTest(VyOSUnitTestSHIM.TestCase):
 
         # verify configuration file(s)
         for interface in self._interfaces:
-            user = f'VyOS-user-{interface}'
-            passwd = f'VyOS-passwd-{interface}'
+            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)
@@ -255,5 +402,8 @@ class PPPoEInterfaceTest(VyOSUnitTestSHIM.TestCase):
             tmp = get_config_value(interface, 'ifname')[1]
             self.assertEqual(tmp, interface)
 
+            # Validate and verify assigned IP addresses
+            self._verify_interface_address(interface)
+
 if __name__ == '__main__':
     unittest.main(verbosity=2, failfast=VyOSUnitTestSHIM.TestCase.debug_on())