Compare commits

...
Author SHA1 Message Date
Tobias Waldekranz bc064c53f3 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.
2024-02-09 21:16:15 +01:00
+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)