[PATCH net v6 3/4] selftests: Add MACsec VLAN propagation traffic test

[Cosmin Ratiu <cratiu@nvidia.com>]

Add macsec_traffic.py using NetDrvEpEnv to verify VLAN filter
propagation through offloaded MACsec devices via actual traffic.

The test creates MACsec tunnels with matching SAs on both endpoints,
stacks VLANs on top, and verifies connectivity with ping. Covers:
- Offloaded MACsec with VLAN (filters propagate to HW)
- Software MACsec with VLAN (no HW filter propagation)
- Toggle offload on/off and verify traffic still works

On netdevsim this makes use of the VLAN filter debugfs file to actually
validate that filters are applied/removed correctly.
On real hardware the traffic should validate actual VLAN filter
propagation.

Signed-off-by: Cosmin Ratiu <cratiu@nvidia.com>
---
 tools/testing/selftests/drivers/net/config    |   1 +
 .../selftests/drivers/net/lib/py/env.py       |   9 ++
 tools/testing/selftests/drivers/net/macsec.py | 151 ++++++++++++++++++
 3 files changed, 161 insertions(+)

diff --git a/tools/testing/selftests/drivers/net/config b/tools/testing/selftests/drivers/net/config
index d4b31a317c09..fd16994366f4 100644
--- a/tools/testing/selftests/drivers/net/config
+++ b/tools/testing/selftests/drivers/net/config
@@ -8,4 +8,5 @@ CONFIG_NETCONSOLE=m
 CONFIG_NETCONSOLE_DYNAMIC=y
 CONFIG_NETCONSOLE_EXTENDED_LOG=y
 CONFIG_NETDEVSIM=m
+CONFIG_VLAN_8021Q=m
 CONFIG_XDP_SOCKETS=y
diff --git a/tools/testing/selftests/drivers/net/lib/py/env.py b/tools/testing/selftests/drivers/net/lib/py/env.py
index 41cc248ac848..b80666277e36 100644
--- a/tools/testing/selftests/drivers/net/lib/py/env.py
+++ b/tools/testing/selftests/drivers/net/lib/py/env.py
@@ -255,6 +255,15 @@ class NetDrvEpEnv(NetDrvEnvBase):
         if nsim_test is False and self._ns is not None:
             raise KsftXfailEx("Test does not work on netdevsim")
 
+    def get_local_nsim_dev(self):
+        """Returns the local netdevsim device or None.
+           Using this method is discouraged, as it makes tests nsim-specific.
+           Standard interfaces available on all HW should ideally be used.
+           This method is intended for the few cases where nsim-specific
+           assertions need to be verified which cannot be verified otherwise.
+        """
+        return self._ns
+
     def _require_cmd(self, comm, key, host=None):
         cached = self._required_cmd.get(comm, {})
         if cached.get(key) is None:
diff --git a/tools/testing/selftests/drivers/net/macsec.py b/tools/testing/selftests/drivers/net/macsec.py
index a17b9f7ef584..b634d55e38c2 100755
--- a/tools/testing/selftests/drivers/net/macsec.py
+++ b/tools/testing/selftests/drivers/net/macsec.py
@@ -6,10 +6,14 @@
 import os
 
 from lib.py import ksft_run, ksft_exit, ksft_eq, ksft_raises
+from lib.py import ksft_variants, KsftNamedVariant
 from lib.py import CmdExitFailure, KsftSkipEx
 from lib.py import NetDrvEpEnv
 from lib.py import cmd, ip, defer, ethtool
 
+MACSEC_KEY = "12345678901234567890123456789012"
+MACSEC_VLAN_VID = 10
+
 # Unique prefix per run to avoid collisions in the shared netns.
 # Keep it short: IFNAMSIZ is 16 (incl. NUL), and VLAN names append ".<vid>".
 MACSEC_PFX = f"ms{os.getpid()}_"
@@ -25,6 +29,16 @@ def _get_macsec_offload(dev):
     return info.get("linkinfo", {}).get("info_data", {}).get("offload")
 
 
+def _require_ip_macsec(cfg):
+    """SKIP if iproute2 on local or remote lacks 'ip macsec' support."""
+    for host in [None, cfg.remote]:
+        out = cmd("ip macsec help", fail=False, host=host)
+        if "macsec" not in out.stdout + out.stderr:
+            where = "remote" if host else "local"
+            raise KsftSkipEx(f"iproute2 too old on {where},"
+                             " missing macsec support")
+
+
 def _require_ip_macsec_offload():
     """SKIP if local iproute2 doesn't understand 'ip macsec offload'."""
     out = cmd("ip macsec help", fail=False)
@@ -44,6 +58,78 @@ def _require_macsec_offload(cfg):
         raise KsftSkipEx("macsec-hw-offload not supported")
 
 
