pseudo-ethernet: T8540: Add anycast-gateway support for EVPN

Introduce 'anycast-gateway' leafNode for pseudo-ethernet interfaces.
When set, a local FDB entry is installed on the parent bridge to
prevent the shared anycast MAC from leaking over the VXLAN overlay.

Author: alexandr-san4ez

interface-definitions/interfaces_pseudo-ethernet.xml.in

@@ -62,6 +62,12 @@
           #include <include/interface/redirect.xml.i>
           #include <include/interface/vif-s.xml.i>
           #include <include/interface/vif.xml.i>
+          <leafNode name="anycast-gateway">
+            <properties>
+              <help>Use the interface as EVPN Anycast Gateway</help>
+              <valueless/>
+            </properties>
+          </leafNode>
         </children>
       </tagNode>
     </children>

python/vyos/ifconfig/bridge.py

@@ -104,6 +104,12 @@ class BridgeIf(Interface):
         'del_port': {
             'shellcmd': 'ip link set dev {value} nomaster',
         },
+        'add_local_fdb_entry': {
+            'shellcmd': 'bridge fdb add {value} dev {ifname} self local',
+        },
+        'del_local_fdb_entry': {
+            'shellcmd': 'bridge fdb del {value} dev {ifname} self local',
+        },
     }}
 
     def _create(self):
@@ -272,6 +278,26 @@ def set_vlan_protocol(self, protocol):
 
         return self.set_interface('vlan_protocol', map[protocol])
 
+    def add_local_fdb_entry(self, mac_address: str):
+        """
+        Add a local FDB entry for the given MAC address on the bridge interface.
+
+        Example:
+        >>> from vyos.ifconfig import BridgeIf
+        >>> BridgeIf('br0').add_local_fdb_entry('cc:38:2e:bf:7b:0d')
+        """
+        self.set_interface('add_local_fdb_entry', mac_address.lower())
+
+    def del_local_fdb_entry(self, mac_address: str):
+        """
+        Remove a local FDB entry for the given MAC address on the bridge interface.
+
+        Example:
+        >>> from vyos.ifconfig import BridgeIf
+        >>> BridgeIf('br0').del_local_fdb_entry('cc:38:2e:bf:7b:0d')
+        """
+        self.set_interface('del_local_fdb_entry', mac_address.lower())
+
     def update(self, config):
         """ General helper function which works on a dictionary retrieved by
         get_config_dict(). It's main intention is to consolidate the scattered

python/vyos/ifconfig/macvlan.py

@@ -14,6 +14,7 @@
 # License along with this library.  If not, see <http://www.gnu.org/licenses/>.
 
 from vyos.ifconfig.interface import Interface
+from vyos.utils.network import get_interface_config
 
 @Interface.register
 class MACVLANIf(Interface):
@@ -43,3 +44,7 @@ def _create(self):
     def set_mode(self, mode):
         cmd = f'ip link set dev {self.ifname} type macvlan mode {mode}'
         return self._cmd(cmd)
+
+    def get_source_interface(self):
+        interface_config = get_interface_config(self.ifname)
+        return interface_config['link'] if interface_config is not None else None

python/vyos/utils/network.py

@@ -158,6 +158,41 @@ def get_vrf_tableid(interface: str):
         table = tmp['linkinfo']['info_slave_data']['table']
     return table
 
+
+def split_interface_vlans(interface: str) -> tuple:
+    """
+    Parse a interface name (with optional VLAN suffixes) into its
+    component parts.
+
+    Handles three input forms:
+      - 'br0'         -> root bridge interface only
+      - 'br0.100'     -> root bridge interface + one VLAN sub-interface level
+      - 'br0.100.200' -> root bridge interface + two VLAN sub-interface levels (QinQ)
+
+    Returns a tuple with the following parts on success:
+      (
+        'br0',  # root interface (always present)
+        '100',  # first VLAN suffix  (None if not present)
+        '200',  # second VLAN suffix (None if not present)
+      )
+    """
+
+    parts = interface.split('.')
+
+    # Guard: we support at most bridge + 2 VLAN levels (e.g. br0.100.200)
+    if len(parts) > 3:
+        raise ValueError(
+            f'Interface "{interface}" has too many VLAN suffixes. '
+            'Only up to two levels are supported (e.g. br0.100.200).'
+        )
+
+    iface = parts[0]
+    vlan_id = parts[1] if len(parts) >= 2 else None
+    inner_vlan_id = parts[2] if len(parts) == 3 else None
+
+    return iface, vlan_id, inner_vlan_id
+
+
 def get_interface_config(interface):
     """ Returns the used encapsulation protocol for given interface.
         If interface does not exist, None is returned.

