From fcc9fe84f4606c1bf95da46fba4e86ac0252419b Mon Sep 17 00:00:00 2001
From: Tobias Waldekranz <tobias@waldekranz.com>
Date: Fri, 9 Feb 2024 16:13:25 +0100
Subject: [PATCH] test/infamy: Support calling arbitrary functions from within
 a netns

Add a helper that lets you call a Python function (or lambda) in the
context of a network samespace.

Example:

    with infamy.IsolatedMacVlan("eth0") as ns:
        res = ns.call(lambda: subprocess.run(["ip", "link"],
	                                     capture_output=True))
	print(res.stdout, f"\n[exitcode:{res.returncode}]")

    The call to subprocess.run will be executed in the context of the
    network namespace, and thus only list "lo" and the "iface"
    macvlan.

The return value of the function passed to ns.call() is passed back
over a multiprocessing.Pipe, which requires that the object be
"picklable". This is true for most objects (even more complex ones
like the CompletedProcess object seen in the example above). Some
notable exceptions are things like file objects, sockets, etc.

The setns(2) syscall is unfortunately not available in current
versions of Python shipped with neither Ubuntu nor Alpine at the time
of this writing. Therefore, shoot from the hip, assume that we're
running on an x86_64 CPU, and just hotwire the syscall directly with
some constants we found in a dumpster.
---
 test/infamy/netns.py | 36 ++++++++++++++++++++++++++++++++++++
 1 file changed, 36 insertions(+)

diff --git a/test/infamy/netns.py b/test/infamy/netns.py
index 5c3cd618..9b83678f 100644
--- a/test/infamy/netns.py
+++ b/test/infamy/netns.py
@@ -1,4 +1,16 @@
+import ctypes
+import multiprocessing
 import subprocess
+import os
+
+__libc = ctypes.CDLL(None)
+CLONE_NEWUSER = 0x10000000
+CLONE_NEWNET  = 0x40000000
+
+# TODO: Replace me with os.setns once Python 3.12 is old news
+def setns(fd, nstype):
+    __NR_setns = 308
+    __libc.syscall(__NR_setns, fd, nstype)
 
 class IsolatedMacVlan:
     """Create an isolated interface on top of a PC interface."""
@@ -40,6 +52,30 @@ class IsolatedMacVlan:
         self.sleeper.kill()
         self.sleeper.wait()
 
+    def __ns_call(self, fn, tx):
+        pid = self.sleeper.pid
+
+        uns = os.open(f"/proc/{pid}/ns/user", os.O_RDONLY)
+        setns(uns, CLONE_NEWUSER)
+        os.close(uns)
+
+        nns = os.open(f"/proc/{pid}/ns/net", os.O_RDONLY)
+        setns(nns, CLONE_NEWNET)
+        os.close(nns)
+
+        tx.send(fn())
+        tx.close()
+
+    def call(self, fn):
+        rx, tx = multiprocessing.Pipe(duplex=False)
+
+        proc = multiprocessing.Process(target=self.__ns_call, args=(fn, tx))
+        proc.start()
+        ret = rx.recv()
+        rx.close()
+        proc.join()
+        return ret
+
     def _mangle_subprocess_args(self, args, kwargs):
         if args:
             args = list(args)