path: root/examples/unschroot_proc.py

#!/usr/bin/python3
# Copyright 2025 Helmut Grohne <helmut@subdivi.de>
# SPDX-License-Identifier: LGPL-2.0-or-later

"""Emulate schroot using namespaces and a background session process
sufficiently well that sbuild can deal with it but not any better. For using
it with sbuild, your sbuildrc should contain:

    $chroot_mode = "schroot";
    $schroot = "/path/to/unschroot";

It may automatically discover chroots from ~/.cache/sbuild in the layout that
sbuild's unshare mode consumes, but you may also create a
~/.config/unschroot.ini to apply more detailed configuration.

State and sessions are retained via a background process and namespaces as well
as a varlink socket below $RUNTIME_DIR/unschroot.
"""

import argparse
import asyncio
import collections.abc
import configparser
import contextlib
import functools
import getpass
import itertools
import json
import os
from pathlib import Path, PurePath
import pwd
import signal
import socket
import sys
import tempfile
import typing
import uuid

import platformdirs

if __file__.split("/")[-2:-1] == ["examples"]:
    sys.path.insert(0, "/".join(__file__.split("/")[:-2]))

import asyncvarlink
import linuxnamespaces
import pconr


# Ignore $HOME as sbuild sets to something invalid
HOME = Path(pwd.getpwuid(os.getuid()).pw_dir)
UNSCHROOT_CONFIG = HOME / ".config/unschroot.ini"
CACHE_SBUILD = HOME / ".cache/sbuild"
CACHE_DIRECTORY_CHROOTS = HOME / ".cache/directory_chroots"


_P = typing.ParamSpec("_P")
_T = typing.TypeVar("_T")


def make_relative(path: PurePath) -> PurePath:
    """Make a given path relative to / without any traversal (..). If it
    already is relative, it is assumed to be relative to /.
    """
    parts: list[str] = []
    for part in path.parts:
        if part == "..":
            try:
                parts.pop()
            except IndexError:
                pass
        elif part not in (".", "/"):
            parts.append(part)

    return PurePath(*parts)


def async_as_sync(
    func: collections.abc.Callable[
        _P, collections.abc.Coroutine[object, object, _T]
    ],
) -> collections.abc.Callable[_P, _T]:
    """Turn an async function into a sync one by running it its own loop."""

    @functools.wraps(func)
    def wrapped(*args: _P.args, **kwargs: _P.kwargs) -> _T:
        return asyncio.run(func(*args, **kwargs))

    return wrapped


def runtime_path() -> Path:
    """Return the location where IPC sockets are to be stored."""
    return platformdirs.user_runtime_path("unschroot")


def guess_comp(name: str) -> str | None:
    """Guess the compression scheme of a tar archive from its name. Returns
    None for uncompressed archives and raises a ValueError when the name does
    not indicate a tar archive.
    """
    if name.endswith(".tar"):
        return None
    if name.endswith((".tar.bz2", ".tbz", ".tbz2")):
        return "bzip2"
    if name.endswith((".tar.gz", ".tgz")):
        return "gz"
    if name.endswith(".tar.lzma"):
        return "lzma"
    if name.endswith((".tar.xz", ".txz")):
        return "xz"
    if name.endswith((".tar.zst", ".tar.zstd", ".tzst", ".tzstd")):
        return "zstd"
    raise ValueError("tar archive not recognized")


class ChrootBase:
    """Base class for chroots from a schroot way of looking at them."""

    namespace: str

    def __init__(self, name: str):
        self.name = name

    def aliases(self) -> set[str]:
        """Return the set of alternative names recorded for this chroot."""
        return set()

    def infodata(self) -> dict[str, str]:
        """Return a mapping with information needed for schroot -i."""
        return {
            "Name": self.name,
            "Aliases": " ".join(sorted(self.aliases())),
            "Session Managed": "true",
        }

    def infostr(self) -> str:
        """Construct the schroot -i output for this chroot."""
        return f"--- {self.namespace} ---\n" + "".join(
            map("%-22s %s\n".__mod__, self.infodata().items())
        )