smoketest/scripts/cli/test_interfaces_pseudo-ethernet.py

@@ -19,6 +19,8 @@
 
 from base_interfaces_test import BasicInterfaceTest
 from base_vyostest_shim import VyOSUnitTestSHIM
+from vyos.configsession import ConfigSessionError
+from vyos.utils.process import cmd
 
 from vyos.ifconfig import Section
 
@@ -44,5 +46,55 @@ def setUpClass(cls):
         # call base-classes classmethod
         super(PEthInterfaceTest, cls).setUpClass()
 
+    def test_anycast_gateway(self):
+        # Create the underlying bridge and sub-interface in the test
+
+        for i, peth in enumerate(self._interfaces):
+            eth = peth[1:]  # Convert peth0 -> eth0
+            br = f'br{i}'
+            vlan = str(i + 100)
+
+            # Format the MAC using index as a two-digit hexadecimal
+            mac_address = f'00:aa:aa:aa:aa:{i:02x}'
+
+            with self.subTest(peth=peth, eth=eth, mac_address=mac_address, i=i):
+                base_bridge_path = ['interfaces', 'bridge', br]
+                base_br_member_path = base_bridge_path + ['member', 'interface']
+
+                self.cli_set(base_bridge_path + ['enable-vlan'])
+                self.cli_set(base_br_member_path + [eth, 'native-vlan', vlan])
+                self.cli_set(base_br_member_path + [f'vxlan{i}'])
+                self.cli_set(base_bridge_path + ['vif', vlan])
+
+                self.cli_set(
+                    self._base_path + [peth, 'source-interface', f'{br}.{vlan}']
+                )
+                self.cli_set(self._base_path + [peth, 'anycast-gateway'])
+
+                # Anycast gateway requires MAC
+                with self.assertRaises(ConfigSessionError):
+                    self.cli_commit()
+
+                self.cli_set(self._base_path + [peth, 'mac', mac_address])
+                self.cli_commit()
+
+                # Verify FDB entry exists with flag
+                fdb = cmd(f'bridge fdb show dev {br}')
+                self.assertIn(f'{mac_address} master {br} permanent', fdb)
+                self.assertIn(f'{mac_address} self permanent', fdb)
+
+                # Then remove just the anycast-gateway flag
+                self.cli_delete(self._base_path + [peth, 'anycast-gateway'])
+                self.cli_commit()
+
+                fdb = cmd(f'bridge fdb show dev {br}')
+                self.assertNotIn(f'{mac_address} master {br} permanent', fdb)
+
+                # Clean up temp bridge and peth
+                self.cli_delete(self._base_path + [peth])
+                self.cli_delete(base_bridge_path)
+                self.cli_commit()
+
+
 if __name__ == '__main__':
     unittest.main(verbosity=2, failfast=VyOSUnitTestSHIM.TestCase.debug_on())

src/conf_mode/interfaces_pseudo-ethernet.py

@@ -31,7 +31,9 @@
 from vyos.configverify import verify_mtu_ipv6
 from vyos.configverify import verify_mirror_redirect
 from vyos.ifconfig import MACVLANIf
+from vyos.ifconfig import BridgeIf
 from vyos.utils.network import interface_exists
+from vyos.utils.network import split_interface_vlans
 from vyos import ConfigError
 
 from vyos import airbag
@@ -68,6 +70,35 @@ def get_config(config=None):
 
     return peth
 
