Merge pull request #5021 from alexandr-san4ez/T8276-current

vpp: T8276: Add verification for virtual interfaces in PPPoE server configuration

2 files changed, 69 insertions, 7 deletions

diff --git a/python/vyos/configverify.py b/python/vyos/configverify.py
index 177c4f9b0..5ff8dda2c 100644
--- a/python/vyos/configverify.py
+++ b/python/vyos/configverify.py
@@ -22,7 +22,11 @@
 # makes use of it!
 
 from vyos import ConfigError
+from vyos.base import Warning
 from vyos.utils.dict import dict_search
+from vyos.utils.dict import dict_search_recursive
+from vyos.utils.network import interface_exists
+
 # pattern re-used in ipsec migration script
 dynamic_interface_pattern = r'(ppp|pppoe|sstpc|l2tp|ipoe)[0-9]+'
 
@@ -100,7 +104,6 @@ def verify_vrf(config):
     Common helper function used by interface implementations to perform
     recurring validation of VRF configuration.
     """
-    from vyos.utils.network import interface_exists
     if 'vrf' in config:
         vrfs = config['vrf']
         if isinstance(vrfs, str):
@@ -177,7 +180,6 @@ def verify_mirror_redirect(config):
 
     It makes no sense to mirror traffic back at yourself!
     """
-    from vyos.utils.network import interface_exists
     if {'mirror', 'redirect'} <= set(config):
         raise ConfigError('Mirror and redirect can not be enabled at the same time!')
 
@@ -247,10 +249,6 @@ def verify_interface_exists(config, ifname, state_required=False, warning_only=F
     if the interface is defined on the CLI, if it's not found we try if
     it exists at the OS level.
     """
-    from vyos.base import Warning
-    from vyos.utils.dict import dict_search_recursive
-    from vyos.utils.network import interface_exists
-
     if not state_required:
         # Check if interface is present in CLI config
         tmp = getattr(config, 'interfaces_root', {})
@@ -267,6 +265,47 @@ def verify_interface_exists(config, ifname, state_required=False, warning_only=F
         return False
     raise ConfigError(message)
 
+def verify_virtual_interface_exists(
+    config, ifname, state_required=False, warning_only=False
+):
+    """
+    Verify the existence of a virtual network interface in the configuration or the Linux kernel.
+
+    This function checks whether a specified virtual interface exists in the provided configuration
+    or in the Linux kernel. It can return a warning or raise an error based on the parameters provided.
+    """
+    physical_ifname, vif_id = ifname.split('.', maxsplit=1)
+
+    if vif_id and '.' in vif_id:
+        vif_s, vif_c = vif_id.split('.', maxsplit=1)
+        vif_id = None
+    else:
+        vif_s = vif_c = None
+
+    if not state_required:
+        # Check if sub-interface is present in CLI config
+        interfaces_root = getattr(config, 'interfaces_root', {})
+
+        if vif_s and vif_c:
+            path = ['ethernet', physical_ifname, 'vif-s', vif_s, 'vif-c']
+            key = vif_c
+        else:
+            path = ['ethernet', physical_ifname, 'vif']
+            key = vif_id
+
+        if bool(list(dict_search_recursive(interfaces_root, key, path=path))):
+            return True
+
+    # Interface not found on CLI, try Linux Kernel
+    if interface_exists(ifname):
+        return True
+
+    message = f'Virtual Interface "{ifname}" does not exist!'
+    if warning_only:
+        Warning(message)
+        return False
+    raise ConfigError(message)
+
 def verify_source_interface(config):
     """
     Common helper function used by interface implementations to
@@ -274,7 +313,6 @@ def verify_source_interface(config):
     required by e.g. peth/MACvlan, MACsec ...
     """
     import re
-    from vyos.utils.network import interface_exists
 
     ifname = config['ifname']
     if 'source_interface' not in config:
diff --git a/python/vyos/vpp/control_vpp.py b/python/vyos/vpp/control_vpp.py
index 3fab4039a..b37bb865f 100644
--- a/python/vyos/vpp/control_vpp.py
+++ b/python/vyos/vpp/control_vpp.py
@@ -477,6 +477,30 @@ class VPPControl:
             )
 
     @_Decorators.api_call
+    def get_pppoe_interface_mapping(self) -> dict:
+        """
+        Get PPPoE mapping between data-plain and control-plane interfaces
+
+        Returns:
+            dict: Dict where key is the dataplane interface and value is the control interface
+        """
+
+        reply = self.cli_cmd('show pppoe control-plane binding').reply
+
+        if 'No PPPoE control plane interface configured' in reply:
+            return {}
+
+        lines = reply.splitlines()
+        lines = lines[2:]  # Ignore two lines with headers
+
+        result = {}
+        for line in lines:
+            key, value = line.split()
+            result[key] = value
+
+        return result
+
+    @_Decorators.api_call
     def enable_dhcp_client(self, ifname: str) -> None:
         """Enable DHCP client detection on a given interface