selftests: drv-net: add userns devmem RX test

Add userns_devmem.py, which mirrors nk_devmem.py but places the netkit
guest in a netns whose owning user_ns is non-init. ncdevmem is ran there
via nsenter so the bind-rx call is issued with creds that hold
CAP_NET_ADMIN only in the child user_ns.

Without the preceding GENL_UNS_ADMIN_PERM patch the test fails at
bind-rx with EPERM, but with the patch the transfer completes and tests
pass.

Signed-off-by: Bobby Eshleman <bobbyeshleman@meta.com>
Acked-by: Stanislav Fomichev <sdf@fomichev.me>
Link: https://patch.msgid.link/20260602-nl-prov-v2-2-ad721142c641@meta.com
Signed-off-by: Jakub Kicinski <kuba@kernel.org>

tools/testing/selftests/drivers/net/hw/Makefile

@@ -47,6 +47,7 @@ TEST_PROGS = \
 	rss_input_xfrm.py \
 	toeplitz.py \
 	tso.py \
+	userns_devmem.py \
 	uso.py \
 	xdp_metadata.py \
 	xsk_reconfig.py \

tools/testing/selftests/drivers/net/hw/config

@@ -20,5 +20,6 @@ CONFIG_NET_IPIP=y
 CONFIG_NETKIT=y
 CONFIG_NET_SCH_INGRESS=y
 CONFIG_UDMABUF=y
+CONFIG_USER_NS=y
 CONFIG_VXLAN=y
 CONFIG_XFRM_USER=y

tools/testing/selftests/drivers/net/hw/lib/py/__init__.py

@@ -18,7 +18,7 @@ try:
     sys.path.append(KSFT_DIR.as_posix())
 
     # Import one by one to avoid pylint false positives
-    from net.lib.py import NetNS, NetNSEnter, NetdevSimDev
+    from net.lib.py import NetNS, NetNSEnter, NetdevSimDev, UserNetNS
     from net.lib.py import EthtoolFamily, NetdevFamily, NetshaperFamily, \
         NlError, RtnlFamily, DevlinkFamily, PSPFamily, Netlink
     from net.lib.py import CmdExitFailure
@@ -34,7 +34,7 @@ try:
     from drivers.net.lib.py import GenerateTraffic, Remote, Iperf3Runner
     from drivers.net.lib.py import NetDrvEnv, NetDrvEpEnv, NetDrvContEnv
 
