[mirror_frr.git] / tests / topotests / munet / base.py

# -*- coding: utf-8 eval: (blacken-mode 1) -*-
# SPDX-License-Identifier: GPL-2.0-or-later
#
# July 9 2021, Christian Hopps <chopps@labn.net>
#
# Copyright 2021, LabN Consulting, L.L.C.
#
"""A module that implements core functionality for library or standalone use."""
import asyncio
import datetime
import errno
import ipaddress
import logging
import os
import platform
import re
import readline
import shlex
import signal
import subprocess
import sys
import tempfile
import time as time_mod

from collections import defaultdict
from pathlib import Path
from typing import Union

from . import config as munet_config
from . import linux


try:
    import pexpect

    from pexpect.fdpexpect import fdspawn
    from pexpect.popen_spawn import PopenSpawn

    have_pexpect = True
except ImportError:
    have_pexpect = False

PEXPECT_PROMPT = "PEXPECT_PROMPT>"
PEXPECT_CONTINUATION_PROMPT = "PEXPECT_PROMPT+"

root_hostname = subprocess.check_output("hostname")
our_pid = os.getpid()


class MunetError(Exception):
    """A generic munet error."""


class CalledProcessError(subprocess.CalledProcessError):
    """Improved logging subclass of subprocess.CalledProcessError."""

    def __str__(self):
        o = self.output.strip() if self.output else ""
        e = self.stderr.strip() if self.stderr else ""
        s = f"returncode: {self.returncode} command: {self.cmd}"
        o = "\n\tstdout: " + o if o else ""
        e = "\n\tstderr: " + e if e else ""
        return s + o + e

    def __repr__(self):
        o = self.output.strip() if self.output else ""
        e = self.stderr.strip() if self.stderr else ""
        return f"munet.base.CalledProcessError({self.returncode}, {self.cmd}, {o}, {e})"


class Timeout:
    """An object to passively monitor for timeouts."""

    def __init__(self, delta):
        self.delta = datetime.timedelta(seconds=delta)
        self.started_on = datetime.datetime.now()
        self.expires_on = self.started_on + self.delta

    def elapsed(self):
        elapsed = datetime.datetime.now() - self.started_on
        return elapsed.total_seconds()

    def is_expired(self):
        return datetime.datetime.now() > self.expires_on

    def remaining(self):
        remaining = self.expires_on - datetime.datetime.now()
        return remaining.total_seconds()

    def __iter__(self):
        return self

    def __next__(self):
        remaining = self.remaining()
        if remaining <= 0:
            raise StopIteration()
        return remaining


def fsafe_name(name):
    return "".join(x if x.isalnum() else "_" for x in name)


def indent(s):
    return "\t" + s.replace("\n", "\n\t")


def shell_quote(command):
    """Return command wrapped in single quotes."""
    if sys.version_info[0] >= 3:
        return shlex.quote(command)
    return "'" + command.replace("'", "'\"'\"'") + "'"


def cmd_error(rc, o, e):
    s = f"rc {rc}"
    o = "\n\tstdout: " + o.strip() if o and o.strip() else ""
    e = "\n\tstderr: " + e.strip() if e and e.strip() else ""
    return s + o + e


def proc_str(p):
    if hasattr(p, "args"):
        args = p.args if isinstance(p.args, str) else " ".join(p.args)
    else:
        args = ""
    return f"proc pid: {p.pid} args: {args}"


def proc_error(p, o, e):
    if hasattr(p, "args"):
        args = p.args if isinstance(p.args, str) else " ".join(p.args)
    else:
        args = ""

    s = f"rc {p.returncode} pid {p.pid}"
    a = "\n\targs: " + args if args else ""
    o = "\n\tstdout: " + (o.strip() if o and o.strip() else "*empty*")
    e = "\n\tstderr: " + (e.strip() if e and e.strip() else "*empty*")
    return s + a + o + e


def comm_error(p):
    rc = p.poll()
    assert rc is not None
    if not hasattr(p, "saved_output"):
        p.saved_output = p.communicate()
    return proc_error(p, *p.saved_output)


async def acomm_error(p):
    rc = p.returncode
    assert rc is not None
    if not hasattr(p, "saved_output"):
        p.saved_output = await p.communicate()
    return proc_error(p, *p.saved_output)


def get_kernel_version():
    kvs = (
        subprocess.check_output("uname -r", shell=True, text=True).strip().split("-", 1)
    )
    kv = kvs[0].split(".")
    kv = [int(x) for x in kv]
    return kv


def convert_number(value) -> int:
    """Convert a number value with a possible suffix to an integer.

    >>> convert_number("100k") == 100 * 1024
    True
    >>> convert_number("100M") == 100 * 1000 * 1000
    True
    >>> convert_number("100Gi") == 100 * 1024 * 1024 * 1024
    True
    >>> convert_number("55") == 55
    True
    """
    if value is None:
        raise ValueError("Invalid value None for convert_number")
    rate = str(value)
    base = 1000
    if rate[-1] == "i":
        base = 1024
        rate = rate[:-1]
    suffix = "KMGTPEZY"
    index = suffix.find(rate[-1])
    if index == -1:
        base = 1024
        index = suffix.lower().find(rate[-1])
    if index != -1:
        rate = rate[:-1]
    return int(rate) * base ** (index + 1)


def is_file_like(fo):
    return isinstance(fo, int) or hasattr(fo, "fileno")


def get_tc_bits_value(user_value):
    value = convert_number(user_value) / 1000
    return f"{value:03f}kbit"


def get_tc_bytes_value(user_value):
    # Raw numbers are bytes in tc
    return convert_number(user_value)


def get_tmp_dir(uniq):
    return os.path.join(tempfile.mkdtemp(), uniq)


async def _async_get_exec_path(binary, cmdf, cache):
    if isinstance(binary, str):
        bins = [binary]
    else:
        bins = binary
    for b in bins:
        if b in cache:
            return cache[b]

        rc, output, _ = await cmdf("which " + b, warn=False)
        if not rc:
            cache[b] = os.path.abspath(output.strip())
            return cache[b]
    return None


def _get_exec_path(binary, cmdf, cache):
    if isinstance(binary, str):
        bins = [binary]
    else:
        bins = binary
    for b in bins:
        if b in cache:
            return cache[b]

        rc, output, _ = cmdf("which " + b, warn=False)
        if not rc:
            cache[b] = os.path.abspath(output.strip())
            return cache[b]
    return None


def get_event_loop():
    """Configure and return our non-thread using event loop.

    This function configures a new child watcher to not use threads.
    Threads cannot be used when we inline unshare a PID namespace.
    """
    policy = asyncio.get_event_loop_policy()
    loop = policy.get_event_loop()
    owatcher = policy.get_child_watcher()
    logging.debug(
        "event_loop_fixture: global policy %s, current loop %s, current watcher %s",
        policy,
        loop,
        owatcher,
    )

    policy.set_child_watcher(None)
    owatcher.close()

    try:
        watcher = asyncio.PidfdChildWatcher()  # pylint: disable=no-member
    except Exception:
        watcher = asyncio.SafeChildWatcher()
    loop = policy.get_event_loop()

    logging.debug(
        "event_loop_fixture: attaching new watcher %s to loop and setting in policy",
        watcher,
    )
    watcher.attach_loop(loop)
    policy.set_child_watcher(watcher)
    policy.set_event_loop(loop)
    assert asyncio.get_event_loop_policy().get_child_watcher() is watcher

    return loop


