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.
This commit is contained in:
Tobias Waldekranz
2024-02-25 19:49:27 +01:00
committed by Joachim Wiberg
parent 9ce763662f
commit fcc9fe84f4
+36
View File
@@ -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)