-    __all__ = ["NetNS", "NetNSEnter", "NetdevSimDev",
+    __all__ = ["NetNS", "NetNSEnter", "NetdevSimDev", "UserNetNS",
                "EthtoolFamily", "NetdevFamily", "NetshaperFamily",
                "NlError", "RtnlFamily", "DevlinkFamily", "PSPFamily", "Netlink",
                "CmdExitFailure",

tools/testing/selftests/drivers/net/hw/userns_devmem.py

@@ -0,0 +1,49 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: GPL-2.0
+"""
+Devmem tests for non-init userns.
+"""
+
+import os
+
+from devmem_lib import run_rx, run_rx_hds, run_tx, run_tx_chunks, setup_test
+from lib.py import NetDrvContEnv, ksft_disruptive, ksft_exit, ksft_run
+
+
+@ksft_disruptive
+def check_userns_rx(cfg) -> None:
+    """Run the devmem RX test through non-init userns netkit."""
+    run_rx(cfg)
+
+
+@ksft_disruptive
+def check_userns_tx(cfg) -> None:
+    """Run the devmem TX test through non-init userns netkit."""
+    run_tx(cfg)
+
+
+@ksft_disruptive
+def check_userns_tx_chunks(cfg) -> None:
+    """Run the devmem TX chunking test through non-init userns netkit."""
+    run_tx_chunks(cfg)
+
+
+def check_userns_rx_hds(cfg) -> None:
+    """Run the HDS test through non-init userns netkit."""
+    run_rx_hds(cfg)
+
+
+def main() -> None:
+    """Run userns devmem RX selftests against the test environment."""
+    with NetDrvContEnv(__file__, userns=True, rxqueues=2,
+                       primary_rx_redirect=True) as cfg:
+        setup_test(cfg,
+                   os.path.join(os.path.dirname(os.path.abspath(__file__)),
+                                "ncdevmem"))
+        ksft_run([check_userns_rx, check_userns_tx, check_userns_tx_chunks,
+                  check_userns_rx_hds], args=(cfg,))
+    ksft_exit()
+
+
+if __name__ == "__main__":
+    main()

tools/testing/selftests/drivers/net/lib/py/__init__.py

@@ -18,7 +18,7 @@ try:
     sys.path.append(KSFT_DIR.as_posix())
 
     # Import one by one to avoid pylint false positives
-    from net.lib.py import NetNS, NetNSEnter, NetdevSimDev
+    from net.lib.py import NetNS, NetNSEnter, NetdevSimDev, UserNetNS
     from net.lib.py import EthtoolFamily, NetdevFamily, NetshaperFamily, \
         NlError, RtnlFamily, DevlinkFamily, PSPFamily, Netlink
     from net.lib.py import CmdExitFailure
@@ -32,7 +32,7 @@ try:
     from net.lib.py import ksft_eq, ksft_ge, ksft_in, ksft_is, ksft_lt, \
         ksft_ne, ksft_not_in, ksft_raises, ksft_true, ksft_gt, ksft_not_none
 
-    __all__ = ["NetNS", "NetNSEnter", "NetdevSimDev",
+    __all__ = ["NetNS", "NetNSEnter", "NetdevSimDev", "UserNetNS",
                "EthtoolFamily", "NetdevFamily", "NetshaperFamily",
                "NlError", "RtnlFamily", "DevlinkFamily", "PSPFamily", "Netlink",
                "CmdExitFailure",

tools/testing/selftests/drivers/net/lib/py/env.py

@@ -9,7 +9,7 @@ from pathlib import Path
 from lib.py import KsftSkipEx, KsftXfailEx
 from lib.py import ksft_setup, wait_file
 from lib.py import cmd, ethtool, ip, CmdExitFailure
-from lib.py import NetNS, NetdevSimDev
+from lib.py import NetNS, NetdevSimDev, UserNetNS
 from .remote import Remote
 from . import bpftool, RtnlFamily, Netlink
 
@@ -337,8 +337,10 @@ class NetDrvContEnv(NetDrvEpEnv):
               +---------------+
     """
 
-    def __init__(self, src_path, rxqueues=1, primary_rx_redirect=False, **kwargs):
+    def __init__(self, src_path, rxqueues=1, primary_rx_redirect=False,
+                 userns=False, **kwargs):
         self.netns = None
+        self._userns = userns
         self._nk_host_ifname = None
         self.nk_guest_ifname = None
         self._tc_clsact_added = False
@@ -463,7 +465,7 @@ class NetDrvContEnv(NetDrvEpEnv):
         with open(ra_path, "w", encoding="utf-8") as f:
             f.write("2")
 
-        self.netns = NetNS()
+        self.netns = UserNetNS() if self._userns else NetNS()
         cmd("ip netns attach init 1")
         self._init_ns_attached = True
         ip("netns set init 0", ns=self.netns)

tools/testing/selftests/net/lib/py/__init__.py

@@ -10,7 +10,7 @@ from .ksft import KsftFailEx, KsftSkipEx, KsftXfailEx, ksft_pr, ksft_eq, \
     ksft_ge, ksft_gt, ksft_lt, ksft_raises, ksft_busy_wait, \
     ktap_result, ksft_disruptive, ksft_setup, ksft_run, ksft_exit, \
     ksft_variants, KsftNamedVariant
-from .netns import NetNS, NetNSEnter
+from .netns import NetNS, NetNSEnter, UserNetNS
 from .nsim import NetdevSim, NetdevSimDev
 from .utils import CmdExitFailure, fd_read_timeout, cmd, bkg, defer, \
     bpftool, ip, ethtool, bpftrace, rand_port, rand_ports, wait_port_listen, \
@@ -26,7 +26,7 @@ __all__ = ["KSRC",
            "ksft_is", "ksft_ge", "ksft_gt", "ksft_lt", "ksft_raises",
            "ksft_busy_wait", "ktap_result", "ksft_disruptive", "ksft_setup",
            "ksft_run", "ksft_exit", "ksft_variants", "KsftNamedVariant",
-           "NetNS", "NetNSEnter",
+           "NetNS", "NetNSEnter", "UserNetNS",
            "CmdExitFailure", "fd_read_timeout", "cmd", "bkg", "defer",
            "bpftool", "ip", "ethtool", "bpftrace", "rand_port", "rand_ports",
            "wait_port_listen", "wait_file", "tool", "tc",

tools/testing/selftests/net/lib/py/netns.py

@@ -1,9 +1,14 @@
 # SPDX-License-Identifier: GPL-2.0
 
-from .utils import ip
 import ctypes
+import os
 import random
 import string
+import subprocess
+import time
+from pathlib import Path
+
+from .utils import ip
 
 libc = ctypes.cdll.LoadLibrary('libc.so.6')
 
@@ -34,6 +39,74 @@ class NetNS:
         return f"NetNS({self.name})"
 
 
+class UserNetNS:
+    """Network namespace owned by a non-init user namespace."""
+
+    def __init__(self):
+        self.name = ''.join(
+            random.choice(string.ascii_lowercase) for _ in range(8))
+        self.user_ns_path = f"/run/userns/{self.name}"
+        self.net_ns_path = f"/run/netns/{self.name}"
+        self._user_mounted = False
+        self._net_mounted = False
+
+        os.makedirs("/run/userns", exist_ok=True)
+        os.makedirs("/run/netns", exist_ok=True)
+
+        Path(self.user_ns_path).touch()
+        Path(self.net_ns_path).touch()
+
+        with subprocess.Popen(
+                ["unshare", "--user", "--net", "--map-root-user",
+                 "sleep", "infinity"]) as proc:
+            try:
+                pid = proc.pid
+                init_user = os.readlink("/proc/self/ns/user")
+                for _ in range(200):
+                    try:
+                        if os.readlink(f"/proc/{pid}/ns/user") != init_user:
+                            break
+                    except OSError:
+                        pass
+                    time.sleep(0.01)
+                else:
+                    raise RuntimeError("unshare child did not create userns")
+
+                subprocess.run(["mount", "--bind", f"/proc/{pid}/ns/user",
+                                self.user_ns_path], check=True)
+                self._user_mounted = True
+                subprocess.run(["mount", "--bind", f"/proc/{pid}/ns/net",
+                                self.net_ns_path], check=True)
+                self._net_mounted = True
+            finally:
+                proc.kill()
+
+    def __del__(self):
+        if self._net_mounted:
+            subprocess.run(["umount", self.net_ns_path], check=False)
+            self._net_mounted = False
+        if self._user_mounted:
+            subprocess.run(["umount", self.user_ns_path], check=False)
+            self._user_mounted = False
+        for path in (self.net_ns_path, self.user_ns_path):
+            try:
+                os.unlink(path)
+            except OSError:
+                pass
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, ex_type, ex_value, ex_tb):
+        self.__del__()
+
+    def __str__(self):
+        return self.name
+
+    def __repr__(self):
+        return f"UserNetNS({self.name})"
+
+
 class NetNSEnter:
     def __init__(self, ns_name):
         self.ns_path = f"/run/netns/{ns_name}"

tools/testing/selftests/net/lib/py/utils.py

@@ -47,7 +47,12 @@ class cmd:
                  background=False, host=None, timeout=5, ksft_ready=None,
                  ksft_wait=None):
         if ns:
-            comm = f'ip netns exec {ns} ' + comm
+            if hasattr(ns, 'user_ns_path'):
+                comm = (f'nsenter --user={ns.user_ns_path} '
+                        f'--net={ns.net_ns_path} --setuid=0 --setgid=0 -- '
+                        + comm)
+            else:
+                comm = f'ip netns exec {ns} ' + comm
 
         self.stdout = None
         self.stderr = None