class Commander:  # pylint: disable=R0904
    """An object that can execute commands."""

    tmux_wait_gen = 0

    def __init__(self, name, logger=None, unet=None, **kwargs):
        """Create a Commander.

        Args:
            name: name of the commander object
            logger: logger to use for logging commands a defualt is supplied if this
                is None
            unet: unet that owns this object, only used by Commander in run_in_window,
                otherwise can be None.
        """
        # del kwargs  # deal with lint warning
        # logging.warning("Commander: name %s kwargs %s", name, kwargs)

        self.name = name
        self.unet = unet
        self.deleting = False
        self.last = None
        self.exec_paths = {}

        if not logger:
            logname = f"munet.{self.__class__.__name__.lower()}.{name}"
            self.logger = logging.getLogger(logname)
            self.logger.setLevel(logging.DEBUG)
        else:
            self.logger = logger

        super().__init__(**kwargs)

    @property
    def is_vm(self):
        return False

    @property
    def is_container(self):
        return False

    def set_logger(self, logfile):
        self.logger = logging.getLogger(__name__ + ".commander." + self.name)
        self.logger.setLevel(logging.DEBUG)
        if isinstance(logfile, str):
            handler = logging.FileHandler(logfile, mode="w")
        else:
            handler = logging.StreamHandler(logfile)

        fmtstr = "%(asctime)s.%(msecs)03d %(levelname)s: {}({}): %(message)s".format(
            self.__class__.__name__, self.name
        )
        handler.setFormatter(logging.Formatter(fmt=fmtstr))
        self.logger.addHandler(handler)

    def _get_pre_cmd(self, use_str, use_pty, **kwargs):
        """Get the pre-user-command values.

        The values returned here should be what is required to cause the user's command
        to execute in the correct context (e.g., namespace, container, sshremote).
        """
        del kwargs
        del use_pty
        return "" if use_str else []

    def __str__(self):
        return f"{self.__class__.__name__}({self.name})"

    async def async_get_exec_path(self, binary):
        """Return the full path to the binary executable.

        `binary` :: binary name or list of binary names
        """
        return await _async_get_exec_path(
            binary, self.async_cmd_status_nsonly, self.exec_paths
        )

    def get_exec_path(self, binary):
        """Return the full path to the binary executable.

        `binary` :: binary name or list of binary names
        """
        return _get_exec_path(binary, self.cmd_status_nsonly, self.exec_paths)

    def get_exec_path_host(self, binary):
        """Return the full path to the binary executable.

        If the object is actually a derived class (e.g., a container) this method will
        return the exec path for the native namespace rather than the container. The
        path is the one which the other xxx_host methods will use.

        `binary` :: binary name or list of binary names
        """
        return get_exec_path_host(binary)

    def test(self, flags, arg):
        """Run test binary, with flags and arg."""
        test_path = self.get_exec_path(["test"])
        rc, _, _ = self.cmd_status([test_path, flags, arg], warn=False)
        return not rc

    def test_nsonly(self, flags, arg):
        """Run test binary, with flags and arg."""
        test_path = self.get_exec_path(["test"])
        rc, _, _ = self.cmd_status_nsonly([test_path, flags, arg], warn=False)
        return not rc

    def path_exists(self, path):
        """Check if path exists."""
        return self.test("-e", path)

    async def cleanup_pid(self, pid, kill_pid=None):
        """Signal a pid to exit with escalating forcefulness."""
        if kill_pid is None:
            kill_pid = pid

        for sn in (signal.SIGHUP, signal.SIGKILL):
            self.logger.debug(
                "%s: %s %s (wait %s)", self, signal.Signals(sn).name, kill_pid, pid
            )

            os.kill(kill_pid, sn)

            # No need to wait after this.
            if sn == signal.SIGKILL:
                return

            # try each signal, waiting 15 seconds for exit before advancing
            wait_sec = 30
            self.logger.debug("%s: waiting %ss for pid to exit", self, wait_sec)
            for _ in Timeout(wait_sec):
                try:
                    status = os.waitpid(pid, os.WNOHANG)
                    if status == (0, 0):
                        await asyncio.sleep(0.1)
                    else:
                        self.logger.debug("pid %s exited status %s", pid, status)
                        return
                except OSError as error:
                    if error.errno == errno.ECHILD:
                        self.logger.debug("%s: pid %s was reaped", self, pid)
                    else:
                        self.logger.warning(
                            "%s: error waiting on pid %s: %s", self, pid, error
                        )
                    return
            self.logger.debug("%s: timeout waiting on pid %s to exit", self, pid)

    def _get_sub_args(self, cmd_list, defaults, use_pty=False, ns_only=False, **kwargs):
        """Returns pre-command, cmd, and default keyword args."""
        assert not isinstance(cmd_list, str)

        defaults["shell"] = False
        pre_cmd_list = self._get_pre_cmd(False, use_pty, ns_only=ns_only, **kwargs)
        cmd_list = [str(x) for x in cmd_list]

        # os_env = {k: v for k, v in os.environ.items() if k.startswith("MUNET")}
        # env = {**os_env, **(kwargs["env"] if "env" in kwargs else {})}
        env = {**(kwargs["env"] if "env" in kwargs else os.environ)}
        if "MUNET_NODENAME" not in env:
            env["MUNET_NODENAME"] = self.name
        kwargs["env"] = env

        defaults.update(kwargs)

        return pre_cmd_list, cmd_list, defaults

    def _common_prologue(self, async_exec, method, cmd, skip_pre_cmd=False, **kwargs):
        cmd_list = self._get_cmd_as_list(cmd)
        if method == "_spawn":
            defaults = {
                "encoding": "utf-8",
                "codec_errors": "ignore",
            }
        else:
            defaults = {
                "stdout": subprocess.PIPE,
                "stderr": subprocess.PIPE,
            }
            if not async_exec:
                defaults["encoding"] = "utf-8"

        pre_cmd_list, cmd_list, defaults = self._get_sub_args(
            cmd_list, defaults, **kwargs
        )

        use_pty = kwargs.get("use_pty", False)
        if method == "_spawn":
            # spawn doesn't take "shell" keyword arg
            if "shell" in defaults:
                del defaults["shell"]
            # this is required to avoid receiving a STOPPED signal on expect!
            if not use_pty:
                defaults["preexec_fn"] = os.setsid
            defaults["env"]["PS1"] = "$ "

        self.logger.debug(
            '%s: %s %s("%s", pre_cmd: "%s" use_pty: %s kwargs: %.120s)',
            self,
            "XXX" if method == "_spawn" else "",
            method,
            cmd_list,
            pre_cmd_list if not skip_pre_cmd else "",
            use_pty,
            defaults,
        )

        actual_cmd_list = cmd_list if skip_pre_cmd else pre_cmd_list + cmd_list
        return actual_cmd_list, defaults

    async def _async_popen(self, method, cmd, **kwargs):
        """Create a new asynchronous subprocess."""
        acmd, kwargs = self._common_prologue(True, method, cmd, **kwargs)
        p = await asyncio.create_subprocess_exec(*acmd, **kwargs)
        return p, acmd

    def _popen(self, method, cmd, **kwargs):
        """Create a subprocess."""
        acmd, kwargs = self._common_prologue(False, method, cmd, **kwargs)
        p = subprocess.Popen(acmd, **kwargs)
        return p, acmd

    def _fdspawn(self, fo, **kwargs):
        defaults = {}
        defaults.update(kwargs)

        if "echo" in defaults:
            del defaults["echo"]

        if "encoding" not in defaults:
            defaults["encoding"] = "utf-8"
            if "codec_errors" not in defaults:
                defaults["codec_errors"] = "ignore"
        encoding = defaults["encoding"]

        self.logger.debug("%s: _fdspawn(%s, kwargs: %s)", self, fo, defaults)

        p = fdspawn(fo, **defaults)

        # We don't have TTY like conversions of LF to CRLF
        p.crlf = os.linesep.encode(encoding)

        # we own the socket now detach the file descriptor to keep it from closing
        if hasattr(fo, "detach"):
            fo.detach()

        return p

    def _spawn(self, cmd, skip_pre_cmd=False, use_pty=False, echo=False, **kwargs):
        logging.debug(
            '%s: XXX _spawn: cmd "%s" skip_pre_cmd %s use_pty %s echo %s kwargs %s',
            self,
            cmd,
            skip_pre_cmd,
            use_pty,
            echo,
            kwargs,
        )
        actual_cmd, defaults = self._common_prologue(
            False, "_spawn", cmd, skip_pre_cmd=skip_pre_cmd, use_pty=use_pty, **kwargs
        )

        self.logger.debug(
            '%s: XXX %s("%s", use_pty %s echo %s defaults: %s)',
            self,
            "PopenSpawn" if not use_pty else "pexpect.spawn",
            actual_cmd,
            use_pty,
            echo,
            defaults,
        )

        # We don't specify a timeout it defaults to 30s is that OK?
        if not use_pty:
            p = PopenSpawn(actual_cmd, **defaults)
        else:
            p = pexpect.spawn(actual_cmd[0], actual_cmd[1:], echo=echo, **defaults)
        return p, actual_cmd

    def spawn(
        self,
        cmd,
        spawned_re,
        expects=(),
        sends=(),
        use_pty=False,
        logfile=None,
        logfile_read=None,
        logfile_send=None,
        trace=None,
        **kwargs,
    ):
        """Create a spawned send/expect process.

        Args:
            cmd: list of args to exec/popen with, or an already open socket
            spawned_re: what to look for to know when done, `spawn` returns when seen
            expects: a list of regex other than `spawned_re` to look for. Commonly,
                "ogin:" or "[Pp]assword:"r.
            sends: what to send when an element of `expects` matches. So e.g., the
                username or password if thats what corresponding expect matched. Can
                be the empty string to send nothing.
            use_pty: true for pty based expect, otherwise uses popen (pipes/files)
            trace: if true then log send/expects
            **kwargs - kwargs passed on the _spawn.

        Returns:
            A pexpect process.

        Raises:
            pexpect.TIMEOUT, pexpect.EOF as documented in `pexpect`
            CalledProcessError if EOF is seen and `cmd` exited then
                raises a CalledProcessError to indicate the failure.
        """
        if is_file_like(cmd):
            assert not use_pty
            ac = "*socket*"
            p = self._fdspawn(cmd, **kwargs)
        else:
            p, ac = self._spawn(cmd, use_pty=use_pty, **kwargs)

        if logfile:
            p.logfile = logfile
        if logfile_read:
            p.logfile_read = logfile_read
        if logfile_send:
            p.logfile_send = logfile_send

        # for spawned shells (i.e., a direct command an not a console)
        # this is wrong and will cause 2 prompts
        if not use_pty:
            # This isn't very nice looking
            p.echo = False
            if not is_file_like(cmd):
                p.isalive = lambda: p.proc.poll() is None
            if not hasattr(p, "close"):
                p.close = p.wait

        # Do a quick check to see if we got the prompt right away, otherwise we may be
        # at a console so we send a \n to re-issue the prompt
        index = p.expect([spawned_re, pexpect.TIMEOUT, pexpect.EOF], timeout=0.1)
        if index == 0:
            assert p.match is not None
            self.logger.debug(
                "%s: got spawned_re quick: '%s' matching '%s'",
                self,
                p.match.group(0),
                spawned_re,
            )
            return p

        # Now send a CRLF to cause the prompt (or whatever else) to re-issue
        p.send("\n")
        try:
            patterns = [spawned_re, *expects]

            self.logger.debug("%s: expecting: %s", self, patterns)

            while index := p.expect(patterns):
                if trace:
                    assert p.match is not None
                    self.logger.debug(
                        "%s: got expect: '%s' matching %d '%s', sending '%s'",
                        self,
                        p.match.group(0),
                        index,
                        patterns[index],
                        sends[index - 1],
                    )
                if sends[index - 1]:
                    p.send(sends[index - 1])

                self.logger.debug("%s: expecting again: %s", self, patterns)
            self.logger.debug(
                "%s: got spawned_re: '%s' matching '%s'",
                self,
                p.match.group(0),
                spawned_re,
            )
            return p
        except pexpect.TIMEOUT:
            self.logger.error(
                "%s: TIMEOUT looking for spawned_re '%s' expect buffer so far:\n%s",
                self,
                spawned_re,
                indent(p.buffer),
            )
            raise
        except pexpect.EOF as eoferr:
            if p.isalive():
                raise
            rc = p.status
            before = indent(p.before)
            error = CalledProcessError(rc, ac, output=before)
            self.logger.error(
                "%s: EOF looking for spawned_re '%s' before EOF:\n%s",
                self,
                spawned_re,
                before,
            )
            p.close()
            raise error from eoferr

    async def shell_spawn(
        self,
        cmd,
        prompt,
        expects=(),
        sends=(),
        use_pty=False,
        will_echo=False,
        is_bourne=True,
        init_newline=False,
        **kwargs,
    ):
        """Create a shell REPL (read-eval-print-loop).

        Args:
            cmd: shell and list of args to popen with, or an already open socket
            prompt: the REPL prompt to look for, the function returns when seen
            expects: a list of regex other than `spawned_re` to look for. Commonly,
                "ogin:" or "[Pp]assword:"r.
            sends: what to send when an element of `expects` matches. So e.g., the
                username or password if thats what corresponding expect matched. Can
                be the empty string to send nothing.
            is_bourne: if False then do not modify shell prompt for internal
                parser friently format, and do not expect continuation prompts.
            init_newline: send an initial newline for non-bourne shell spawns, otherwise
                expect the prompt simply from running the command
            use_pty: true for pty based expect, otherwise uses popen (pipes/files)
            will_echo: bash is buggy in that it echo's to non-tty unlike any other
                sh/ksh, set this value to true if running back
            **kwargs - kwargs passed on the _spawn.
        """
        combined_prompt = r"({}|{})".format(re.escape(PEXPECT_PROMPT), prompt)

        assert not is_file_like(cmd) or not use_pty
        p = self.spawn(
            cmd,
            combined_prompt,
            expects=expects,
            sends=sends,
            use_pty=use_pty,
            echo=False,
            **kwargs,
        )
        assert not p.echo

        if not is_bourne:
            if init_newline:
                p.send("\n")
            return ShellWrapper(p, prompt, will_echo=will_echo)

        ps1 = PEXPECT_PROMPT
        ps2 = PEXPECT_CONTINUATION_PROMPT

        # Avoid problems when =/usr/bin/env= prints the values
        ps1p = ps1[:5] + "${UNSET_V}" + ps1[5:]
        ps2p = ps2[:5] + "${UNSET_V}" + ps2[5:]

        ps1 = re.escape(ps1)
        ps2 = re.escape(ps2)

        extra = "PAGER=cat; export PAGER; TERM=dumb; unset HISTFILE; set +o emacs +o vi"
        pchg = "PS1='{0}' PS2='{1}' PROMPT_COMMAND=''\n".format(ps1p, ps2p)
        p.send(pchg)
        return ShellWrapper(p, ps1, ps2, extra_init_cmd=extra, will_echo=will_echo)

    def popen(self, cmd, **kwargs):
        """Creates a pipe with the given `command`.

        Args:
            cmd: `str` or `list` of command to open a pipe with.
            **kwargs: kwargs is eventually passed on to Popen. If `command` is a string
                then will be invoked with `bash -c`, otherwise `command` is a list and
                will be invoked without a shell.

        Returns:
            a subprocess.Popen object.
        """
        return self._popen("popen", cmd, **kwargs)[0]

    def popen_nsonly(self, cmd, **kwargs):
        """Creates a pipe with the given `command`.

        Args:
            cmd: `str` or `list` of command to open a pipe with.
            **kwargs: kwargs is eventually passed on to Popen. If `command` is a string
                then will be invoked with `bash -c`, otherwise `command` is a list and
                will be invoked without a shell.

        Returns:
            a subprocess.Popen object.
        """
        return self._popen("popen_nsonly", cmd, ns_only=True, **kwargs)[0]

    async def async_popen(self, cmd, **kwargs):
        """Creates a pipe with the given `command`.

        Args:
            cmd: `str` or `list` of command to open a pipe with.
            **kwargs: kwargs is eventually passed on to create_subprocess_exec. If
                `command` is a string then will be invoked with `bash -c`, otherwise
                `command` is a list and will be invoked without a shell.

        Returns:
            a asyncio.subprocess.Process object.
        """
        p, _ = await self._async_popen("async_popen", cmd, **kwargs)
        return p

    async def async_popen_nsonly(self, cmd, **kwargs):
        """Creates a pipe with the given `command`.

        Args:
            cmd: `str` or `list` of command to open a pipe with.
            **kwargs: kwargs is eventually passed on to create_subprocess_exec. If
                `command` is a string then will be invoked with `bash -c`, otherwise
                `command` is a list and will be invoked without a shell.

        Returns:
            a asyncio.subprocess.Process object.
        """
        p, _ = await self._async_popen(
            "async_popen_nsonly", cmd, ns_only=True, **kwargs
        )
        return p

    async def async_cleanup_proc(self, p, pid=None):
        """Terminate a process started with a popen call.

        Args:
            p: return value from :py:`async_popen`, :py:`popen`, et al.
            pid: pid to signal instead of p.pid, typically a child of
                cmd_p == nsenter.

        Returns:
            None on success, the ``p`` if multiple timeouts occur even
            after a SIGKILL sent.
        """
        if not p:
            return None

        if p.returncode is not None:
            if isinstance(p, subprocess.Popen):
                o, e = p.communicate()
            else:
                o, e = await p.communicate()
            self.logger.debug(
                "%s: cmd_p already exited status: %s", self, proc_error(p, o, e)
            )
            return None

        if pid is None:
            pid = p.pid

        self.logger.debug("%s: terminate process: %s (pid %s)", self, proc_str(p), pid)
        try:
            # This will SIGHUP and wait a while then SIGKILL and return immediately
            await self.cleanup_pid(p.pid, pid)

            # Wait another 2 seconds after the possible SIGKILL above for the
            # parent nsenter to cleanup and exit
            wait_secs = 2
            if isinstance(p, subprocess.Popen):
                o, e = p.communicate(timeout=wait_secs)
            else:
                o, e = await asyncio.wait_for(p.communicate(), timeout=wait_secs)
            self.logger.debug(
                "%s: cmd_p exited after kill, status: %s", self, proc_error(p, o, e)
            )
        except (asyncio.TimeoutError, subprocess.TimeoutExpired):
            self.logger.warning("%s: SIGKILL timeout", self)
            return p
        except Exception as error:
            self.logger.warning(
                "%s: kill unexpected exception: %s", self, error, exc_info=True
            )
            return p
        return None

    @staticmethod
    def _cmd_status_input(stdin):
        pinput = None
        if isinstance(stdin, (bytes, str)):
            pinput = stdin
            stdin = subprocess.PIPE
        return pinput, stdin

    def _cmd_status_finish(self, p, c, ac, o, e, raises, warn):
        rc = p.returncode
        self.last = (rc, ac, c, o, e)
        if rc:
            if warn:
                self.logger.warning("%s: proc failed: %s", self, proc_error(p, o, e))
            if raises:
                # error = Exception("stderr: {}".format(stderr))
                # This annoyingly doesnt' show stderr when printed normally
                raise CalledProcessError(rc, ac, o, e)
        return rc, o, e

    def _cmd_status(self, cmds, raises=False, warn=True, stdin=None, **kwargs):
        """Execute a command."""
        pinput, stdin = Commander._cmd_status_input(stdin)
        p, actual_cmd = self._popen("cmd_status", cmds, stdin=stdin, **kwargs)
        o, e = p.communicate(pinput)
        return self._cmd_status_finish(p, cmds, actual_cmd, o, e, raises, warn)

    async def _async_cmd_status(
        self, cmds, raises=False, warn=True, stdin=None, text=None, **kwargs
    ):
        """Execute a command."""
        pinput, stdin = Commander._cmd_status_input(stdin)
        p, actual_cmd = await self._async_popen(
            "async_cmd_status", cmds, stdin=stdin, **kwargs
        )

        if text is False:
            encoding = None
        else:
            encoding = kwargs.get("encoding", "utf-8")

        if encoding is not None and isinstance(pinput, str):
            pinput = pinput.encode(encoding)
        o, e = await p.communicate(pinput)
        if encoding is not None:
            o = o.decode(encoding) if o is not None else o
            e = e.decode(encoding) if e is not None else e
        return self._cmd_status_finish(p, cmds, actual_cmd, o, e, raises, warn)

    def _get_cmd_as_list(self, cmd):
        """Given a list or string return a list form for execution.

        If `cmd` is a string then the returned list uses bash and looks
        like this: ["/bin/bash", "-c", cmd]. Some node types override
        this function if they utilize a different shell as to return
        a different list of values.

        Args:
            cmd: list or string representing the command to execute.

        Returns:
            list of commands to execute.
        """
        if not isinstance(cmd, str):
            cmds = cmd
        else:
            # Make sure the code doesn't think `cd` will work.
            assert not re.match(r"cd(\s*|\s+(\S+))$", cmd)
            cmds = ["/bin/bash", "-c", cmd]
        return cmds

    def cmd_nostatus(self, cmd, **kwargs):
        """Run given command returning output[s].

        Args:
            cmd: `str` or `list` of the command to execute.  If a string is given
                it is run using a shell, otherwise the list is executed directly
                as the binary and arguments.
            **kwargs: kwargs is eventually passed on to Popen. If `command` is a string
                then will be invoked with `bash -c`, otherwise `command` is a list and
                will be invoked without a shell.

        Returns:
            if "stderr" is in kwargs and not equal to subprocess.STDOUT, then
            both stdout and stderr are returned, otherwise stderr is combined
            with stdout and only stdout is returned.
        """
        #
        # This method serves as the basis for all derived sync cmd variations, so to
        # override sync cmd behavior simply override this function and *not* the other
        # variations, unless you are changing only that variation's behavior
        #

        # XXX change this back to _cmd_status instead of cmd_status when we
        # consolidate and cleanup the container overrides of *cmd_* functions

        cmds = cmd
        if "stderr" in kwargs and kwargs["stderr"] != subprocess.STDOUT:
            _, o, e = self.cmd_status(cmds, **kwargs)
            return o, e
        if "stderr" in kwargs:
            del kwargs["stderr"]
        _, o, _ = self.cmd_status(cmds, stderr=subprocess.STDOUT, **kwargs)
        return o

    def cmd_status(self, cmd, **kwargs):
        """Run given command returning status and outputs.

        Args:
            cmd: `str` or `list` of the command to execute.  If a string is given
                it is run using a shell, otherwise the list is executed directly
                as the binary and arguments.
            **kwargs: kwargs is eventually passed on to Popen. If `command` is a string
                then will be invoked with `bash -c`, otherwise `command` is a list and
                will be invoked without a shell.

        Returns:
            (status, output, error) are returned
            status: the returncode of the command.
            output: stdout as a string from the command.
            error: stderr as a string from the command.
        """
        #
        # This method serves as the basis for all derived sync cmd variations, so to
        # override sync cmd behavior simply override this function and *not* the other
        # variations, unless you are changing only that variation's behavior
        #
        return self._cmd_status(cmd, **kwargs)

    def cmd_raises(self, cmd, **kwargs):
        """Execute a command. Raise an exception on errors.

        Args:
            cmd: `str` or `list` of the command to execute.  If a string is given
                it is run using a shell, otherwise the list is executed directly
                as the binary and arguments.
            **kwargs: kwargs is eventually passed on to Popen. If `command` is a string
                then will be invoked with `bash -c`, otherwise `command` is a list and
                will be invoked without a shell.

        Returns:
            output: stdout as a string from the command.

        Raises:
            CalledProcessError: on non-zero exit status
        """
        _, stdout, _ = self._cmd_status(cmd, raises=True, **kwargs)
        return stdout

    def cmd_nostatus_nsonly(self, cmd, **kwargs):
        # Make sure the command runs on the host and not in any container.
        return self.cmd_nostatus(cmd, ns_only=True, **kwargs)

    def cmd_status_nsonly(self, cmd, **kwargs):
        # Make sure the command runs on the host and not in any container.
        return self._cmd_status(cmd, ns_only=True, **kwargs)

    def cmd_raises_nsonly(self, cmd, **kwargs):
        # Make sure the command runs on the host and not in any container.
        _, stdout, _ = self._cmd_status(cmd, raises=True, ns_only=True, **kwargs)
        return stdout

    async def async_cmd_status(self, cmd, **kwargs):
        """Run given command returning status and outputs.

        Args:
            cmd: `str` or `list` of the command to execute.  If a string is given
                it is run using a shell, otherwise the list is executed directly
                as the binary and arguments.
            **kwargs: kwargs is eventually passed on to create_subprocess_exec. If
                `cmd` is a string then will be invoked with `bash -c`, otherwise
                `cmd` is a list and will be invoked without a shell.

        Returns:
            (status, output, error) are returned
            status: the returncode of the command.
            output: stdout as a string from the command.
            error: stderr as a string from the command.
        """
        #
        # This method serves as the basis for all derived async cmd variations, so to
        # override async cmd behavior simply override this function and *not* the other
        # variations, unless you are changing only that variation's behavior
        #
        return await self._async_cmd_status(cmd, **kwargs)

    async def async_cmd_nostatus(self, cmd, **kwargs):
        """Run given command returning output[s].

        Args:
            cmd: `str` or `list` of the command to execute.  If a string is given
                it is run using a shell, otherwise the list is executed directly
                as the binary and arguments.
            **kwargs: kwargs is eventually passed on to create_subprocess_exec. If
                `cmd` is a string then will be invoked with `bash -c`, otherwise
                `cmd` is a list and will be invoked without a shell.

        Returns:
            if "stderr" is in kwargs and not equal to subprocess.STDOUT, then
            both stdout and stderr are returned, otherwise stderr is combined
            with stdout and only stdout is returned.

        """
        # XXX change this back to _async_cmd_status instead of cmd_status when we
        # consolidate and cleanup the container overrides of *cmd_* functions

        cmds = cmd
        if "stderr" in kwargs and kwargs["stderr"] != subprocess.STDOUT:
            _, o, e = await self._async_cmd_status(cmds, **kwargs)
            return o, e
        if "stderr" in kwargs:
            del kwargs["stderr"]
        _, o, _ = await self._async_cmd_status(cmds, stderr=subprocess.STDOUT, **kwargs)
        return o

    async def async_cmd_raises(self, cmd, **kwargs):
        """Execute a command. Raise an exception on errors.

        Args:
            cmd: `str` or `list` of the command to execute.  If a string is given
                it is run using a shell, otherwise the list is executed directly
                as the binary and arguments.
            **kwargs: kwargs is eventually passed on to create_subprocess_exec. If
                `cmd` is a string then will be invoked with `bash -c`, otherwise
                `cmd` is a list and will be invoked without a shell.

        Returns:
            output: stdout as a string from the command.

        Raises:
            CalledProcessError: on non-zero exit status
        """
        _, stdout, _ = await self._async_cmd_status(cmd, raises=True, **kwargs)
        return stdout

    async def async_cmd_status_nsonly(self, cmd, **kwargs):
        # Make sure the command runs on the host and not in any container.
        return await self._async_cmd_status(cmd, ns_only=True, **kwargs)

    async def async_cmd_raises_nsonly(self, cmd, **kwargs):
        # Make sure the command runs on the host and not in any container.
        _, stdout, _ = await self._async_cmd_status(
            cmd, raises=True, ns_only=True, **kwargs
        )
        return stdout

    def cmd_legacy(self, cmd, **kwargs):
        """Execute a command with stdout and stderr joined, *IGNORES ERROR*."""
        defaults = {"stderr": subprocess.STDOUT}
        defaults.update(kwargs)
        _, stdout, _ = self._cmd_status(cmd, raises=False, **defaults)
        return stdout

    # Run a command in a new window (gnome-terminal, screen, tmux, xterm)
    def run_in_window(
        self,
        cmd,
        wait_for=False,
        background=False,
        name=None,
        title=None,
        forcex=False,
        new_window=False,
        tmux_target=None,
        ns_only=False,
    ):
        """Run a command in a new window (TMUX, Screen or XTerm).

        Args:
            cmd: string to execute.
            wait_for: True to wait for exit from command or `str` as channel neme to
                signal on exit, otherwise False
            background: Do not change focus to new window.
            title: Title for new pane (tmux) or window (xterm).
            name: Name of the new window (tmux)
            forcex: Force use of X11.
            new_window: Open new window (instead of pane) in TMUX
            tmux_target: Target for tmux pane.

        Returns:
            the pane/window identifier from TMUX (depends on `new_window`)
        """
        channel = None
        if isinstance(wait_for, str):
            channel = wait_for
        elif wait_for is True:
            channel = "{}-wait-{}".format(our_pid, Commander.tmux_wait_gen)
            Commander.tmux_wait_gen += 1

        if forcex or ("TMUX" not in os.environ and "STY" not in os.environ):
            root_level = False
        else:
            root_level = True

        # SUDO: The important thing to note is that with all these methods we are
        # executing on the users windowing system, so even though we are normally
        # running as root, we will not be when the command is dispatched. Also
        # in the case of SCREEN and X11 we need to sudo *back* to the user as well
        # This is also done by SSHRemote by defualt so we should *not* sudo back
        # if we are SSHRemote.

        # XXX need to test ssh in screen
        # XXX need to test ssh in Xterm
        sudo_path = get_exec_path_host(["sudo"])
        # This first test case seems same as last but using list instead of string?
        if self.is_vm and self.use_ssh:  # pylint: disable=E1101
            if isinstance(cmd, str):
                cmd = shlex.split(cmd)
            cmd = ["/usr/bin/env", f"MUNET_NODENAME={self.name}"] + cmd

            # get the ssh cmd
            cmd = self._get_pre_cmd(False, True, ns_only=ns_only) + [shlex.join(cmd)]
            unet = self.unet  # pylint: disable=E1101
            uns_cmd = unet._get_pre_cmd(  # pylint: disable=W0212
                False, True, ns_only=True, root_level=root_level
            )
            # get the nsenter for munet
            nscmd = [
                sudo_path,
                *uns_cmd,
                *cmd,
            ]
        else:
            # This is the command to execute to be inside the namespace.
            # We are getting into trouble with quoting.
            # Why aren't we passing in MUNET_RUNDIR?
            cmd = f"/usr/bin/env MUNET_NODENAME={self.name} {cmd}"
            # We need sudo b/c we are executing as the user inside the window system.
            sudo_path = get_exec_path_host(["sudo"])
            nscmd = (
                sudo_path
                + " "
                + self._get_pre_cmd(True, True, ns_only=ns_only, root_level=root_level)
                + " "
                + cmd
            )

        if "TMUX" in os.environ and not forcex:
            cmd = [get_exec_path_host("tmux")]
            if new_window:
                cmd.append("new-window")
                cmd.append("-P")
                if name:
                    cmd.append("-n")
                    cmd.append(name)
                if tmux_target:
                    cmd.append("-t")
                    cmd.append(tmux_target)
            else:
                cmd.append("split-window")
                cmd.append("-P")
                cmd.append("-h")
                if not tmux_target:
                    tmux_target = os.getenv("TMUX_PANE", "")
            if background:
                cmd.append("-d")
            if tmux_target:
                cmd.append("-t")
                cmd.append(tmux_target)

            # nscmd is always added as single string argument
            if not isinstance(nscmd, str):
                nscmd = shlex.join(nscmd)
            if title:
                nscmd = f"printf '\033]2;{title}\033\\'; {nscmd}"
            if channel:
                nscmd = f'trap "tmux wait -S {channel}; exit 0" EXIT; {nscmd}'
            cmd.append(nscmd)

        elif "STY" in os.environ and not forcex:
            # wait for not supported in screen for now
            channel = None
            cmd = [get_exec_path_host("screen")]
            if not os.path.exists(
                "/run/screen/S-{}/{}".format(os.environ["USER"], os.environ["STY"])
            ):
                # XXX not appropriate for ssh
                cmd = ["sudo", "-Eu", os.environ["SUDO_USER"]] + cmd

            if title:
                cmd.append("-t")
                cmd.append(title)

            if isinstance(nscmd, str):
                nscmd = shlex.split(nscmd)
            cmd.extend(nscmd)
        elif "DISPLAY" in os.environ:
            cmd = [get_exec_path_host("xterm")]
            if "SUDO_USER" in os.environ:
                # Do this b/c making things work as root with xauth seems hard
                cmd = [
                    get_exec_path_host("sudo"),
                    "-Eu",
                    os.environ["SUDO_USER"],
                ] + cmd
            if title:
                cmd.append("-T")
                cmd.append(title)

            cmd.append("-e")
            if isinstance(nscmd, str):
                cmd.extend(shlex.split(nscmd))
            else:
                cmd.extend(nscmd)

            # if channel:
            #    return self.cmd_raises(cmd, skip_pre_cmd=True)
            # else:
            p = commander.popen(
                cmd,
                # skip_pre_cmd=True,
                stdin=None,
                shell=False,
            )
            # We should reap the child and report the error then.
            time_mod.sleep(2)
            if p.poll() is not None:
                self.logger.error("%s: Failed to launch xterm: %s", self, comm_error(p))
            return p
        else:
            self.logger.error(
                "DISPLAY, STY, and TMUX not in environment, can't open window"
            )
            raise Exception("Window requestd but TMUX, Screen and X11 not available")

        # pane_info = self.cmd_raises(cmd, skip_pre_cmd=True, ns_only=True).strip()
        # We are prepending the nsenter command, so use unet.rootcmd
        pane_info = commander.cmd_raises(cmd).strip()

        # Re-adjust the layout
        if "TMUX" in os.environ:
            cmd = [
                get_exec_path_host("tmux"),
                "select-layout",
                "-t",
                pane_info if not tmux_target else tmux_target,
                "tiled",
            ]
            commander.cmd_status(cmd)

        # Wait here if we weren't handed the channel to wait for
        if channel and wait_for is True:
            cmd = [get_exec_path_host("tmux"), "wait", channel]
            # commander.cmd_status(cmd, skip_pre_cmd=True)
            commander.cmd_status(cmd)

        return pane_info

    def delete(self):
        """Calls self.async_delete within an exec loop."""
        asyncio.run(self.async_delete())

    async def _async_delete(self):
        """Delete this objects resources.

        This is the actual implementation of the resource cleanup, each class
        should cleanup it's own resources, generally catching and reporting,
        but not reraising any exceptions for it's own cleanup, then it should
        invoke `super()._async_delete() without catching any exceptions raised
        therein. See other examples in `base.py` or `native.py`
        """
        self.logger.info("%s: deleted", self)

    async def async_delete(self):
        """Delete the Commander (or derived object).

        The actual implementation for any class should be in `_async_delete`
        new derived classes should look at the documentation for that function.
        """
        try:
            self.deleting = True
            await self._async_delete()
        except Exception as error:
            self.logger.error("%s: error while deleting: %s", self, error)


