[PATCH v2 11/22] tasst: Add helpers to run commands with nstool

[David Gibson <david@gibson.dropbear.id.au>]

Use our existing nstool C helper, add python wrappers to easily run
commands in various namespaces.

Signed-off-by: David Gibson <david@gibson.dropbear.id.au>
---
 test/Makefile          |   8 +-
 test/tasst/__main__.py |   2 +-
 test/tasst/nstool.py   | 170 +++++++++++++++++++++++++++++++++++++++++
 test/tasst/snh.py      |  16 ++++
 4 files changed, 192 insertions(+), 4 deletions(-)
 create mode 100644 test/tasst/nstool.py

diff --git a/test/Makefile b/test/Makefile
index 8373ae77..83725f59 100644
--- a/test/Makefile
+++ b/test/Makefile
@@ -64,13 +64,15 @@ LOCAL_ASSETS = mbuto.img mbuto.mem.img podman/bin/podman QEMU_EFI.fd \
 	$(TESTDATA_ASSETS)
 
 ASSETS = $(DOWNLOAD_ASSETS) $(LOCAL_ASSETS)
+AVOCADO_ASSETS =
+META_ASSETS = nstool
 
 EXETER_SH = build/static_checkers.sh
 EXETER_PY = build/build.py
 EXETER_JOBS = $(EXETER_SH:%.sh=%.json) $(EXETER_PY:%.py=%.json)
 AVOCADO_JOBS = $(EXETER_JOBS) avocado/static_checkers.json
 
-TASST_SRCS = __init__.py __main__.py snh.py
+TASST_SRCS = __init__.py __main__.py nstool.py snh.py
 
 EXETER_META = meta/lint.json meta/tasst.json
 META_JOBS = $(EXETER_META)
@@ -157,11 +159,11 @@ meta/tasst.json: $(TASST_SRCS:%=tasst/%) $(VENV) pull-exeter
 	cd ..; PYTHONPATH=$(PYPATH_BASE) $(PYTHON) -m tasst --avocado > test/$@
 
 .PHONY: avocado
-avocado: venv $(AVOCADO_JOBS)
+avocado: venv $(AVOCADO_ASSETS) $(AVOCADO_JOBS)
 	$(RUN_AVOCADO) all $(AVOCADO_JOBS)
 
 .PHONY: meta
-meta: venv $(META_JOBS)
+meta: venv $(META_ASSETS) $(META_JOBS)
 	$(RUN_AVOCADO) meta $(META_JOBS)
 
 flake8:
diff --git a/test/tasst/__main__.py b/test/tasst/__main__.py
index 91499128..9fd6174e 100644
--- a/test/tasst/__main__.py
+++ b/test/tasst/__main__.py
@@ -13,7 +13,7 @@ library of test helpers for passt & pasta
 import exeter
 
 # We import just to get the exeter tests, which flake8 can't see
-from . import snh  # noqa: F401
+from . import nstool, snh  # noqa: F401
 
 
 if __name__ == '__main__':