+def _get_mac(ifname, host=None):
+    """Gets MAC address of an interface."""
+    dev = ip(f"-d link show dev {ifname}", json=True, host=host)
+    return dev[0]["address"]
+
+
+def _setup_macsec_sa(cfg, name):
+    """Adds matching TX/RX SAs on both ends."""
+    local_mac = _get_mac(name)
+    remote_mac = _get_mac(name, host=cfg.remote)
+
+    ip(f"macsec add {name} tx sa 0 pn 1 on key 01 {MACSEC_KEY}")
+    ip(f"macsec add {name} rx port 1 address {remote_mac}")
+    ip(f"macsec add {name} rx port 1 address {remote_mac} "
+       f"sa 0 pn 1 on key 02 {MACSEC_KEY}")
+
+    ip(f"macsec add {name} tx sa 0 pn 1 on key 02 {MACSEC_KEY}",
+       host=cfg.remote)
+    ip(f"macsec add {name} rx port 1 address {local_mac}", host=cfg.remote)
+    ip(f"macsec add {name} rx port 1 address {local_mac} "
+       f"sa 0 pn 1 on key 01 {MACSEC_KEY}", host=cfg.remote)
+
+
+def _setup_macsec_devs(cfg, name, offload):
+    """Creates macsec devices on both ends.
+
+    Only the local device gets HW offload; the remote always uses software
+    MACsec since it may not support offload at all.
+    """
+    offload_arg = "mac" if offload else "off"
+
+    ip(f"link add link {cfg.ifname} {name} "
+       f"type macsec encrypt on offload {offload_arg}")
+    defer(ip, f"link del {name}")
+    ip(f"link add link {cfg.remote_ifname} {name} "
+       f"type macsec encrypt on", host=cfg.remote)
+    defer(ip, f"link del {name}", host=cfg.remote)
+
+
+def _set_offload(name, offload):
+    """Sets offload on the local macsec device only."""
+    offload_arg = "mac" if offload else "off"
+
+    ip(f"link set {name} type macsec encrypt on offload {offload_arg}")
+
+
+def _setup_vlans(cfg, name, vid):
+    """Adds VLANs on top of existing macsec devs."""
+    vlan_name = f"{name}.{vid}"
+
+    ip(f"link add link {name} {vlan_name} type vlan id {vid}")
+    defer(ip, f"link del {vlan_name}")
+    ip(f"link add link {name} {vlan_name} type vlan id {vid}", host=cfg.remote)
+    defer(ip, f"link del {vlan_name}", host=cfg.remote)
+
+
+def _setup_vlan_ips(cfg, name, vid):
+    """Adds VLANs and IPs and brings up the macsec + VLAN devices."""
+    local_ip = "198.51.100.1"
+    remote_ip = "198.51.100.2"
+    vlan_name = f"{name}.{vid}"
+
+    ip(f"addr add {local_ip}/24 dev {vlan_name}")
+    ip(f"addr add {remote_ip}/24 dev {vlan_name}", host=cfg.remote)
+    ip(f"link set {name} up")
+    ip(f"link set {name} up", host=cfg.remote)
+    ip(f"link set {vlan_name} up")
+    ip(f"link set {vlan_name} up", host=cfg.remote)
+
+    return vlan_name, remote_ip
+
+
 def test_offload_api(cfg) -> None:
     """MACsec offload API: create SecY, add SA/rx, toggle offload."""
 
@@ -164,6 +250,69 @@ def test_offload_state(cfg) -> None:
             "offload enabled after create: should be mac")
 
 
+def _check_nsim_vid(cfg, vid, expected) -> None:
+    """Checks if a VLAN is present. Only works on netdevsim."""
+
+    nsim = cfg.get_local_nsim_dev()
+    if not nsim:
+        return
+
+    vlan_path = nsim.nsims[0].dfs_dir + "vlan"
+    with open(vlan_path, encoding="utf-8") as f:
+        vids = f.read()
+    found = f"ctag {vid}\n" in vids
+    ksft_eq(found, expected,
+            f"VLAN {vid} {'expected' if expected else 'not expected'}"
+            f" in debugfs")
+
+
+@ksft_variants([
+    KsftNamedVariant("offloaded", True),
+    KsftNamedVariant("software", False),
+])
+def test_vlan(cfg, offload) -> None:
+    """Ping through VLAN-over-macsec."""
+
+    _require_ip_macsec(cfg)
+    if offload:
+        _require_macsec_offload(cfg)
+    else:
+        _require_ip_macsec_offload()
+    name = _macsec_name()
+    _setup_macsec_devs(cfg, name, offload=offload)
+    _setup_macsec_sa(cfg, name)
+    _setup_vlans(cfg, name, MACSEC_VLAN_VID)
+    vlan_name, remote_ip = _setup_vlan_ips(cfg, name, MACSEC_VLAN_VID)
+    _check_nsim_vid(cfg, MACSEC_VLAN_VID, offload)
+    # nsim doesn't handle the data path for offloaded macsec, so skip
+    # the ping when offloaded on nsim.
+    if not offload or not cfg.get_local_nsim_dev():
+        cmd(f"ping -I {vlan_name} -c 1 -W 5 {remote_ip}")
+
+
+@ksft_variants([
+    KsftNamedVariant("on_to_off", True),
+    KsftNamedVariant("off_to_on", False),
+])
+def test_vlan_toggle(cfg, offload) -> None:
+    """Toggle offload: VLAN filters propagate/remove correctly."""
+
+    _require_ip_macsec(cfg)
+    _require_macsec_offload(cfg)
+    name = _macsec_name()
+    _setup_macsec_devs(cfg, name, offload=offload)
+    _setup_vlans(cfg, name, MACSEC_VLAN_VID)
+    _check_nsim_vid(cfg, MACSEC_VLAN_VID, offload)
+    _set_offload(name, offload=not offload)
+    _check_nsim_vid(cfg, MACSEC_VLAN_VID, not offload)
+    vlan_name, remote_ip = _setup_vlan_ips(cfg, name, MACSEC_VLAN_VID)
+    _setup_macsec_sa(cfg, name)
+    # nsim doesn't handle the data path for offloaded macsec, so skip
+    # the ping when the final state is offloaded on nsim.
+    if offload or not cfg.get_local_nsim_dev():
+        cmd(f"ping -I {vlan_name} -c 1 -W 5 {remote_ip}")
+
+
 def main() -> None:
     """Main program."""
     with NetDrvEpEnv(__file__) as cfg:
@@ -171,6 +320,8 @@ def main() -> None:
                   test_max_secy,
                   test_max_sc,
                   test_offload_state,
+                  test_vlan,
+                  test_vlan_toggle,
                   ], args=(cfg,))
     ksft_exit()
 
-- 
2.53.0