class InterfaceMixin:
    """A mixin class to support interface functionality."""

    def __init__(self, *args, **kwargs):
        # del kwargs  # get rid of lint
        # logging.warning("InterfaceMixin: args: %s kwargs: %s", args, kwargs)

        self._intf_addrs = defaultdict(lambda: [None, None])
        self.net_intfs = {}
        self.next_intf_index = 0
        self.basename = "eth"
        # self.basename = name + "-eth"
        super().__init__(*args, **kwargs)

    @property
    def intfs(self):
        return sorted(self._intf_addrs.keys())

    @property
    def networks(self):
        return sorted(self.net_intfs.keys())

    def get_intf_addr(self, ifname, ipv6=False):
        if ifname not in self._intf_addrs:
            return None
        return self._intf_addrs[ifname][bool(ipv6)]

    def set_intf_addr(self, ifname, ifaddr):
        ifaddr = ipaddress.ip_interface(ifaddr)
        self._intf_addrs[ifname][ifaddr.version == 6] = ifaddr

    def net_addr(self, netname, ipv6=False):
        if netname not in self.net_intfs:
            return None
        return self.get_intf_addr(self.net_intfs[netname], ipv6=ipv6)

    def set_intf_basename(self, basename):
        self.basename = basename

    def get_next_intf_name(self):
        while True:
            ifname = self.basename + str(self.next_intf_index)
            self.next_intf_index += 1
            if ifname not in self._intf_addrs:
                break
        return ifname

    def get_ns_ifname(self, ifname):
        """Return a namespace unique interface name.

        This function is primarily overriden by L3QemuVM, IOW by any class
        that doesn't create it's own network namespace and will share that
        with the root (unet) namespace.

        Args:
            ifname: the interface name.

        Returns:
            A name unique to the namespace of this object. By defualt the assumption
            is the ifname is namespace unique.
        """
        return ifname

    def register_interface(self, ifname):
        if ifname not in self._intf_addrs:
            self._intf_addrs[ifname] = [None, None]

    def register_network(self, netname, ifname):
        if netname in self.net_intfs:
            assert self.net_intfs[netname] == ifname
        else:
            self.net_intfs[netname] = ifname

    def get_linux_tc_args(self, ifname, config):
        """Get interface constraints (jitter, delay, rate) for linux TC.

        The keys and their values are as follows:

        delay (int): number of microseconds
        jitter (int): number of microseconds
        jitter-correlation (float): % correlation to previous (default 10%)
        loss (float): % of loss
        loss-correlation (float): % correlation to previous (default 0%)
        rate  (int or str): bits per second, string allows for use of
            {KMGTKiMiGiTi} prefixes "i" means K == 1024 otherwise K == 1000
        """
        del ifname  # unused

        netem_args = ""

        def get_number(c, v, d=None):
            if v not in c or c[v] is None:
                return d
            return convert_number(c[v])

        delay = get_number(config, "delay")
        if delay is not None:
            netem_args += f" delay {delay}usec"

        jitter = get_number(config, "jitter")
        if jitter is not None:
            if not delay:
                raise ValueError("jitter but no delay specified")
            jitter_correlation = get_number(config, "jitter-correlation", 10)
            netem_args += f" {jitter}usec {jitter_correlation}%"

        loss = get_number(config, "loss")
        if loss is not None:
            loss_correlation = get_number(config, "loss-correlation", 0)
            if loss_correlation:
                netem_args += f" loss {loss}% {loss_correlation}%"
            else:
                netem_args += f" loss {loss}%"

        if (o_rate := config.get("rate")) is None:
            return netem_args, ""

        #
        # This comment is not correct, but is trying to talk through/learn the
        # machinery.
        #
        # tokens arrive at `rate` into token buffer.
        # limit - number of bytes that can be queued waiting for tokens
        #   -or-
        # latency - maximum amount of time a packet may sit in TBF queue
        #
        # So this just allows receiving faster than rate for latency amount of
        # time, before dropping.
        #
        # latency = sizeofbucket(limit) / rate (peakrate?)
        #
        #   32kbit
        # -------- = latency = 320ms
        #  100kbps
        #
        #  -but then-
        # burst ([token] buffer) the largest number of instantaneous
        # tokens available (i.e, bucket size).

        tbf_args = ""
        DEFLIMIT = 1518 * 1
        DEFBURST = 1518 * 2
        try:
            tc_rate = o_rate["rate"]
            tc_rate = convert_number(tc_rate)
            limit = convert_number(o_rate.get("limit", DEFLIMIT))
            burst = convert_number(o_rate.get("burst", DEFBURST))
        except (KeyError, TypeError):
            tc_rate = convert_number(o_rate)
            limit = convert_number(DEFLIMIT)
            burst = convert_number(DEFBURST)
        tbf_args += f" rate {tc_rate/1000}kbit"
        if delay:
            # give an extra 1/10 of buffer space to handle delay
            tbf_args += f" limit {limit} burst {burst}"
        else:
            tbf_args += f" limit {limit} burst {burst}"

        return netem_args, tbf_args

    def set_intf_constraints(self, ifname, **constraints):
        """Set interface outbound constraints.

        Set outbound constraints (jitter, delay, rate) for an interface. All arguments
        may also be passed as a string and will be converted to numerical format. All
        arguments are also optional. If not specified then that existing constraint will
        be cleared.

        Args:
            ifname: the name of the interface
            delay (int): number of microseconds.
            jitter (int): number of microseconds.
            jitter-correlation (float): Percent correlation to previous (default 10%).
            loss (float): Percent of loss.
            loss-correlation (float): Percent correlation to previous (default 25%).
            rate (int): bits per second, string allows for use of
                {KMGTKiMiGiTi} prefixes "i" means K == 1024 otherwise K == 1000.
        """
        nsifname = self.get_ns_ifname(ifname)
        netem_args, tbf_args = self.get_linux_tc_args(nsifname, constraints)
        count = 1
        selector = f"root handle {count}:"
        if netem_args:
            self.cmd_raises(
                f"tc qdisc add dev {nsifname} {selector} netem {netem_args}"
            )
            count += 1
            selector = f"parent {count-1}: handle {count}"
        # Place rate limit after delay otherwise limit/burst too complex
        if tbf_args:
            self.cmd_raises(f"tc qdisc add dev {nsifname} {selector} tbf {tbf_args}")

        self.cmd_raises(f"tc qdisc show dev {nsifname}")