class SourceChroot(ChrootBase):
    """Represent a schroot source chroot to be instantiated into a session."""

    namespace = "Chroot"

    def __init__(self, name: str, config: collections.abc.Mapping[str, str]):
        super().__init__(name)
        self.config = config

    @typing.override
    def aliases(self) -> set[str]:
        try:
            aliasstr = self.config["aliases"]
        except KeyError:
            return set()
        return set(map(str.strip, aliasstr.split(",")))

    @typing.override
    def infodata(self) -> dict[str, str]:
        data = super().infodata()
        if (
            self.config.get("rootfstype") in ("none", None)
            and self.config.get("rootfsextract")
        ):
            data["Type"] = "file"
            data["File"] = self.config["rootfsextract"]
        elif (
            self.config["rootfstype"] == "bind"
            and self.config.get("rootfsdir")
        ):
            data["Type"] = "directory"
            data["Directory"] = self.config["rootfsdir"]
        elif (
            self.config["rootfstype"].startswith("fuse.")
            and self.config.get("rootfsdev")
        ):
            data["Type"] = "file"
            data["File"] = self.config["rootfsdev"]
        else:
            assert False, f"unexpected chroot configuration {self.config!r}"
        return data

    async def mount_fuse(
        self,
        conn: pconr.ContainerConnection,
        device: Path,
        target: PurePath,
        subtype: str,
        options: list[str],
    ) -> None:
        """Mount a single fuse filesystem in the proxied process."""
        driver = {
            "bind": "bindfs",
            "erofs": "erofsfuse",
            "ext4": "fuse2fs",
            "squashfs": "squashfuse",
        }[subtype]
        with await conn.mount_fuse(
            os.fspath(target),
            source=os.fspath(device),
            options={
                "rootmode": "040755",
                "user_id": "0",
                "group_id": "0",
                "allow_other": None,
            },
            fstype=subtype,
        ) as fusefd:
            argv = [
                driver,
                "-f",
                *itertools.chain.from_iterable(
                    zip(itertools.repeat("-o"), options)
                ),
                os.fspath(device),
                f"/dev/fd/{fusefd.fileno()}",
            ]

            @linuxnamespaces.run_in_fork.now
            def _() -> None:
                linuxnamespaces.close_all_but([2, fusefd])
                fd = os.open("/dev/null", os.O_RDONLY)
                if fd != 0:
                    os.dup2(fd, 0)
                    os.close(fd)
                fd = os.open("/dev/null", os.O_WRONLY)
                if fd != 1:
                    os.dup2(fd, 1)
                    os.close(fd)
                os.execvp(driver, argv)

    async def extract_tar(
        self, conn: pconr.ContainerConnection, tarpath: Path
    ) -> None:
        """Extract the given tar archive to the working directory."""
        command = [
            "tar",
            "--xattrs",
            "--sparse",
            "--preserve-permissions",
            "--same-owner",
            "--extract",
        ]
        if comptype := guess_comp(tarpath.name):
            command.append(f"--{comptype}")
        with tarpath.open("rb") as tarf:
            with await conn.fork() as tarconn:
                await tarconn.exec(command, fds=[tarf, 1, 2])
                assert tarconn.pidfd is not None
                if (status := await conn.waitpidfd(tarconn.pidfd)) != 0:
                    raise ValueError(f"tar exited {status}")

    async def mount(
        self, conn: pconr.ContainerConnection, backingdir: Path | None
    ) -> None:
        """Create the root filesystem and chdir the supervisor to it."""
        match self.config.get("backingstore"):
            case "tmpfs" | None:
                await conn.mount_tmpfs("/opt", options={"mode": "0755"})
                await conn.chdir("/opt")
            case "directory":
                assert backingdir is not None
                await conn.chdir(backingdir)
            case _:
                raise NotImplementedError("unsupported backingstore")
        await conn.mkdir("lower")
        match self.config.get("rootfstype", "").split("."):
            case ["none"] | [""]:
                pass
            case ["bind"]:
                await conn.bind_mount(
                    Path(self.config["rootfsdir"]).expanduser(),
                    PurePath("lower"),
                    readonly=configparser.ConfigParser.BOOLEAN_STATES[
                        self.config.get("overlayfs", "false").lower()
                    ],
                )
            case ["fuse", subtype]:
                await self.mount_fuse(
                    conn,
                    Path(self.config["rootfsdev"]).expanduser(),
                    PurePath("lower"),
                    subtype,
                    [],
                )
            case _:
                raise NotImplementedError("unsupported rootfstype")
        if self.config.get("overlayfs"):
            assert self.config.get("rootfstype") not in ("none", None)
            await conn.mkdir("work")
            await conn.mkdir("upper")
            await conn.mkdir("mnt")
            # Mount to a subdir such that we may chdir("..")
            await conn.mkdir("mnt/mnt")
            await conn.mount_overlayfs(
                "lower",
                "mnt/mnt",
                upper="upper",
                work="work",
                options={"userxattr": None},
            )
            await conn.chdir("mnt/mnt")
        if self.config.get("rootfsextract"):
            await self.extract_tar(
                conn, Path(self.config["rootfsextract"]).expanduser()
            )
        for mount in self.config.get("mounts", "").split():
            [devstr, targetstr, fstypestr, *options] = mount.split(",")
            device = Path(devstr).expanduser()
            target = make_relative(Path(targetstr))
            fstype = fstypestr.split(".", 1)
            match fstype:
                case ["bind"]:
                    readonly = False
                    if options == ["ro"]:
                        readonly = True
                    elif options:
                        raise NotImplementedError(
                            "unsupported bind mount option"
                        )
                    await conn.bind_mount(device, target, readonly=readonly)
                case ["fuse", subtype]:
                    await self.mount_fuse(
                        conn, device, target, subtype, options
                    )
                case _:
                    raise NotImplementedError(
                        f"unsupported filesystem type {fstypestr}"
                    )