+def _verify_anycast_gateway(peth: dict):
+    """Validate anycast-gateway requirements."""
+
+    if 'anycast_gateway' not in peth:
+        return
+
+    ifname = peth['ifname']
+
+    # Requirement 1: MAC address must be explicitly configured
+    if 'mac' not in peth:
+        raise ConfigError(
+            f'Anycast-gateway requires an explicit MAC address to be set on interface {ifname}. '
+            f'Use: set interfaces pseudo-ethernet {ifname} mac <mac>'
+        )
+
+    # Requirement 2: source-interface must be a bridge or bridge sub-interface
+    source_iface = peth.get('source_interface')
+    if not source_iface:
+        raise ConfigError(
+            f'Anycast-gateway requires source-interface to be set on interface {ifname}'
+        )
+
+    if not source_iface.startswith('br'):
+        raise ConfigError(
+            'Anycast-gateway requires source-interface to be a bridge '
+            'or a bridge vlan interface (e.g. br0 or br0.100), but '
+            f'"{source_iface}" is neither of these two.'
+        )
+
 def verify(peth):
     if 'deleted' in peth:
         verify_bridge_delete(peth)
@@ -82,12 +113,75 @@ def verify(peth):
     # use common function to verify VLAN configuration
     verify_vlan_config(peth)
 
+    _verify_anycast_gateway(peth)
+
     return None
 
 def generate(peth):
     return None
 
+
+def _apply_anycast_gateway(peth: dict, old_mac: str | None, old_source: str | None):
+    """Apply anycast gateway FDB entry for bridge if configured."""
+
+    def _get_bridge_by_source(source_interface: str | None):
+        source_interface = source_interface or ''
+        if source_interface.startswith('br'):
+            bridge_ifname, *_ = split_interface_vlans(source_interface)
+            if interface_exists(bridge_ifname):
+                return BridgeIf(bridge_ifname)
+        return None
+
+    # Is anycast-gateway set in the new config?
+    is_anycast = 'anycast_gateway' in peth
+
+    # Was anycast-gateway previously active?
+    # (old entry exists if both old_mac and old_source set)
+    was_anycast = old_mac is not None and old_source is not None
+
+    # Detect a MAC address change while anycast-gateway stays enabled
+    new_mac = peth.get('mac')
+    mac_changed = is_anycast and old_mac and old_mac != new_mac
+
+    # Detect a source interface change while anycast-gateway stays enabled
+    new_source = peth.get('source_interface')
+    source_changed = is_anycast and old_source and old_source != new_source
+
+    # Clean up old FDB entry in the next situations:
+    #   1. anycast-gateway was removed
+    #   2. the MAC address changed (old entry is now stale)
+    #   3. the source interface changed (old entry is now stale)
+    #   4. the interface is being deleted
+    if not is_anycast or mac_changed or source_changed or 'deleted' in peth:
+        if old_source and old_mac:
+            bridge = _get_bridge_by_source(old_source)
+            if bridge:
+                try:
+                    bridge.del_local_fdb_entry(old_mac)
+                except OSError:
+                    pass  # Bridge may already be gone, that is fine
+
+    if 'deleted' not in peth:
+        # Add only on transition or key change
+        should_add = (is_anycast and not was_anycast) or mac_changed or source_changed
+        if should_add and new_source and new_mac:
+            bridge = _get_bridge_by_source(new_source)
+            if bridge:
+                bridge.add_local_fdb_entry(new_mac)
+
+
 def apply(peth):
+    # Obtain current MAC address and source interface if pseudo-ethernet exists.
+    # It is neccessary to apply anycast gateway.
+    current_mac = current_source = None
+    if 'deleted' not in peth:
+        peth_ifname = peth['ifname']
+        if interface_exists(peth_ifname):
+            piface = MACVLANIf(peth_ifname, create=False)
+            if piface:
+                current_mac = piface.get_mac()
+                current_source = piface.get_source_interface()
+
     # Check if the MACVLAN interface already exists
     if 'rebuild_required' in peth or 'deleted' in peth:
         if interface_exists(peth['ifname']):
@@ -100,6 +194,8 @@ def apply(peth):
         p = MACVLANIf(**peth)
         p.update(peth)
 
+    _apply_anycast_gateway(peth, current_mac, current_source)
+
     if 'static_arp' in peth:
         call_dependents()