class LinuxNamespace(Commander, InterfaceMixin):
    """A linux Namespace.

    An object that creates and executes commands in a linux namespace
    """

    def __init__(
        self,
        name,
        net=True,
        mount=True,
        uts=True,
        cgroup=False,
        ipc=False,
        pid=False,
        time=False,
        user=False,
        unshare_inline=False,
        set_hostname=True,
        private_mounts=None,
        **kwargs,
    ):
        """Create a new linux namespace.

        Args:
            name: Internal name for the namespace.
            net: Create network namespace.
            mount: Create network namespace.
            uts: Create UTS (hostname) namespace.
            cgroup: Create cgroup namespace.
            ipc: Create IPC namespace.
            pid: Create PID namespace, also mounts new /proc.
            time: Create time namespace.
            user: Create user namespace, also keeps capabilities.
            set_hostname: Set the hostname to `name`, uts must also be True.
            private_mounts: List of strings of the form
                "[/external/path:]/internal/path. If no external path is specified a
                tmpfs is mounted on the internal path. Any paths specified are first
                passed to `mkdir -p`.
            unshare_inline: Unshare the process itself rather than using a proxy.
            logger: Passed to superclass.
        """
        # logging.warning("LinuxNamespace: name %s kwargs %s", name, kwargs)

        super().__init__(name, **kwargs)

        unet = self.unet

        self.logger.debug("%s: creating", self)

        self.cwd = os.path.abspath(os.getcwd())

        self.nsflags = []
        self.ifnetns = {}
        self.uflags = 0
        self.p_ns_fds = None
        self.p_ns_fnames = None
        self.pid_ns = False
        self.init_pid = None
        self.unshare_inline = unshare_inline
        self.nsenter_fork = True

        #
        # Collect the namespaces to unshare
        #
        if hasattr(self, "proc_path") and self.proc_path:  # pylint: disable=no-member
            pp = Path(self.proc_path)  # pylint: disable=no-member
        else:
            pp = unet.proc_path if unet else Path("/proc")
        pp = pp.joinpath("%P%", "ns")

        flags = ""
        uflags = 0
        nslist = []
        nsflags = []
        if cgroup:
            nselm = "cgroup"
            nslist.append(nselm)
            nsflags.append(f"--{nselm}={pp / nselm}")
            flags += "C"
            uflags |= linux.CLONE_NEWCGROUP
        if ipc:
            nselm = "ipc"
            nslist.append(nselm)
            nsflags.append(f"--{nselm}={pp / nselm}")
            flags += "i"
            uflags |= linux.CLONE_NEWIPC
        if mount or pid:
            # We need a new mount namespace for pid
            nselm = "mnt"
            nslist.append(nselm)
            nsflags.append(f"--mount={pp / nselm}")
            mount = True
            flags += "m"
            uflags |= linux.CLONE_NEWNS
        if net:
            nselm = "net"
            nslist.append(nselm)
            nsflags.append(f"--{nselm}={pp / nselm}")
            # if pid:
            #     os.system(f"touch /tmp/netns-{name}")
            #     cmd.append(f"--net=/tmp/netns-{name}")
            # else:
            flags += "n"
            uflags |= linux.CLONE_NEWNET
        if pid:
            self.pid_ns = True
            # We look for this b/c the unshare pid will share with /sibn/init
            nselm = "pid_for_children"
            nslist.append(nselm)
            nsflags.append(f"--pid={pp / nselm}")
            flags += "p"
            uflags |= linux.CLONE_NEWPID
        if time:
            nselm = "time"
            # XXX time_for_children?
            nslist.append(nselm)
            nsflags.append(f"--{nselm}={pp / nselm}")
            flags += "T"
            uflags |= linux.CLONE_NEWTIME
        if user:
            nselm = "user"
            nslist.append(nselm)
            nsflags.append(f"--{nselm}={pp / nselm}")
            flags += "U"
            uflags |= linux.CLONE_NEWUSER
        if uts:
            nselm = "uts"
            nslist.append(nselm)
            nsflags.append(f"--{nselm}={pp / nselm}")
            flags += "u"
            uflags |= linux.CLONE_NEWUTS

        assert flags, "LinuxNamespace with no namespaces requested"

        # Should look path up using resources maybe...
        mutini_path = get_our_script_path("mutini")
        if not mutini_path:
            mutini_path = get_our_script_path("mutini.py")
        assert mutini_path
        cmd = [mutini_path, f"--unshare-flags={flags}", "-v"]
        fname = fsafe_name(self.name) + "-mutini.log"
        fname = (unet or self).rundir.joinpath(fname)
        stdout = open(fname, "w", encoding="utf-8")
        stderr = subprocess.STDOUT

        #
        # Save the current namespace info to compare against later
        #

        if not unet:
            nsdict = {x: os.readlink(f"/proc/self/ns/{x}") for x in nslist}
        else:
            nsdict = {
                x: os.readlink(f"{unet.proc_path}/{unet.pid}/ns/{x}") for x in nslist
            }

        #
        # (A) Basically we need to save the pid of the unshare call for nsenter.
        #
        # For `unet is not None` (node case) the level this exists at is based on wether
        # unet is using a forking nsenter or not. So if unet.nsenter_fork == True then
        # we need the child pid of the p.pid (child of pid returned to us), otherwise
        # unet.nsenter_fork == False and we just use p.pid as it will be unshare after
        # nsenter exec's it.
        #
        # For the `unet is None` (unet case) the unshare is at the top level or
        # non-existent so we always save the returned p.pid. If we are unshare_inline we
        # won't have a __pre_cmd but we can save our child_pid to kill later, otherwise
        # we set unet.pid to None b/c there's literally nothing to do.
        #
        # ---------------------------------------------------------------------------
        # Breakdown for nested (non-unet) namespace creation, and what PID
        # to use for __pre_cmd nsenter use.
        # ---------------------------------------------------------------------------
        #
        # tl;dr
        #   - for non-inline unshare: Use BBB with pid_for_children, unless none/none
        #     #then (AAA) returned
        #   - for inline unshare: use returned pid (AAA) with pid_for_children
        #
        # All commands use unet.popen to launch the unshare of mutini or cat.
        # mutini for PID unshare, otherwise cat. AAA is the returned pid BBB is the
        # child of the returned.
        #
        # Unshare Variant
        # ---------------
        #
        # Here we are running mutini if we are creating new pid namespace workspace,
        # cat otherwise.
        #
        # [PID+PID] pid tree looks like this:
        #
        # PID  NSPID PPID PGID
        # uuu    -   N/A  uuu  main unet process
        # AAA    -   uuu  AAA  nsenter (forking, from unet) (in unet namespaces -pid)
        # BBB    -   AAA  AAA  unshare --fork --kill-child (forking)
        # CCC    1   BBB  CCC  mutini (non-forking since it is pid 1 in new namespace)
        #
        # Use BBB if we use pid_for_children, CCC for all
        #
        # [PID+none] For non-pid workspace creation (but unet pid) we use cat and pid
        # tree looks like this:
        #
        # PID  PPID PGID
        # uuu  N/A  uuu  main unet process
        # AAA  uuu  AAA  nsenter (forking) (in unet namespaces -pid)
        # BBB  AAA  AAA  unshare -> cat (from unshare non-forking)
        #
        # Use BBB for all
        #
        # [none+PID] For pid workspace creation (but NOT unet pid) we use mutini and pid
        # tree looks like this:
        #
        # PID  NSPID PPID PGID
        # uuu    -   N/A  uuu  main unet process
        # AAA    -   uuu  AAA  nsenter -> unshare --fork --kill-child
        # BBB    1   AAA  AAA  mutini (non-forking since it is pid 1 in new namespace)
        #
        # Use AAA if we use pid_for_children, BBB for all
        #
        # [none+none] For non-pid workspace and non-pid unet we use cat and pid tree
        # looks like this:
        #
        # PID  PPID PGID
        # uuu  N/A  uuu  main unet process
        # AAA  uuu  AAA  nsenter -> unshare -> cat
        #
        # Use AAA for all, there's no BBB
        #
        # Inline-Unshare Variant
        # ----------------------
        #
        # For unshare_inline and new PID namespace we have unshared all but our PID
        # namespace, but our children end up in the new namespace so the fork popen
        # does is good enough.
        #
        # [PID+PID] pid tree looks like this:
        #
        # PID  NSPID PPID PGID
        # uuu    -   N/A  uuu  main unet process
        # AAA    -   uuu  AAA  unshare --fork --kill-child (forking)
        # BBB    1   AAA  BBB  mutini
        #
        # Use AAA if we use pid_for_children, BBB for all
        #
        # [PID+none] For non-pid workspace creation (but unet pid) we use cat and pid
        # tree looks like this:
        #
        # PID  PPID PGID
        # uuu  N/A  uuu  main unet process
        # AAA  uuu  AAA  unshare -> cat
        #
        # Use AAA for all
        #
        # [none+PID] For pid workspace creation (but NOT unet pid) we use mutini and pid
        # tree looks like this:
        #
        # PID  NSPID PPID PGID
        # uuu    -   N/A  uuu  main unet process
        # AAA    -   uuu  AAA  unshare --fork --kill-child
        # BBB    1   AAA  BBB  mutini
        #
        # Use AAA if we use pid_for_children, BBB for all
        #
        # [none+none] For non-pid workspace and non-pid unet we use cat and pid tree
        # looks like this:
        #
        # PID  PPID PGID
        # uuu  N/A  uuu  main unet process
        # AAA  uuu  AAA  unshare -> cat
        #
        # Use AAA for all.
        #
        #
        # ---------------------------------------------------------------------------
        # Breakdown for unet namespace creation, and what PID to use for __pre_cmd
        # ---------------------------------------------------------------------------
        #
        # tl;dr: save returned PID or nothing.
        #   - for non-inline unshare: Use AAA with pid_for_children (returned pid)
        #   - for inline unshare: no __precmd as the fork in popen is enough.
        #
        # Use commander to launch the unshare mutini/cat (for PID/none
        # workspace PID) for non-inline case. AAA is the returned pid BBB is the child
        # of the returned.
        #
        # Unshare Variant
        # ---------------
        #
        # Here we are running mutini if we are creating new pid namespace workspace,
        # cat otherwise.
        #
        # [PID] for unet pid creation pid tree looks like this:
        #
        # PID  NSPID PPID PGID
        # uuu    -   N/A  uuu  main unet process
        # AAA    -   uuu  AAA  unshare --fork --kill-child (forking)
        # BBB    1   AAA  BBB  mutini
        #
        # Use AAA if we use pid_for_children, BBB for all
        #
        # [none] for unet non-pid, pid tree looks like this:
        #
        # PID  PPID PGID
        # uuu  N/A  uuu  main unet process
        # AAA  uuu  AAA  unshare -> cat
        #
        # Use AAA for all
        #
        # Inline-Unshare Variant
        # -----------------------
        #
        # For unshare_inline and new PID namespace we have unshared all but our PID
        # namespace, but our children end up in the new namespace so the fork in popen
        # does is good enough.
        #
        # [PID] for unet pid creation pid tree looks like this:
        #
        # PID  NSPID PPID PGID
        # uuu    -   N/A  uuu  main unet process
        # AAA    1   uuu  AAA  mutini
        #
        # Save p / p.pid, but don't configure any nsenter, uneeded.
        #
        # Use nothing as the fork when doing a popen is enough to be in all the right
        # namepsaces.
        #
        # [none] for unet non-pid, pid tree looks like this:
        #
        # PID  PPID PGID
        # uuu  N/A  uuu  main unet process
        #
        # Nothing, no __pre_cmd.
        #
        #

        self.ppid = os.getppid()
        self.unshare_inline = unshare_inline
        if unshare_inline:
            assert unet is None
            self.uflags = uflags
            #
            # Open file descriptors for current namespaces for later resotration.
            #
            try:
                kversion = [int(x) for x in platform.release().split("-")[0].split(".")]
                kvok = kversion[0] > 5 or (kversion[0] == 5 and kversion[1] >= 8)
            except ValueError:
                kvok = False
            if (
                not kvok
                or sys.version_info[0] < 3
                or (sys.version_info[0] == 3 and sys.version_info[1] < 9)
            ):
                # get list of namespace file descriptors before we unshare
                self.p_ns_fds = []
                self.p_ns_fnames = []
                tmpflags = uflags
                for i in range(0, 64):
                    v = 1 << i
                    if (tmpflags & v) == 0:
                        continue
                    tmpflags &= ~v
                    if v in linux.namespace_files:
                        path = os.path.join("/proc/self", linux.namespace_files[v])
                        if os.path.exists(path):
                            self.p_ns_fds.append(os.open(path, 0))
                            self.p_ns_fnames.append(f"{path} -> {os.readlink(path)}")
                            self.logger.debug(
                                "%s: saving old namespace fd %s (%s)",
                                self,
                                self.p_ns_fnames[-1],
                                self.p_ns_fds[-1],
                            )
                    if not tmpflags:
                        break
            else:
                self.p_ns_fds = None
                self.p_ns_fnames = None
                self.ppid_fd = linux.pidfd_open(self.ppid)

            self.logger.debug(
                "%s: unshare to new namespaces %s",
                self,
                linux.clone_flag_string(uflags),
            )

            linux.unshare(uflags)

            if not pid:
                p = None
                self.pid = None
                self.nsenter_fork = False
            else:
                # Need to fork to create the PID namespace, but we need to continue
                # running from the parent so that things like pytest work. We'll execute
                # a mutini process to manage the child init 1 duties.
                #
                # We (the parent pid) can no longer create threads, due to that being
                # restricted by the kernel. See EINVAL in clone(2).
                #
                p = commander.popen(
                    [mutini_path, "-v"],
                    stdin=subprocess.PIPE,
                    stdout=stdout,
                    stderr=stderr,
                    text=True,
                    # new session/pgid so signals don't propagate
                    start_new_session=True,
                    shell=False,
                )
                self.pid = p.pid
                self.nsenter_fork = False
        else:
            # Using cat and a stdin PIPE is nice as it will exit when we do. However,
            # we also detach it from the pgid so that signals do not propagate to it.
            # This is b/c it would exit early (e.g., ^C) then, at least the main munet
            # proc which has no other processes like frr daemons running, will take the
            # main network namespace with it, which will remove the bridges and the
            # veth pair (because the bridge side veth is deleted).
            self.logger.debug("%s: creating namespace process: %s", self, cmd)

            # Use the parent unet process if we have one this will cause us to inherit
            # the namespaces correctly even in the non-inline case.
            parent = self.unet if self.unet else commander

            p = parent.popen(
                cmd,
                stdin=subprocess.PIPE,
                stdout=stdout,
                stderr=stderr,
                text=True,
                start_new_session=not unet,
                shell=False,
            )

            # The pid number returned is in the global pid namespace. For unshare_inline
            # this can be unfortunate b/c our /proc has been remounted in our new pid
            # namespace and won't contain global pid namespace pids. To solve for this
            # we get all the pid values for the process below.

            # See (A) above for when we need the child pid.
            self.logger.debug("%s: namespace process: %s", self, proc_str(p))
            self.pid = p.pid
            if unet and unet.nsenter_fork:
                assert not unet.unshare_inline
                # Need child pid of p.pid
                pgrep = unet.rootcmd.get_exec_path("pgrep")
                # a sing fork was done
                child_pid = unet.rootcmd.cmd_raises([pgrep, "-o", "-P", str(p.pid)])
                self.pid = int(child_pid.strip())
                self.logger.debug("%s: child of namespace process: %s", self, pid)

        self.p = p

        # Let's always have a valid value.
        if self.pid is None:
            self.pid = our_pid

        #
        # Let's find all our pids in the nested PID namespaces
        #
        if unet:
            proc_path = unet.proc_path
        else:
            proc_path = self.proc_path if hasattr(self, "proc_path") else "/proc"
        proc_path = f"{proc_path}/{self.pid}"

        pid_status = open(f"{proc_path}/status", "r", encoding="ascii").read()
        m = re.search(r"\nNSpid:((?:\t[0-9]+)+)\n", pid_status)
        self.pids = [int(x) for x in m.group(1).strip().split("\t")]
        assert self.pids[0] == self.pid

        self.logger.debug("%s: namespace scoped pids: %s", self, self.pids)

        # -----------------------------------------------
        # Now let's wait until unshare completes it's job
        # -----------------------------------------------
        timeout = Timeout(30)
        if self.pid is not None and self.pid != our_pid:
            while (not p or not p.poll()) and not timeout.is_expired():
                # check new namespace values against old (nsdict), unshare
                # can actually take a bit to complete.
                for fname in tuple(nslist):
                    # self.pid will be the global pid b/c we didn't unshare_inline
                    nspath = f"{proc_path}/ns/{fname}"
                    try:
                        nsf = os.readlink(nspath)
                    except OSError as error:
                        self.logger.debug(
                            "unswitched: error (ok) checking %s: %s", nspath, error
                        )
                        continue
                    if nsdict[fname] != nsf:
                        self.logger.debug(
                            "switched: original %s current %s", nsdict[fname], nsf
                        )
                        nslist.remove(fname)
                    elif unshare_inline:
                        logging.warning(
                            "unshare_inline not unshared %s == %s", nsdict[fname], nsf
                        )
                    else:
                        self.logger.debug(
                            "unswitched: current %s elapsed: %s", nsf, timeout.elapsed()
                        )
                if not nslist:
                    self.logger.debug(
                        "all done waiting for unshare after %s", timeout.elapsed()
                    )
                    break

                elapsed = int(timeout.elapsed())
                if elapsed <= 3:
                    time_mod.sleep(0.1)
                else:
                    self.logger.info(
                        "%s: unshare taking more than %ss: %s", self, elapsed, nslist
                    )
                    time_mod.sleep(1)

        if p is not None and p.poll():
            self.logger.error("%s: namespace process failed: %s", self, comm_error(p))
            assert p.poll() is None, "unshare failed"

        #
        # Setup the pre-command to enter the target namespace from the running munet
        # process using self.pid
        #

        if pid:
            nsenter_fork = True
        elif unet and unet.nsenter_fork:
            # if unet created a pid namespace we need to enter it since we aren't
            # entering a child pid namespace we created for the node. Otherwise
            # we have a /proc remounted under unet, but our process is running in
            # the root pid namepsace
            nselm = "pid_for_children"
            nsflags.append(f"--pid={pp / nselm}")
            nsenter_fork = True
        else:
            # We dont need a fork.
            nsflags.append("-F")
            nsenter_fork = False

        # Save nsenter values if running from root namespace
        # we need this for the unshare_inline case when run externally (e.g., from
        # within tmux server).
        root_nsflags = [x.replace("%P%", str(self.pid)) for x in nsflags]
        self.__root_base_pre_cmd = ["/usr/bin/nsenter", *root_nsflags]
        self.__root_pre_cmd = list(self.__root_base_pre_cmd)

        if unshare_inline:
            assert unet is None
            # We have nothing to do here since our process is now in the correct
            # namespaces and children will inherit from us, even the PID namespace will
            # be corrent b/c commands are run by first forking.
            self.nsenter_fork = False
            self.nsflags = []
            self.__base_pre_cmd = []
        else:
            # We will use nsenter
            self.nsenter_fork = nsenter_fork
            self.nsflags = nsflags
            self.__base_pre_cmd = list(self.__root_base_pre_cmd)

        self.__pre_cmd = list(self.__base_pre_cmd)

        # Always mark new mount namespaces as recursive private
        if mount:
            # if self.p is None and not pid:
            self.cmd_raises_nsonly("mount --make-rprivate /")

        # We need to remount the procfs for the new PID namespace, since we aren't using
        # unshare(1) which does that for us.
        if pid and unshare_inline:
            assert mount
            self.cmd_raises_nsonly("mount -t proc proc /proc")

        # We do not want cmd_status in child classes (e.g., container) for
        # the remaining setup calls in this __init__ function.

        if net:
            # Remount /sys to pickup any changes in the network, but keep root
            # /sys/fs/cgroup. This pattern could be made generic and supported for any
            # overlapping mounts
            if mount:
                tmpmnt = f"/tmp/cgm-{self.pid}"
                self.cmd_status_nsonly(
                    f"mkdir {tmpmnt} && mount --rbind /sys/fs/cgroup {tmpmnt}"
                )
                rc = o = e = None
                for i in range(0, 10):
                    rc, o, e = self.cmd_status_nsonly(
                        "mount -t sysfs sysfs /sys", warn=False
                    )
                    if not rc:
                        break
                    self.logger.debug(
                        "got error mounting new sysfs will retry: %s",
                        cmd_error(rc, o, e),
                    )
                    time_mod.sleep(1)
                else:
                    raise Exception(cmd_error(rc, o, e))

                self.cmd_status_nsonly(
                    f"mount --move {tmpmnt} /sys/fs/cgroup && rmdir {tmpmnt}"
                )

            # Original micronet code
            # self.cmd_raises_nsonly("mount -t sysfs sysfs /sys")
            # self.cmd_raises_nsonly(
            #     "mount -o rw,nosuid,nodev,noexec,relatime "
            #     "-t cgroup2 cgroup /sys/fs/cgroup"
            # )

        # Set the hostname to the namespace name
        if uts and set_hostname:
            self.cmd_status_nsonly("hostname " + self.name)
            nroot = subprocess.check_output("hostname")
            if unshare_inline or (unet and unet.unshare_inline):
                assert (
                    root_hostname != nroot
                ), f'hostname unchanged from "{nroot}" wanted "{self.name}"'
            else:
                # Assert that we didn't just change the host hostname
                assert (
                    root_hostname == nroot
                ), f'root hostname "{root_hostname}" changed to "{nroot}"!'

        if private_mounts:
            if isinstance(private_mounts, str):
                private_mounts = [private_mounts]
            for m in private_mounts:
                s = m.split(":", 1)
                if len(s) == 1:
                    self.tmpfs_mount(s[0])
                else:
                    self.bind_mount(s[0], s[1])

        # this will fail if running inside the namespace with PID
        if pid:
            o = self.cmd_nostatus_nsonly("ls -l /proc/1/ns")
        else:
            o = self.cmd_nostatus_nsonly("ls -l /proc/self/ns")

        self.logger.debug("namespaces:\n %s", o)

        # will cache the path, which is important in delete to avoid running a shell
        # which can hang during cleanup
        self.ip_path = get_exec_path_host("ip")
        if net:
            self.cmd_status_nsonly([self.ip_path, "link", "set", "lo", "up"])

        self.logger.info("%s: created", self)

    def _get_pre_cmd(self, use_str, use_pty, ns_only=False, root_level=False, **kwargs):
        """Get the pre-user-command values.

        The values returned here should be what is required to cause the user's command
        to execute in the correct context (e.g., namespace, container, sshremote).
        """
        del kwargs
        del ns_only
        del use_pty
        pre_cmd = self.__root_pre_cmd if root_level else self.__pre_cmd
        return shlex.join(pre_cmd) if use_str else list(pre_cmd)

    def tmpfs_mount(self, inner):
        self.logger.debug("Mounting tmpfs on %s", inner)
        self.cmd_raises("mkdir -p " + inner)
        self.cmd_raises("mount -n -t tmpfs tmpfs " + inner)

    def bind_mount(self, outer, inner):
        self.logger.debug("Bind mounting %s on %s", outer, inner)
        if commander.test("-f", outer):
            self.cmd_raises(f"mkdir -p {os.path.dirname(inner)} && touch {inner}")
        else:
            if not commander.test("-e", outer):
                commander.cmd_raises_nsonly(f"mkdir -p {outer}")
            self.cmd_raises(f"mkdir -p {inner}")
        self.cmd_raises("mount --rbind {} {} ".format(outer, inner))

    def add_netns(self, ns):
        self.logger.debug("Adding network namespace %s", ns)

        if os.path.exists("/run/netns/{}".format(ns)):
            self.logger.warning("%s: Removing existing nsspace %s", self, ns)
            try:
                self.delete_netns(ns)
            except Exception as ex:
                self.logger.warning(
                    "%s: Couldn't remove existing nsspace %s: %s",
                    self,
                    ns,
                    str(ex),
                    exc_info=True,
                )
        self.cmd_raises_nsonly([self.ip_path, "netns", "add", ns])

    def delete_netns(self, ns):
        self.logger.debug("Deleting network namespace %s", ns)
        self.cmd_raises_nsonly([self.ip_path, "netns", "delete", ns])

    def set_intf_netns(self, intf, ns, up=False):
        # In case a user hard-codes 1 thinking it "resets"
        ns = str(ns)
        if ns == "1":
            ns = str(self.pid)

        self.logger.debug("Moving interface %s to namespace %s", intf, ns)

        cmd = [self.ip_path, "link", "set", intf, "netns", ns]
        if up:
            cmd.append("up")
        self.intf_ip_cmd(intf, cmd)
        if ns == str(self.pid):
            # If we are returning then remove from dict
            if intf in self.ifnetns:
                del self.ifnetns[intf]
        else:
            self.ifnetns[intf] = ns

    def reset_intf_netns(self, intf):
        self.logger.debug("Moving interface %s to default namespace", intf)
        self.set_intf_netns(intf, str(self.pid))

    def intf_ip_cmd(self, intf, cmd):
        """Run an ip command, considering an interface's possible namespace."""
        if intf in self.ifnetns:
            if isinstance(cmd, list):
                assert cmd[0].endswith("ip")
                cmd[1:1] = ["-n", self.ifnetns[intf]]
            else:
                assert cmd.startswith("ip ")
                cmd = "ip -n " + self.ifnetns[intf] + cmd[2:]
        self.cmd_raises_nsonly(cmd)

    def intf_tc_cmd(self, intf, cmd):
        """Run a tc command, considering an interface's possible namespace."""
        if intf in self.ifnetns:
            if isinstance(cmd, list):
                assert cmd[0].endswith("tc")
                cmd[1:1] = ["-n", self.ifnetns[intf]]
            else:
                assert cmd.startswith("tc ")
                cmd = "tc -n " + self.ifnetns[intf] + cmd[2:]
        self.cmd_raises_nsonly(cmd)

    def set_ns_cwd(self, cwd: Union[str, Path]):
        """Common code for changing pre_cmd and pre_nscmd."""
        self.logger.debug("%s: new CWD %s", self, cwd)
        self.__root_pre_cmd = self.__root_base_pre_cmd + ["--wd=" + str(cwd)]
        if self.__pre_cmd:
            self.__pre_cmd = self.__base_pre_cmd + ["--wd=" + str(cwd)]
        elif self.unshare_inline:
            os.chdir(cwd)

    async def _async_delete(self):
        if type(self) == LinuxNamespace:  # pylint: disable=C0123
            self.logger.info("%s: deleting", self)
        else:
            self.logger.debug("%s: LinuxNamespace sub-class deleting", self)

        # Signal pid namespace proc to exit
        if (
            (self.p is None or self.p.pid != self.pid)
            and self.pid
            and self.pid != our_pid
        ):
            self.logger.debug(
                "cleanup pid on separate pid %s from proc pid %s",
                self.pid,
                self.p.pid if self.p else None,
            )
            await self.cleanup_pid(self.pid)

        if self.p is not None:
            self.logger.debug("cleanup proc pid %s", self.p.pid)
            await self.async_cleanup_proc(self.p)

        # return to the previous namespace, need to do this in case anothe munet
        # is being created, especially when it plans to inherit the parent's (host)
        # namespace.
        if self.uflags:
            logging.info("restoring from inline unshare: cwd: %s", os.getcwd())
            # This only works in linux>=5.8
            if self.p_ns_fds is None:
                self.logger.debug(
                    "%s: restoring namespaces %s",
                    self,
                    linux.clone_flag_string(self.uflags),
                )
                # fd = linux.pidfd_open(self.ppid)
                fd = self.ppid_fd
                retry = 3
                for i in range(0, retry):
                    try:
                        linux.setns(fd, self.uflags)
                    except OSError as error:
                        self.logger.warning(
                            "%s: could not reset to old namespace fd %s: %s",
                            self,
                            fd,
                            error,
                        )
                        if i == retry - 1:
                            raise
                        time_mod.sleep(1)
                os.close(fd)
            else:
                while self.p_ns_fds:
                    fd = self.p_ns_fds.pop()
                    fname = self.p_ns_fnames.pop()
                    self.logger.debug(
                        "%s: restoring namespace from fd %s (%s)", self, fname, fd
                    )
                    retry = 3
                    for i in range(0, retry):
                        try:
                            linux.setns(fd, 0)
                            break
                        except OSError as error:
                            self.logger.warning(
                                "%s: could not reset to old namespace fd %s (%s): %s",
                                self,
                                fname,
                                fd,
                                error,
                            )
                            if i == retry - 1:
                                raise
                        time_mod.sleep(1)
                    os.close(fd)
                self.p_ns_fds = None
                self.p_ns_fnames = None
            logging.info("restored from unshare: cwd: %s", os.getcwd())

        self.__root_base_pre_cmd = ["/bin/false"]
        self.__base_pre_cmd = ["/bin/false"]
        self.__root_pre_cmd = ["/bin/false"]
        self.__pre_cmd = ["/bin/false"]

        await super()._async_delete()