diff --git a/test/tasst/nstool.py b/test/tasst/nstool.py
new file mode 100644
index 00000000..0b23fbfb
--- /dev/null
+++ b/test/tasst/nstool.py
@@ -0,0 +1,170 @@
+#! /usr/bin/env python3
+
+# SPDX-License-Identifier: GPL-2.0-or-later
+#
+# Copyright Red Hat
+# Author: David Gibson <david@gibson.dropbear.id.au>
+
+"""
+Test A Simple Socket Transport
+
+nstool.py - Run commands in namespaces via 'nstool'
+"""
+
+import contextlib
+import os
+import subprocess
+import tempfile
+
+import exeter
+
+from .snh import RealHost, SimNetHost
+
+# FIXME: Can this be made more portable?
+UNIX_PATH_MAX = 108
+
+NSTOOL_BIN = 'test/nstool'
+
+
+class NsTool(SimNetHost):
+    """A bundle of Linux namespaces managed by nstool"""
+
+    def __init__(self, name, sockpath, parent=RealHost()):
+        if len(sockpath) > UNIX_PATH_MAX:
+            raise ValueError(
+                f'Unix domain socket path "{sockpath}" is too long'
+            )
+
+        super().__init__(name)
+        self.sockpath = sockpath
+        self.parent = parent
+        self._pid = None
+
+    def __enter__(self):
+        cmd = [f'{NSTOOL_BIN}', 'info', '-wp', f'{self.sockpath}']
+        pid = self.parent.output(*cmd, timeout=1)
+        self._pid = int(pid)
+        return self
+
+    def __exit__(self, *exc_details):
+        pass
+
+    # PID of the nstool hold process as seen by the parent snh
+    def pid(self):
+        return self._pid
+
+    # PID of the nstool hold process as seen by another snh which can
+    # see the nstool socket (important when using PID namespaces)
+    def relative_pid(self, relative_to):
+        cmd = [f'{NSTOOL_BIN}', 'info', '-p', f'{self.sockpath}']
+        relpid = relative_to.output(*cmd)
+        return int(relpid)
+
+    def hostify(self, *cmd, capable=False, **kwargs):
+        hostcmd = [f'{NSTOOL_BIN}', 'exec']
+        if capable:
+            hostcmd.append('--keep-caps')
+        hostcmd += [self.sockpath, '--']
+        hostcmd += list(cmd)
+        return hostcmd, kwargs
+
+
+@contextlib.contextmanager
+def unshare_snh(name, *opts, parent=RealHost(), capable=False):
+    # Create path for temporary nstool Unix socket
+    with tempfile.TemporaryDirectory() as tmpd:
+        sockpath = os.path.join(tmpd, name)
+        cmd = ['unshare'] + list(opts)
+        cmd += ['--', f'{NSTOOL_BIN}', 'hold', f'{sockpath}']
+        with parent.bg(*cmd, capable=capable) as holder:
+            try:
+                with NsTool(name, sockpath, parent=parent) as snh:
+                    yield snh
+            finally:
+                try:
+                    parent.fg(f'{NSTOOL_BIN}', 'stop', f'{sockpath}')
+                finally:
+                    try:
+                        holder.run(timeout=0.1)
+                        holder.kill()
+                    finally:
+                        try:
+                            os.remove(sockpath)
+                        except FileNotFoundError:
+                            pass
+
+
+TEST_EXC = ValueError
+
+
+def test_sockdir_cleanup(s):
+    def mess(sockpaths):
+        with s as snh:
+            ns = snh
+            while isinstance(ns, NsTool):
+                sockpaths.append(ns.sockpath)
+                ns = ns.parent
+            raise TEST_EXC
+
+    sockpaths = []
+    exeter.assert_raises(TEST_EXC, mess, sockpaths)
+    assert sockpaths
+    for path in sockpaths:
+        assert not os.path.exists(os.path.dirname(path))
+
+
+def userns_snh():
+    return unshare_snh('usernetns', '-Ucn')
+
+
+@exeter.test
+def test_userns():
+    cmd = ['capsh', '--has-p=CAP_SETUID']
+    with RealHost() as realhost:
+        status = realhost.fg(*cmd, check=False)
+        assert status.returncode != 0
+    with userns_snh() as ns:
+        ns.fg(*cmd, capable=True)
+
+
+@contextlib.contextmanager
+def nested_snh():
+    with unshare_snh('userns', '-Uc') as userns:
+        with unshare_snh('netns', '-n', parent=userns, capable=True) as netns:
+            yield netns
+
+
+def pidns_snh():
+    return unshare_snh('pidns', '-Upfn')
+
+
+@exeter.test
+def test_relative_pid():
+    with pidns_snh() as snh:
+        # The holder is init (pid 1) within its own pidns
+        exeter.assert_eq(snh.relative_pid(snh), 1)
+
+
+# General tests for all the nstool examples
+for setup in [userns_snh, nested_snh, pidns_snh]:
+    # Common snh tests
+    SimNetHost.selftest_isolated(setup)
+    exeter.register_pipe(f'{setup.__qualname__}|test_sockdir_cleanup',
+                         setup, test_sockdir_cleanup)
+
+
+@contextlib.contextmanager
+def connect_snh():
+    with tempfile.TemporaryDirectory() as tmpd:
+        sockpath = os.path.join(tmpd, 'nons')
+        holdcmd = [f'{NSTOOL_BIN}', 'hold', f'{sockpath}']
+        with subprocess.Popen(holdcmd) as holder:
+            try:
+                with NsTool("fakens", sockpath) as snh:
+                    yield snh
+            finally:
+                holder.kill()
+                os.remove(sockpath)
+
+
+SimNetHost.selftest(connect_snh)
diff --git a/test/tasst/snh.py b/test/tasst/snh.py
index 8ee9802a..598ea979 100644
--- a/test/tasst/snh.py
+++ b/test/tasst/snh.py
@@ -178,6 +178,22 @@ class SimNetHost(contextlib.AbstractContextManager):
             testid = f'{setup.__qualname__}|{t.__qualname__}'
             exeter.register_pipe(testid, setup, t)
 
+    # Additional tests only valid if the snh is isolated (no outside
+    # network connections)
+    def test_is_isolated(self):
+        with self as snh:
+            exeter.assert_eq(snh.ifs(), ['lo'])
+
+    ISOLATED_SELFTESTS = [test_is_isolated]
+
+    @classmethod
+    def selftest_isolated(cls, setup):
+        "Register self tests for an isolated snh example"
+        cls.selftest(setup)
+        for t in cls.ISOLATED_SELFTESTS:
+            testid = f'{setup.__qualname__}|{t.__qualname__}'
+            exeter.register_pipe(testid, setup, t)
+
 
 class RealHost(SimNetHost):
     """Represents the host on which the tests are running (as opposed
-- 
@@ -178,6 +178,22 @@ class SimNetHost(contextlib.AbstractContextManager):
             testid = f'{setup.__qualname__}|{t.__qualname__}'
             exeter.register_pipe(testid, setup, t)
 
+    # Additional tests only valid if the snh is isolated (no outside
+    # network connections)
+    def test_is_isolated(self):
+        with self as snh:
+            exeter.assert_eq(snh.ifs(), ['lo'])
+
+    ISOLATED_SELFTESTS = [test_is_isolated]
+
+    @classmethod
+    def selftest_isolated(cls, setup):
+        "Register self tests for an isolated snh example"
+        cls.selftest(setup)
+        for t in cls.ISOLATED_SELFTESTS:
+            testid = f'{setup.__qualname__}|{t.__qualname__}'
+            exeter.register_pipe(testid, setup, t)
+
 
 class RealHost(SimNetHost):
     """Represents the host on which the tests are running (as opposed
-- 
2.45.2