class SessionChroot(ChrootBase):
    """Represent a schroot session. It's name is the basename of the IPC
    socket.
    """

    namespace = "Session"

    def __init__(self, name: str, source: SourceChroot):
        super().__init__(name)
        self.source = source
        self.cleanup: Path | None = None

    @classmethod
    def load_existing(
        cls, name: str, chroots: dict[str, SourceChroot]
    ) -> "SessionChroot":
        """Load the state for the session given by name."""
        with open(runtime_path() / name / "state", encoding="utf8") as fh:
            state = json.load(fh)
        source = chroots[state["source"]]
        session = SessionChroot(name, source)
        try:
            session.cleanup = Path(state["cleanup"])
        except KeyError:
            pass
        return session

    @functools.cached_property
    def session_directory(self) -> Path:
        return runtime_path() / self.name

    @functools.cached_property
    def socket_path(self) -> Path:
        return self.session_directory / "varlink"

    @functools.cached_property
    def state_path(self) -> Path:
        return self.session_directory / "state"

    def save_state(self) -> None:
        state = {"source": self.source.name}
        if self.cleanup:
            state["cleanup"] = os.fspath(self.cleanup)
        with open(self.state_path, "w", encoding="utf8") as fh:
            json.dump(state, fh)

    def delete(self) -> None:
        """Delete state information from the runtime path."""
        self.state_path.unlink()
        self.socket_path.unlink()
        self.session_directory.rmdir()

    @typing.override
    def infodata(self) -> dict[str, str]:
        data = super().infodata()
        data["Session Purged"] = "true"
        data["Type"] = "unshare"
        # It's a gross lie. It has to exist as a directory or sbuild won't
        # work.
        data["Location"] = "/opt"
        data["Original Chroot Name"] = self.source.name
        return data


def load_config() -> configparser.ConfigParser:
    config = configparser.ConfigParser(interpolation=None, delimiters=("=",))
    config.read([UNSCHROOT_CONFIG])
    return config


def scan_chroots() -> dict[str, SourceChroot]:
    """Scan chroots:
    * ~/.config/unschroot.ini trumps
    * ~/.cache/sbuild automatic tar-based
    * ~/.cache/directory_chroots automatic overlay
    """
    chrootmap = {
        name: SourceChroot(name, chrootconfig)
        for name, chrootconfig in load_config().items()
        if name != "DEFAULT"
    }

    for loc in (CACHE_SBUILD, CACHE_DIRECTORY_CHROOTS):
        if not loc.is_dir():
            continue
        chroots = []
        aliases: dict[str, set[str]] = {}
        for path in loc.iterdir():
            if path.is_symlink():
                alias = path.name.split(".", 1)[0]
                aliases.setdefault(str(path.readlink()), set()).add(alias)
            else:
                chroots.append(path)
        for path in chroots:
            if loc == CACHE_SBUILD and path.is_file():
                try:
                    guess_comp(path.name)
                except ValueError:
                    continue
                chrootconfig = {
                    "rootfstype": "none",
                    "rootfsextract": str(path),
                }
            elif path.is_dir():
                chrootconfig = {
                    "rootfstype": "bind",
                    "rootfsdir": str(path),
                    "overlayfs": True,
                }
            else:
                continue

            if chrootaliases := aliases.get(path.name):
                chrootconfig["aliases"] = ",".join(sorted(chrootaliases))

            chroot = SourceChroot(path.name.split(".", 1)[0], chrootconfig)
            if chroot.name not in chrootmap:
                chrootmap[chroot.name] = chroot

    for chroot in list(chrootmap.values()):
        for alias in chroot.aliases():
            if alias not in chrootmap:
                chrootmap[alias] = chroot

    return chrootmap