class SharedNamespace(Commander):
    """Share another namespace.

    An object that executes commands in an existing pid's linux namespace
    """

    def __init__(self, name, pid=None, nsflags=None, **kwargs):
        """Share a linux namespace.

        Args:
            name: Internal name for the namespace.
            pid: PID of the process to share with.
            nsflags: nsenter flags to pass to inherit namespaces from
        """
        super().__init__(name, **kwargs)

        self.logger.debug("%s: Creating", self)

        self.cwd = os.path.abspath(os.getcwd())
        self.pid = pid if pid is not None else our_pid

        nsflags = (x.replace("%P%", str(self.pid)) for x in nsflags) if nsflags else []
        self.__base_pre_cmd = ["/usr/bin/nsenter", *nsflags] if nsflags else []
        self.__pre_cmd = self.__base_pre_cmd
        self.ip_path = self.get_exec_path("ip")

    def _get_pre_cmd(self, use_str, use_pty, ns_only=False, root_level=False, **kwargs):
        """Get the pre-user-command values.

        The values returned here should be what is required to cause the user's command
        to execute in the correct context (e.g., namespace, container, sshremote).
        """
        del kwargs
        del ns_only
        del use_pty
        assert not root_level
        return shlex.join(self.__pre_cmd) if use_str else list(self.__pre_cmd)

    def set_ns_cwd(self, cwd: Union[str, Path]):
        """Common code for changing pre_cmd and pre_nscmd."""
        self.logger.debug("%s: new CWD %s", self, cwd)
        self.__pre_cmd = self.__base_pre_cmd + ["--wd=" + str(cwd)]


class Bridge(SharedNamespace, InterfaceMixin):
    """A linux bridge."""

    next_ord = 1

    @classmethod
    def _get_next_id(cls):
        # Do not use `cls` here b/c that makes the variable class specific
        n = Bridge.next_ord
        Bridge.next_ord = n + 1
        return n

    def __init__(self, name=None, mtu=None, unet=None, **kwargs):
        """Create a linux Bridge."""
        self.id = self._get_next_id()
        if not name:
            name = "br{}".format(self.id)

        unet_pid = our_pid if unet.pid is None else unet.pid

        super().__init__(name, pid=unet_pid, nsflags=unet.nsflags, unet=unet, **kwargs)

        self.set_intf_basename(self.name + "-e")

        self.mtu = mtu

        self.logger.debug("Bridge: Creating")

        assert len(self.name) <= 16  # Make sure fits in IFNAMSIZE
        self.cmd_raises(f"ip link delete {name} || true")
        self.cmd_raises(f"ip link add {name} type bridge")
        if self.mtu:
            self.cmd_raises(f"ip link set {name} mtu {self.mtu}")
        self.cmd_raises(f"ip link set {name} up")

        self.logger.debug("%s: Created, Running", self)

    def get_ifname(self, netname):
        return self.net_intfs[netname] if netname in self.net_intfs else None

    async def _async_delete(self):
        """Stop the bridge (i.e., delete the linux resources)."""
        if type(self) == Bridge:  # pylint: disable=C0123
            self.logger.info("%s: deleting", self)
        else:
            self.logger.debug("%s: Bridge sub-class deleting", self)

        rc, o, e = await self.async_cmd_status(
            [self.ip_path, "link", "show", self.name],
            stdin=subprocess.DEVNULL,
            start_new_session=True,
            warn=False,
        )
        if not rc:
            rc, o, e = await self.async_cmd_status(
                [self.ip_path, "link", "delete", self.name],
                stdin=subprocess.DEVNULL,
                start_new_session=True,
                warn=False,
            )
        if rc:
            self.logger.error(
                "%s: error deleting bridge %s: %s",
                self,
                self.name,
                cmd_error(rc, o, e),
            )
        await super()._async_delete()