def scan_sessions(
    chrootmap: dict[str, SourceChroot]
) -> dict[str, SessionChroot]:
    sessionmap = {}
    if (rtdir := runtime_path()).is_dir():
        for session in rtdir.iterdir():
            if session.is_dir():
                sessionmap[session.name] = SessionChroot.load_existing(
                    session.name, chrootmap
                )
    return sessionmap


def do_info(args: argparse.Namespace) -> None:
    """Show information about selected chroots"""
    chrootmap = scan_chroots()
    sessionmap = scan_sessions(chrootmap)
    chroots: collections.abc.Iterable[ChrootBase]
    if args.chroot:
        try:
            chroots = [chrootmap[args.chroot.removeprefix("chroot:")]]
        except KeyError:
            chroots = [sessionmap[args.chroot.removeprefix("session:")]]
    else:
        chroots = itertools.chain(
            (
                chroot
                for name, chroot in chrootmap.items()
                if chroot.name == name
            ),
            sessionmap.values()
        )
    sys.stdout.write("\n".join(chroot.infostr() for chroot in chroots))


def do_list(args: argparse.Namespace) -> None:
    """List available chroots"""
    chrootmap = scan_chroots()
    sessionmap = scan_sessions(chrootmap)
    sys.stdout.write(
        "".join(
            itertools.chain(
                (
                    f"chroot:{name}\n"
                    for name in chrootmap
                    if (
                        name == args.chroot.removeprefix("chroot:")
                        if args.chroot
                        else "Chroot" in args.all
                    )
                ),
                (
                    f"session:{name}\n"
                    for name in sessionmap
                    if (
                        name == args.chroot.removeprefix("session:")
                        if args.chroot
                        else "Session" in args.all
                    )
                ),
            ),
        ),
    )


async def create_session(
    source: SourceChroot, session: SessionChroot
) -> pconr.SpawnedContainerConnection:
    uidmap = [
        linuxnamespaces.IDAllocation.loadsubid("uid").allocatemap(65536),
        linuxnamespaces.IDMapping(65536, os.getuid(), 1),
    ]
    gidmap = [
        linuxnamespaces.IDAllocation.loadsubid("gid").allocatemap(65536),
        linuxnamespaces.IDMapping(65536, os.getgid(), 1),
    ]
    session.socket_path.parent.mkdir(exist_ok=True, parents=True)
    supervisor = await pconr.launch_manager(exit_on_close=False)

    try:
        await supervisor.listen_varlink(session.socket_path)
        await supervisor.setpriv(subreaper=True)
        common_namespaces = (
            linuxnamespaces.CloneFlags.NEWUSER
            | linuxnamespaces.CloneFlags.NEWNS
        )
        await supervisor.unshare(common_namespaces)
        linuxnamespaces.newidmaps(supervisor.pid, uidmap, gidmap)
        tdir: Path | None = None
        if source.config.get("backingstore") == "directory":
            tdir = Path(tempfile.mkdtemp(prefix="unschroot"))
            session.cleanup = tdir
        await supervisor.setpriv(uid=0, gid=0, groups=[0], dumpable=True)
        await source.mount(supervisor, tdir)
        await supervisor.populate_dev("dev")
        # We may or may not unshare a network namespace. Pretend that we don't,
        # and lock down sysfs more than we might have to.
        await supervisor.populate_sys("sys", common_namespaces)
        await supervisor.write_text(
            "etc/hosts",
            f"""127.0.0.1 localhost
127.0.1.1 {socket.gethostname()}
::1 localhost ip6-localhost ip6-loopback
""",
        )
        session.save_state()
    except:
        supervisor.close()
        await asyncio.sleep(0)  # close uses call_soon
        raise
    return supervisor