class BaseMunet(LinuxNamespace):
    """Munet."""

    def __init__(
        self,
        name="munet",
        isolated=True,
        pid=True,
        rundir=None,
        pytestconfig=None,
        **kwargs,
    ):
        """Create a Munet."""
        # logging.warning("BaseMunet: %s", name)

        self.hosts = {}
        self.switches = {}
        self.links = {}
        self.macs = {}
        self.rmacs = {}
        self.isolated = isolated

        self.cli_server = None
        self.cli_sockpath = None
        self.cli_histfile = None
        self.cli_in_window_cmds = {}
        self.cli_run_cmds = {}

        #
        # We need a directory for various files
        #
        if not rundir:
            rundir = "/tmp/munet"
        self.rundir = Path(rundir)

        #
        # Always having a global /proc is required to keep things from exploding
        # complexity with nested new pid namespaces..
        #
        if pid:
            self.proc_path = Path(tempfile.mkdtemp(suffix="-proc", prefix="mu-"))
            logging.debug("%s: mounting /proc on proc_path %s", name, self.proc_path)
            linux.mount("proc", str(self.proc_path), "proc")
        else:
            self.proc_path = Path("/proc")

        #
        # Now create a root level commander that works regardless of whether we inline
        # unshare or not. Save it in the global variable as well
        #

        if not self.isolated:
            self.rootcmd = commander
        elif not pid:
            nsflags = (
                f"--mount={self.proc_path / '1/ns/mnt'}",
                f"--net={self.proc_path / '1/ns/net'}",
                f"--uts={self.proc_path / '1/ns/uts'}",
                # f"--ipc={self.proc_path / '1/ns/ipc'}",
                # f"--time={self.proc_path / '1/ns/time'}",
                # f"--cgroup={self.proc_path / '1/ns/cgroup'}",
            )
            self.rootcmd = SharedNamespace("root", pid=1, nsflags=nsflags)
        else:
            # XXX user
            nsflags = (
                # XXX Backing up PID namespace just doesn't work.
                # f"--pid={self.proc_path / '1/ns/pid_for_children'}",
                f"--mount={self.proc_path / '1/ns/mnt'}",
                f"--net={self.proc_path / '1/ns/net'}",
                f"--uts={self.proc_path / '1/ns/uts'}",
                # f"--ipc={self.proc_path / '1/ns/ipc'}",
                # f"--time={self.proc_path / '1/ns/time'}",
                # f"--cgroup={self.proc_path / '1/ns/cgroup'}",
            )
            self.rootcmd = SharedNamespace("root", pid=1, nsflags=nsflags)
        global roothost  # pylint: disable=global-statement

        roothost = self.rootcmd

        self.cfgopt = munet_config.ConfigOptionsProxy(pytestconfig)

        super().__init__(
            name, mount=True, net=isolated, uts=isolated, pid=pid, unet=None, **kwargs
        )

        # This allows us to cleanup any leftover running munet's
        if "MUNET_PID" in os.environ:
            if os.environ["MUNET_PID"] != str(our_pid):
                logging.error(
                    "Found env MUNET_PID != our pid %s, instead its %s, changing",
                    our_pid,
                    os.environ["MUNET_PID"],
                )
        os.environ["MUNET_PID"] = str(our_pid)

        # this is for testing purposes do not use
        if not BaseMunet.g_unet:
            BaseMunet.g_unet = self

        self.logger.debug("%s: Creating", self)

    def __getitem__(self, key):
        if key in self.switches:
            return self.switches[key]
        return self.hosts[key]

    def add_host(self, name, cls=LinuxNamespace, **kwargs):
        """Add a host to munet."""
        self.logger.debug("%s: add_host %s(%s)", self, cls.__name__, name)

        self.hosts[name] = cls(name, unet=self, **kwargs)

        # Create a new mounted FS for tracking nested network namespaces creatd by the
        # user with `ip netns add`

        # XXX why is this failing with podman???
        # self.hosts[name].tmpfs_mount("/run/netns")

        return self.hosts[name]

    def add_link(self, node1, node2, if1, if2, mtu=None, **intf_constraints):
        """Add a link between switch and node or 2 nodes.

        If constraints are given they are applied to each endpoint. See
        `InterfaceMixin::set_intf_constraints()` for more info.
        """
        isp2p = False

        try:
            name1 = node1.name
        except AttributeError:
            if node1 in self.switches:
                node1 = self.switches[node1]
            else:
                node1 = self.hosts[node1]
            name1 = node1.name

        try:
            name2 = node2.name
        except AttributeError:
            if node2 in self.switches:
                node2 = self.switches[node2]
            else:
                node2 = self.hosts[node2]
            name2 = node2.name

        if name1 in self.switches:
            assert name2 in self.hosts
        elif name2 in self.switches:
            assert name1 in self.hosts
            name1, name2 = name2, name1
            if1, if2 = if2, if1
        else:
            # p2p link
            assert name1 in self.hosts
            assert name2 in self.hosts
            isp2p = True

        lname = "{}:{}-{}:{}".format(name1, if1, name2, if2)
        self.logger.debug("%s: add_link %s%s", self, lname, " p2p" if isp2p else "")
        self.links[lname] = (name1, if1, name2, if2)

        # And create the veth now.
        if isp2p:
            lhost, rhost = self.hosts[name1], self.hosts[name2]
            lifname = "i1{:x}".format(lhost.pid)

            # Done at root level
            nsif1 = lhost.get_ns_ifname(if1)
            nsif2 = rhost.get_ns_ifname(if2)

            # Use pids[-1] to get the unet scoped pid for hosts
            self.cmd_raises_nsonly(
                f"ip link add {lifname} type veth peer name {nsif2}"
                f" netns {rhost.pids[-1]}"
            )
            self.cmd_raises_nsonly(f"ip link set {lifname} netns {lhost.pids[-1]}")

            lhost.cmd_raises_nsonly("ip link set {} name {}".format(lifname, nsif1))
            if mtu:
                lhost.cmd_raises_nsonly("ip link set {} mtu {}".format(nsif1, mtu))
            lhost.cmd_raises_nsonly("ip link set {} up".format(nsif1))
            lhost.register_interface(if1)

            if mtu:
                rhost.cmd_raises_nsonly("ip link set {} mtu {}".format(nsif2, mtu))
            rhost.cmd_raises_nsonly("ip link set {} up".format(nsif2))
            rhost.register_interface(if2)
        else:
            switch = self.switches[name1]
            rhost = self.hosts[name2]

            nsif1 = switch.get_ns_ifname(if1)
            nsif2 = rhost.get_ns_ifname(if2)

            if mtu is None:
                mtu = switch.mtu

            if len(nsif1) > 16:
                self.logger.error('"%s" len %s > 16', nsif1, len(nsif1))
            elif len(nsif2) > 16:
                self.logger.error('"%s" len %s > 16', nsif2, len(nsif2))
            assert len(nsif1) <= 16 and len(nsif2) <= 16  # Make sure fits in IFNAMSIZE

            self.logger.debug("%s: Creating veth pair for link %s", self, lname)

            # Use pids[-1] to get the unet scoped pid for hosts
            # switch is already in our namespace so nothing to convert.
            self.cmd_raises_nsonly(
                f"ip link add {nsif1} type veth peer name {nsif2}"
                f" netns {rhost.pids[-1]}"
            )

            if mtu:
                # if switch.mtu:
                #     # the switch interface should match the switch config
                #     switch.cmd_raises_nsonly(
                #         "ip link set {} mtu {}".format(if1, switch.mtu)
                #     )
                switch.cmd_raises_nsonly("ip link set {} mtu {}".format(nsif1, mtu))
                rhost.cmd_raises_nsonly("ip link set {} mtu {}".format(nsif2, mtu))

            switch.register_interface(if1)
            rhost.register_interface(if2)
            rhost.register_network(switch.name, if2)

            switch.cmd_raises_nsonly(f"ip link set {nsif1} master {switch.name}")

            switch.cmd_raises_nsonly(f"ip link set {nsif1} up")
            rhost.cmd_raises_nsonly(f"ip link set {nsif2} up")

        # Cache the MAC values, and reverse mapping
        self.get_mac(name1, nsif1)
        self.get_mac(name2, nsif2)

        # Setup interface constraints if provided
        if intf_constraints:
            node1.set_intf_constraints(if1, **intf_constraints)
            node2.set_intf_constraints(if2, **intf_constraints)

    def add_switch(self, name, cls=Bridge, **kwargs):
        """Add a switch to munet."""
        self.logger.debug("%s: add_switch %s(%s)", self, cls.__name__, name)
        self.switches[name] = cls(name, unet=self, **kwargs)
        return self.switches[name]

    def get_mac(self, name, ifname):
        if name in self.hosts:
            dev = self.hosts[name]
        else:
            dev = self.switches[name]

        nsifname = self.get_ns_ifname(ifname)

        if (name, ifname) not in self.macs:
            _, output, _ = dev.cmd_status_nsonly("ip -o link show " + nsifname)
            m = re.match(".*link/(loopback|ether) ([0-9a-fA-F:]+) .*", output)
            mac = m.group(2)
            self.macs[(name, ifname)] = mac
            self.rmacs[mac] = (name, ifname)

        return self.macs[(name, ifname)]

    async def _delete_link(self, lname):
        rname, rif = self.links[lname][2:4]
        host = self.hosts[rname]
        nsrif = host.get_ns_ifname(rif)

        self.logger.debug("%s: Deleting veth pair for link %s", self, lname)
        rc, o, e = await host.async_cmd_status_nsonly(
            [self.ip_path, "link", "delete", nsrif],
            stdin=subprocess.DEVNULL,
            start_new_session=True,
            warn=False,
        )
        if rc:
            self.logger.error("Err del veth pair %s: %s", lname, cmd_error(rc, o, e))

    async def _delete_links(self):
        # for x in self.links:
        #     await self._delete_link(x)
        return await asyncio.gather(*[self._delete_link(x) for x in self.links])

    async def _async_delete(self):
        """Delete the munet topology."""
        # logger = self.logger if False else logging
        logger = self.logger
        if type(self) == BaseMunet:  # pylint: disable=C0123
            logger.info("%s: deleting.", self)
        else:
            logger.debug("%s: BaseMunet sub-class deleting.", self)

        logger.debug("Deleting links")
        try:
            await self._delete_links()
        except Exception as error:
            logger.error("%s: error deleting links: %s", self, error, exc_info=True)

        logger.debug("Deleting hosts and bridges")
        try:
            # Delete hosts and switches, wait for them all to complete
            # even if there is an exception.
            htask = [x.async_delete() for x in self.hosts.values()]
            stask = [x.async_delete() for x in self.switches.values()]
            await asyncio.gather(*htask, *stask, return_exceptions=True)
        except Exception as error:
            logger.error(
                "%s: error deleting hosts and switches: %s", self, error, exc_info=True
            )

        self.links = {}
        self.hosts = {}
        self.switches = {}

        try:
            if self.cli_server:
                self.cli_server.cancel()
                self.cli_server = None
            if self.cli_sockpath:
                await self.async_cmd_status(
                    "rm -rf " + os.path.dirname(self.cli_sockpath)
                )
                self.cli_sockpath = None
        except Exception as error:
            logger.error(
                "%s: error cli server or sockpaths: %s", self, error, exc_info=True
            )

        try:
            if self.cli_histfile:
                readline.write_history_file(self.cli_histfile)
                self.cli_histfile = None
        except Exception as error:
            logger.error(
                "%s: error saving history file: %s", self, error, exc_info=True
            )

        # XXX for some reason setns during the delete is changing our dir to /.
        cwd = os.getcwd()

        try:
            await super()._async_delete()
        except Exception as error:
            logger.error(
                "%s: error deleting parent classes: %s", self, error, exc_info=True
            )
        os.chdir(cwd)

        try:
            if self.proc_path and str(self.proc_path) != "/proc":
                logger.debug("%s: umount, remove proc_path %s", self, self.proc_path)
                linux.umount(str(self.proc_path), 0)
                os.rmdir(self.proc_path)
        except Exception as error:
            logger.warning(
                "%s: error umount and removing proc_path %s: %s",
                self,
                self.proc_path,
                error,
                exc_info=True,
            )
            try:
                linux.umount(str(self.proc_path), linux.MNT_DETACH)
            except Exception as error2:
                logger.error(
                    "%s: error umount with detach proc_path %s: %s",
                    self,
                    self.proc_path,
                    error2,
                    exc_info=True,
                )

        if BaseMunet.g_unet == self:
            BaseMunet.g_unet = None