@async_as_sync
async def do_begin_session(args: argparse.Namespace) -> None:
    """Begin a session; returns the session ID"""
    chrootmap = scan_chroots()
    source = chrootmap[args.chroot.removeprefix("chroot:")]
    session = SessionChroot(args.session_name or str(uuid.uuid4()), source)
    supervisor = await create_session(source, session)
    supervisor.close()
    await asyncio.sleep(0)  # close uses call_soon
    print(session.name)


def parse_passwd(passwd: str, user: str | int) -> pwd.struct_passwd:
    """Look up the passwd record for a given user (name or uid) in the given
    passwd database content. Similar to getpwnam and getpwduid, raise a
    KeyError if the user cannot be found. Of course, this does not consider
    any LDAP or NIS database.
    """
    if isinstance(user, str) and user.isdigit():
        user = int(user)
    for line in passwd.splitlines():
        parts = line.split(":")
        # Skip over invalid records and bad data
        if len(parts) != 7:
            continue
        try:
            uid = int(parts[2])
            gid = int(parts[3])
        except ValueError:
            continue
        if uid < 0 or gid < 0 or uid > 0x1FFFFFFF or gid > 0x1FFFFFFF:
            continue
        if user == parts[0] if isinstance(user, str) else user == uid:
            return pwd.struct_passwd((*parts[:2], uid, gid, *parts[4:]))
    raise KeyError(user)


def make_environment(*, preserve: bool, root: bool) -> dict[str, str]:
    """Compute a suitable environment dictionary for a target process.  If
    preserve is set, the outer environment is preserved. The PATH added if
    missing. It contains sbin directories if generating for root.
    """
    env = dict(os.environ) if preserve else {}
    if not (orig_path := env.get("PATH")):
        if root:
            env["PATH"] = "/usr/sbin:/sbin:/usr/bin:/bin"
        else:
            env["PATH"] = "/usr/bin:/bin"
    elif root and ":/usr/sbin:" not in f":{orig_path}:":
        env["PATH"] = f"{orig_path}:/usr/sbin"
    return env


async def run_command(
    args: argparse.Namespace,
    supervisor: pconr.ContainerConnection,
) -> int:
    namespaces = (
        linuxnamespaces.CloneFlags.NEWUSER
        | linuxnamespaces.CloneFlags.NEWNS
        | linuxnamespaces.CloneFlags.NEWIPC
        | linuxnamespaces.CloneFlags.NEWPID
    )
    if args.isolate_network:
        namespaces |= linuxnamespaces.CloneFlags.NEWNET

    with contextlib.ExitStack() as stack:
        intermediate = stack.enter_context(await supervisor.fork())
        passwd_record = parse_passwd(
            await supervisor.read_text("etc/passwd"), args.user
        )
        environ = make_environment(
            preserve=args.preserve_environment, root=passwd_record.pw_uid == 0
        )

        await intermediate.unshare(namespaces)
        identity = linuxnamespaces.IDMapping.identity(0, 65536)
        await supervisor.newidmaps(intermediate.pid, [identity], [identity])
        # In order for pivot_root to work, the new root must be a mount point,
        # but as we unshared both a user and mount namespace, the working
        # directory no longer is a mount point.
        await intermediate.bind_mount(".", ".", umount_old=True)
        await intermediate.chdir(await intermediate.getcwd())
        # In order to be able to mount a real proc later, the original /proc
        # must be visible somewhere inside our container. Temporarily mount it
        # to /bin.
        await intermediate.bind_mount("/proc", "bin")
        await intermediate.mount_dev_pts("dev/pts")
        await intermediate.pivot_root(".", ".")
        if args.isolate_network:
            await intermediate.enable_loopback_if()
        target = stack.enter_context(await intermediate.fork())
        # Now that we have forked and thus entered our PID namespace, we
        # may mount /proc.
        await target.mount_proc("proc")
        # Get rid of the mount that granted us mounting /proc.
        await target.umount(target="bin", detach=True)

        if passwd_record.pw_uid != 0 or passwd_record.pw_gid != 0:
            await target.setpriv(
                uid=passwd_record.pw_uid, gid=passwd_record.pw_gid
            )

        if args.directory:
            await target.chdir(args.directory)
        elif args.command:
            await target.chdir(os.getcwd())
        else:
            try:
                await target.chdir(os.getcwd())
            except pconr.ContainerOSError:
                try:
                    await target.chdir(environ["HOME"])
                except (KeyError, pconr.ContainerOSError):
                    try:
                        await target.chdir(passwd_record.pw_dir)
                    except pconr.ContainerOSError:
                        pass  # we're already at /
        # Termination reparents to the supervisor who is a subreaper.
        await intermediate.exit()
        assert intermediate.pidfd is not None
        await supervisor.waitpidfd(intermediate.pidfd)
        # The container may change this, but it's still useful for
        # robustness when it does not.
        await target.setpriv(pdeathsig=signal.SIGKILL)
        await target.exec(
            args.command or [environ.get("SHELL", "bash")],
            fds=[
                asyncvarlink.FileDescriptor(0),
                asyncvarlink.FileDescriptor(1),
                asyncvarlink.FileDescriptor(2),
            ],
            environ=environ,
        )
        assert target.pidfd is not None
        return await supervisor.waitpidfd(target.pidfd)


@async_as_sync
async def do_automatic_session(args: argparse.Namespace) -> None:
    """Begin, run and end a session automatically"""
    async with contextlib.AsyncExitStack() as stack:
        chrootmap = scan_chroots()
        source = chrootmap[args.chroot.removeprefix("chroot:")]
        session = SessionChroot(str(uuid.uuid4()), source)
        conn = await create_session(source, session)
        session.socket_path.parent.mkdir(exist_ok=True, parents=True)
        stack.callback(session.delete)
        stack.push_async_callback(asyncio.sleep, 0)  # close uses call_soon
        stack.callback(conn.close)
        stack.push_async_callback(conn.exit)
        if session.cleanup:
            stack.push_async_callback(
                conn.remove_tree, session.cleanup, umount=True
            )
        sys.exit(await run_command(args, conn))


@async_as_sync
async def do_run_session(args: argparse.Namespace) -> None:
    """Run an existing session"""
    session = SessionChroot.load_existing(args.chroot, scan_chroots())
    async with contextlib.AsyncExitStack() as stack:
        stack.push_async_callback(asyncio.sleep, 0)  # close uses call_soon
        conn = stack.enter_context(
            await pconr.connect_manager(session.socket_path)
        )
        sys.exit(await run_command(args, conn))


@async_as_sync
async def do_end_session(args: argparse.Namespace) -> None:
    """End an existing session"""
    session = SessionChroot.load_existing(args.chroot, scan_chroots())
    with contextlib.closing(
        await pconr.connect_manager(session.socket_path)
    ) as conn:
        if session.cleanup:
            await conn.remove_tree(session.cleanup, umount=True)
        await conn.exit()
    session.delete()


def main() -> None:
    parser = argparse.ArgumentParser()
    group = parser.add_mutually_exclusive_group(required=False)
    group.add_argument(
        "--automatic-session",
        dest="subcommand",
        action="store_const",
        default=do_automatic_session,
        help=do_automatic_session.__doc__,
    )
    for comm in (
        "begin-session", "end-session", "info", "list", "run-session"
    ):
        func = globals()["do_" + comm.replace("-", "_")]
        group.add_argument(
            f"-{comm[0]}",
            f"--{comm}",
            dest="subcommand",
            action="store_const",
            const=func,
            help=func.__doc__,
        )
    parser.add_argument(
        "-a",
        "--all",
        action="store_const",
        default=("Chroot",),
        const=("Chroot", "Session"),
    )
    parser.add_argument(
        "--all-chroots",
        dest="all",
        action="store_const",
        const=("Chroot",),
    )
    parser.add_argument(
        "--all-sessions",
        dest="all",
        action="store_const",
        const=("Session",),
    )
    parser.add_argument(
        "-c",
        "--chroot",
        dest="chroot",
        action="store",
        help="Use specified chroot",
    )
    parser.add_argument("-d", "--directory", action="store")
    parser.add_argument("-n", "--session-name", action="store", default=None)
    parser.add_argument("-p", "--preserve-environment", action="store_true")
    parser.add_argument("-q", "--quiet", action="store_true")
    parser.add_argument(
        "-u", "--user", action="store", default=getpass.getuser()
    )
    parser.add_argument("--isolate-network", action="store_true")
    parser.add_argument("command", nargs="*")
    args = parser.parse_args()
    assert args.subcommand is not None
    args.subcommand(args)


if __name__ == "__main__":
    main()