BaseMunet.g_unet = None

if True:  # pylint: disable=using-constant-test

    class ShellWrapper:
        """A Read-Execute-Print-Loop (REPL) interface.

        A newline or prompt changing command should be sent to the
        spawned child prior to creation as the `prompt` will be `expect`ed
        """

        def __init__(
            self,
            spawn,
            prompt,
            continuation_prompt=None,
            extra_init_cmd=None,
            will_echo=False,
            escape_ansi=False,
        ):
            self.echo = will_echo
            self.escape = (
                re.compile(r"(\x9B|\x1B\[)[0-?]*[ -\/]*[@-~]") if escape_ansi else None
            )

            logging.debug(
                'ShellWraper: XXX prompt "%s" will_echo %s child.echo %s',
                prompt,
                will_echo,
                spawn.echo,
            )

            self.child = spawn
            if self.child.echo:
                logging.info("Setting child to echo")
                self.child.setecho(False)
                self.child.waitnoecho()
                assert not self.child.echo

            self.prompt = prompt
            self.cont_prompt = continuation_prompt

            # Use expect_exact if we can as it should be faster
            self.expects = [prompt]
            if re.escape(prompt) == prompt and hasattr(self.child, "expect_exact"):
                self._expectf = self.child.expect_exact
            else:
                self._expectf = self.child.expect
            if continuation_prompt:
                self.expects.append(continuation_prompt)
                if re.escape(continuation_prompt) != continuation_prompt:
                    self._expectf = self.child.expect

            if extra_init_cmd:
                self.expect_prompt()
                self.child.sendline(extra_init_cmd)
            self.expect_prompt()

        def expect_prompt(self, timeout=-1):
            return self._expectf(self.expects, timeout=timeout)

        def run_command(self, command, timeout=-1):
            """Pexpect REPLWrapper compatible run_command.

            This will split `command` into lines and feed each one to the shell.

            Args:
                command: string of commands separated by newlines, a trailing
                    newline will cause and empty line to be sent.
                timeout: pexpect timeout value.
            """
            lines = command.splitlines()
            if command[-1] == "\n":
                lines.append("")
            output = ""
            index = 0
            for line in lines:
                self.child.sendline(line)
                index = self.expect_prompt(timeout=timeout)
                output += self.child.before

            if index:
                if hasattr(self.child, "kill"):
                    self.child.kill(signal.SIGINT)
                else:
                    self.child.send("\x03")
                self.expect_prompt(timeout=30 if self.child.timeout is None else -1)
                raise ValueError("Continuation prompt found at end of commands")

            if self.escape:
                output = self.escape.sub("", output)

            return output

        def cmd_nostatus(self, cmd, timeout=-1):
            r"""Execute a shell command.

            Returns:
                (strip/cleaned \r) output
            """
            output = self.run_command(cmd, timeout)
            output = output.replace("\r\n", "\n")
            if self.echo:
                # remove the command
                idx = output.find(cmd)
                if idx == -1:
                    logging.warning(
                        "Didn't find command ('%s') in expected output ('%s')",
                        cmd,
                        output,
                    )
                else:
                    # Remove up to and including the command from the output stream
                    output = output[idx + len(cmd) :]

            return output.replace("\r", "").strip()

        def cmd_status(self, cmd, timeout=-1):
            r"""Execute a shell command.

            Returns:
                status and (strip/cleaned \r) output
            """
            # Run the command getting the output
            output = self.cmd_nostatus(cmd, timeout)

            # Now get the status
            scmd = "echo $?"
            rcstr = self.run_command(scmd)
            rcstr = rcstr.replace("\r\n", "\n")
            if self.echo:
                # remove the command
                idx = rcstr.find(scmd)
                if idx == -1:
                    if self.echo:
                        logging.warning(
                            "Didn't find status ('%s') in expected output ('%s')",
                            scmd,
                            rcstr,
                        )
                    try:
                        rc = int(rcstr)
                    except Exception:
                        rc = 255
                else:
                    rcstr = rcstr[idx + len(scmd) :].strip()
            try:
                rc = int(rcstr)
            except ValueError as error:
                logging.error(
                    "%s: error with expected status output: %s: %s",
                    self,
                    error,
                    rcstr,
                    exc_info=True,
                )
                rc = 255
            return rc, output

        def cmd_raises(self, cmd, timeout=-1):
            r"""Execute a shell command.

            Returns:
                (strip/cleaned \r) ouptut

            Raises:
               CalledProcessError: on non-zero exit status
            """
            rc, output = self.cmd_status(cmd, timeout)
            if rc:
                raise CalledProcessError(rc, cmd, output)
            return output


# ---------------------------
# Root level utility function
# ---------------------------


def get_exec_path(binary):
    return commander.get_exec_path(binary)


def get_exec_path_host(binary):
    return commander.get_exec_path(binary)


def get_our_script_path(script):
    # would be nice to find this w/o using a path lookup
    sdir = os.path.dirname(os.path.abspath(__file__))
    spath = os.path.join(sdir, script)
    if os.path.exists(spath):
        return spath
    return get_exec_path(script)


commander = Commander("munet")
roothost = None