#!/usr/bin/python3 # This Source Code Form is subject to the terms of the Mozilla Public # License, v. 2.0. If a copy of the MPL was not distributed with this # file, You can obtain one at https://mozilla.org/MPL/2.0/. """bubblewrap stand-in for pressure-vessel on hosts without user namespaces. Steam runs its client helper and every game through pressure-vessel, which builds a bwrap(1) command line describing a container -- bind mounts, symlinks, generated files, environment edits, and the command to run -- and executes it. bwrap needs a user namespace or CAP_SYS_ADMIN, and a container's default seccomp profile grants neither, so pressure-vessel fails at that very last step. pressure-vessel consults ``$BWRAP`` when its bundled bwrap does not work; this program takes that place, accepts the same command line, and runs the command in an equivalent tree without namespaces. The bwrap description is materialized as a root directory under the runtime directory: ``--dir``, ``--symlink``, ``--tmpfs``, and the data file options become real entries, and the bind mounts are handed to one of two executors. * proot, the default: every bind is a proot ``-b`` binding and the root directory is the guest root, which confines paths for every process rather than only those that go through libc. It needs ptrace, and it loads a program through a loader of its own, whose temp file every released proot leaves named in AT_EXECFN -- the multi-call coreutils of Ubuntu 25.10 and later read that to decide which tool they are, so nothing that runs them works under such a build. A proot is probed for that before it is used. * fakechroot, where no proot can do that: an LD_PRELOAD library prefixes every path a dynamically linked program opens with the root directory, and binds become symlinks in it. The dynamic loader and the kernel are not rewritten, so what they follow must resolve on the host: library search paths are given as root-prefixed host paths, absolute symlink targets inside the root are prefixed with it, and the ld.so cache regeneration is replaced by a complete LD_LIBRARY_PATH. The host's glibc is what the container runs on in this mode, which pressure-vessel already chooses whenever the host's is newer than the runtime's. It also runs the one container proot cannot, the Steam client's browser helper, whatever the rest of the launch uses. Real bubblewrap is used whenever it works: pressure-vessel tries its own copy first and reads ``$BWRAP`` only after that fails. ``PROOT_BWRAP_BACKEND`` forces ``proot`` or ``fakechroot``; ``PROOT_BWRAP_DEBUG`` prints the executor command line. """ import fcntl import hashlib import os import shutil import signal import stat import sys import tempfile # json and subprocess are imported where they are used. Every Steam container # launch runs this several times before anything starts, and neither module is # needed on the path that a launch actually takes. VERSION = "bubblewrap 0.9.0 (proot-bwrap)" PROGRAM = "proot-bwrap" # Options with no argument. FLAGS = { "--unshare-all", "--share-net", "--unshare-user", "--unshare-user-try", "--unshare-ipc", "--unshare-pid", "--unshare-net", "--unshare-uts", "--unshare-cgroup", "--unshare-cgroup-try", "--disable-userns", "--assert-userns-disabled", "--clearenv", "--new-session", "--die-with-parent", "--as-pid-1", "--level-prefix", "--not-a-security-boundary", } ONE_ARG = { "--args", "--argv0", "--userns", "--userns2", "--pidns", "--uid", "--gid", "--hostname", "--chdir", "--unsetenv", "--lock-file", "--sync-fd", "--remount-ro", "--exec-label", "--file-label", "--proc", "--dev", "--tmpfs", "--mqueue", "--dir", "--seccomp", "--add-seccomp-fd", "--block-fd", "--userns-block-fd", "--info-fd", "--json-status-fd", "--cap-add", "--cap-drop", "--perms", "--size", "--tmp-overlay", "--ro-overlay", "--overlay-src", } TWO_ARGS = { "--setenv", "--bind", "--bind-try", "--dev-bind", "--dev-bind-try", "--ro-bind", "--ro-bind-try", "--bind-fd", "--ro-bind-fd", "--file", "--bind-data", "--ro-bind-data", "--symlink", "--chmod", } THREE_ARGS = {"--overlay"} # Accepted for compatibility with no effect: each needs a namespace, a # capability set, a seccomp filter, or a security label, and nothing here has one. INERT = { "--unshare-all", "--unshare-user", "--unshare-user-try", "--unshare-ipc", "--unshare-pid", "--unshare-net", "--unshare-uts", "--unshare-cgroup", "--unshare-cgroup-try", "--disable-userns", "--assert-userns-disabled", "--as-pid-1", "--userns", "--userns2", "--pidns", "--uid", "--gid", "--hostname", "--remount-ro", "--cap-add", "--cap-drop", "--seccomp", "--add-seccomp-fd", "--exec-label", "--file-label", } BINDS = {"--bind", "--bind-try", "--dev-bind", "--dev-bind-try", "--ro-bind", "--ro-bind-try", "--bind-fd", "--ro-bind-fd"} DATA_FILES = {"--file", "--bind-data", "--ro-bind-data"} # Options that create the next filesystem entry, which --perms applies to. CREATES = {"--dir", "--tmpfs", "--symlink"} | DATA_FILES HELP = f"""usage: {PROGRAM} [OPTIONS...] [--] COMMAND [ARGS...] bubblewrap-compatible front end that runs COMMAND in the described tree through proot or fakechroot instead of namespaces. Accepted options: """ class UsageError(Exception): """A launch bubblewrap itself would refuse.""" def log(message): sys.stderr.write(f"{PROGRAM}: {message}\n") sys.stderr.flush() def debug(message): if os.environ.get("PROOT_BWRAP_DEBUG"): log(message) def read_fd(fd): """Read an inherited descriptor to its end, from its current position.""" chunks = [] while True: chunk = os.read(fd, 65536) if not chunk: return b"".join(chunks) chunks.append(chunk) def parse(argv): """Turn bubblewrap's command line into ordered operations and a command. Returns: A tuple of (operations, command): operations are (option, arguments, perms) triples in command-line order, perms being the octal mode a preceding ``--perms`` requested for entries that option creates, or None. Raises: UsageError: on an unknown option or a missing argument. """ args = list(argv) ops = [] perms = None i = 0 while i < len(args): arg = args[i] if arg == "--": i += 1 break if not arg.startswith("--"): break if arg == "--args": if i + 1 >= len(args): raise UsageError("--args takes an argument") fd = int(args[i + 1]) spliced = [a.decode("utf-8", "surrogateescape") for a in read_fd(fd).split(b"\0") if a] os.close(fd) args[i:i + 2] = spliced continue if arg in FLAGS: count = 0 elif arg in ONE_ARG: count = 1 elif arg in TWO_ARGS: count = 2 elif arg in THREE_ARGS: count = 3 else: raise UsageError(f"Unknown option {arg}") if i + count >= len(args): raise UsageError(f"{arg} takes {count} argument(s)") values = args[i + 1:i + 1 + count] i += 1 + count if arg == "--perms": perms = int(values[0], 8) continue ops.append((arg, values, perms if arg in CREATES else None)) if arg in CREATES: perms = None command = args[i:] return ops, command def runtime_base(): """The per-user directory that holds one root tree per launch. Without a usable ``$XDG_RUNTIME_DIR`` it lives in the shared temporary directory, and is used there only as a directory of this user's that no one else can write: whoever could would control every tree built in it. Raises: UsageError: when that directory is not one. """ base = os.environ.get("XDG_RUNTIME_DIR") if not base or not os.access(base, os.W_OK): base = os.path.join(tempfile.gettempdir(), f"{PROGRAM}-{os.getuid()}") try: os.mkdir(base, 0o700) except FileExistsError: pass info = os.lstat(base) if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.geteuid() or info.st_mode & 0o022: raise UsageError(f"{base} is not a directory of this user's alone; set XDG_RUNTIME_DIR") base = os.path.join(base, PROGRAM) os.makedirs(base, 0o700, exist_ok=True) return base def stage_directory(base, ops): """The directory this launch builds its root in, named for the container. The name is derived from what the container is rather than from when it started, because a Wine prefix records the container's root as the path behind its Z: drive and is reused by every later launch: a directory named per launch would leave that drive dangling. The same container therefore comes back to the same directory, and one that is still in use gets a fresh directory beside it. Returns: A path whose directory is empty and exists. """ key = hashlib.sha256() copy = runtime_copy(ops) if copy is not None: key.update(os.path.dirname(os.path.dirname(copy)).encode()) for option, values, _ in ops: if option in BINDS or option in ("--dir", "--tmpfs", "--symlink") or option in DATA_FILES: key.update(b"\0" + option.encode() + b"\0" + values[-1].encode()) path = os.path.join(base, key.hexdigest()[:16]) if in_use(path): path = tempfile.mkdtemp(prefix=os.path.basename(path) + ".", dir=base) else: shutil.rmtree(path, ignore_errors=True) os.makedirs(path, 0o700) with open(os.path.join(path, "pid"), "w") as f: f.write(f"{os.getpid()}\n") return path def in_use(path): """Whether the process that built this root directory is still running.""" try: with open(os.path.join(path, "pid")) as f: pid = int(f.read().strip() or "0") except (OSError, ValueError): return False if pid <= 0 or pid == os.getpid(): return False try: os.kill(pid, 0) except ProcessLookupError: return False except PermissionError: pass return True def collect_garbage(base): """Remove the trees of launches whose supervisor is gone.""" for name in os.listdir(base): path = os.path.join(base, name) if os.path.isdir(path) and not in_use(path): shutil.rmtree(path, ignore_errors=True) def proot_candidates(): """proot executables to try, best first, each named once. The build beside this program comes first: the installer puts one there that names the program in AT_EXECFN, and an image that also carries a proot-apps runner has that runner's copy earlier on PATH, which does not. Every PATH entry is offered rather than the first, since the one that answers is worth more than the one that comes first. """ override = os.environ.get("PROOT_BWRAP_PROOT") if override: yield override return here = os.path.dirname(os.path.realpath(__file__)) home = os.path.expanduser("~") paths = [os.path.join(here, "proot"), os.path.join(home, ".local", "bin", "proot")] paths.extend(os.path.join(directory, "proot") for directory in (os.environ.get("PATH") or os.defpath).split(":") if directory) paths.extend(("/opt/proot-apps-cap/proot", "/opt/proot-apps/proot")) seen = set() for path in paths: real = os.path.realpath(path) if real in seen or not os.access(path, os.X_OK): continue seen.add(real) yield path # Reads AT_EXECFN the way a program does: through prctl(PR_GET_AUXV), and # through /proc/self/auxv on the kernels older than 6.4 that have no such # prctl. EXECFN_PROBE = """\ import ctypes, struct, sys width = struct.calcsize('P') form = 'Q' if width == 8 else 'I' vector = b'' try: buffer = ctypes.create_string_buffer(4096) if ctypes.CDLL(None).prctl(0x41555856, buffer, len(buffer), 0, 0) > 0: vector = buffer.raw except (OSError, AttributeError, ValueError): pass if not vector: with open('/proc/self/auxv', 'rb') as handle: vector = handle.read() for start in range(0, len(vector) - 2 * width + 1, 2 * width): key, value = struct.unpack(form * 2, vector[start:start + 2 * width]) if key == 0: break if key == 31 and value: sys.stdout.write(ctypes.string_at(value).decode('utf-8', 'replace')) break """ def proot_state(path): """What this proot is good for here: "none", "traces", or "names". Tracing is the first question. The second is whose name a traced program finds in AT_EXECFN, since a proot that leaves its own loader there breaks every applet of a multi-call coreutils, which is what Ubuntu 25.10 and later ship: the name is what they dispatch on. """ import subprocess def run(argv, capture): try: return subprocess.run(argv, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE if capture else subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=60, check=False) except (OSError, subprocess.SubprocessError): return None completed = run([path, sys.executable, "-c", EXECFN_PROBE], True) if completed is not None and completed.returncode == 0: name = completed.stdout.decode("utf-8", "replace").strip() return "names" if name == sys.executable else "traces" completed = run([path, "/bin/true"], False) return "traces" if completed is not None and completed.returncode == 0 else "none" def fakechroot_libraries(): """The fakechroot preload library per multiarch tuple, or an empty dict.""" found = {} for tuple_ in ("x86_64-linux-gnu", "i386-linux-gnu", "aarch64-linux-gnu"): for prefix in ("/usr/lib", "/usr/local/lib"): path = os.path.join(prefix, tuple_, "fakechroot", "libfakechroot.so") if os.path.exists(path): found[tuple_] = path break return found def steam_browser_helper(command): """Whether this container holds the Steam client's browser helper. It is the one container proot cannot run, so it is worth telling apart. The name is matched as a prefix because the client runs the helper through a per-runtime wrapper, `steamwebhelper_sniper_wrap.sh` today, which execs the helper in place. This is the second and last thing here that knows anything about Steam. """ return any(os.path.basename(word).startswith("steamwebhelper") for word in command) def choose_backend(command): """Pick the executor for this container: proot, or fakechroot. proot leads because it confines paths for every process rather than only those that go through libc, and it costs a few per cent of a game's own work. fakechroot takes over where no proot can trace, or where the only proot available leaves its loader named in AT_EXECFN. The choice is per container, because one container cannot use proot at all. One proot process traces every thread it runs, so the syscalls it translates are served one at a time: 40k a second on one thread and 63k on twenty-four, against 805k and 3.6M untraced. The syscalls it ignores are free and the bind count barely registers, so nothing tunes this away. The Steam client gives its browser helper, two dozen Chromium threads deep in path syscalls, about ten seconds to start, and under proot it does not. So the helper runs under fakechroot however the rest of the launch runs. """ forced = os.environ.get("PROOT_BWRAP_BACKEND", "").strip().lower() if forced not in ("", "proot", "fakechroot"): raise UsageError("PROOT_BWRAP_BACKEND must be proot or fakechroot") libraries = fakechroot_libraries() if libraries and steam_browser_helper(command): if forced == "proot": log("the browser helper runs under fakechroot, the only way it " "starts in time") else: debug("the browser helper runs under fakechroot") return "fakechroot", libraries if forced == "fakechroot": if libraries: return "fakechroot", libraries raise UsageError("libfakechroot.so is not installed") stale = None for path in proot_candidates(): state = proot_state(path) if state == "names": return "proot", path if state == "traces": debug(f"{path} leaves its own loader named in AT_EXECFN") stale = stale or path else: debug(f"{path} cannot trace processes here") # A proot that misnames the loader breaks the multi-call coreutils of # current releases, so a working fakechroot is worth more than it is. if libraries and forced != "proot": return "fakechroot", libraries if stale is not None: return "proot", stale if libraries: return "fakechroot", libraries raise UsageError("neither proot can trace here nor is libfakechroot.so installed") class Root: """The container root as a directory tree, built from bubblewrap operations. Attributes: path: The tree's top directory. binds: (source, destination) pairs for the proot executor, in order. """ def __init__(self, stage, fakechroot): self.stage = stage self.path = os.path.join(stage, "root") self.fakechroot = fakechroot self.binds = [] self.excluded = ["/dev", "/proc", "/sys"] # Top-level names a later operation writes under, filled in by apply() self.built_on = set() self.counter = 0 os.mkdir(self.path, 0o755) def excluded_path(self, dest): """Whether a container path lies in a tree fakechroot passes through. A bind of a host path at its own path is one fakechroot need not translate at all, and must not: a symlink in the tree pointing at the same path would send every program that canonicalizes paths itself, such as ``readlink -f``, round in circles. """ dest = os.path.normpath(dest) return any(dest == e or dest.startswith(e + "/") for e in self.excluded) def translate(self, path): """The host path fakechroot would open for a container path.""" if self.excluded_path(path): return path return self.path + os.path.normpath(path) def pass_through(self, dest): """Let fakechroot pass a host path through at its own name. The name is excluded from translation, and the tree still gets a symlink to the host path: a program that follows an absolute symlink it created inside the container reaches the tree root first, since fakechroot prefixes the targets it writes, and the kernel then walks on from there. The root directory is the exception. Excluding it would exclude only itself, since a pass-through path matches at a directory boundary, so the host's own top-level entries are linked into the tree instead and the container becomes the host filesystem, which is what binding the root asks for. Each of those links is passed through as well, or the link and the translation of its target would resolve into each other until the path resolver gives up. An entry a later operation builds on keeps the link without the pass-through, so that operation still lands. """ dest = os.path.normpath(dest) if dest == "/": for name in os.listdir("/"): here = os.path.join(self.path, name) if not os.path.lexists(here): os.symlink("/" + name, here) if "/" + name not in self.built_on: self.excluded.append("/" + name) return if self.excluded_path(dest): return self.excluded.append(dest) here = self.place(dest) if not os.path.lexists(here): os.symlink(dest, here) def scratch(self, kind): """A fresh path under the stage for a tmpfs directory or a data file.""" self.counter += 1 directory = os.path.join(self.stage, kind) os.makedirs(directory, 0o700, exist_ok=True) return os.path.join(directory, str(self.counter)) def host_reference(self, link): """Whether a symlink in the tree points at the host rather than the tree.""" target = os.readlink(link) return target.startswith("/") and not target.startswith(self.path + "/") def mirror(self, link): """Replace a symlink to a host directory with a directory of symlinks. A bind is a symlink under fakechroot, so an entry placed inside a bound directory would otherwise be written into the host directory itself. One level of symlinks keeps the host's entries visible while the tree gains its own. """ target = os.readlink(link) entries = os.listdir(target) os.unlink(link) os.mkdir(link, 0o755) for entry in entries: os.symlink(os.path.join(target, entry), os.path.join(link, entry)) def place(self, dest): """Resolve a container path to its tree path, owning every parent. Symlinks the operations created are followed inside the tree; a fakechroot bind on the way is mirrored so the parent is a directory of the tree's own. Returns: The tree path at which the entry for ``dest`` belongs. """ parts = [p for p in dest.split("/") if p and p != "."] current = self.path for name in parts[:-1]: nxt = os.path.join(current, name) while os.path.islink(nxt): if self.host_reference(nxt): if not self.fakechroot: break self.mirror(nxt) break target = os.readlink(nxt) if target.startswith("/"): nxt = os.path.normpath(target) else: nxt = os.path.normpath(os.path.join(current, target)) if not os.path.lexists(nxt): os.mkdir(nxt, 0o755) current = nxt return os.path.join(current, parts[-1]) if parts else current def remove(self, path): """Drop an entry of the tree's own without following it.""" if os.path.islink(path) or not os.path.isdir(path): os.unlink(path) else: shutil.rmtree(path) def bind(self, option, source, dest): """Apply a bind mount operation.""" if option.endswith("-fd"): source = os.readlink(f"/proc/self/fd/{int(source)}") if not os.path.lexists(source): if option.endswith("-try"): return raise UsageError(f"Can't find source path {source}") if self.fakechroot: if os.path.normpath(source) == os.path.normpath(dest): self.pass_through(dest) return if self.excluded_path(dest): log(f"cannot bind {source} over {dest}, which fakechroot passes through") return here = self.place(dest) if os.path.lexists(here): self.remove(here) os.symlink(source, here) return self.placeholder(dest, os.path.isdir(source)) self.binds.append((source, dest)) def placeholder(self, dest, is_dir): """Give a proot binding a mount point. A binding needs its guest path to exist: in the tree, or, when an earlier binding already covers the path, inside that binding's source if it is a tmpfs directory of the tree's own. A host directory covering it is left alone, and proot glues the mount point in itself. """ covering = None for source, bound in self.binds: if ((dest == bound or dest.startswith(bound.rstrip("/") + "/")) and (covering is None or len(bound) > len(covering[1]))): covering = (source, bound) if covering is None: here = self.place(dest) elif covering[0].startswith(self.stage + "/"): here = os.path.join(covering[0], os.path.relpath(dest, covering[1])) os.makedirs(os.path.dirname(here), exist_ok=True) else: return if os.path.lexists(here): return if is_dir: os.mkdir(here, 0o755) else: open(here, "w").close() def tmpfs(self, dest, perms): """An empty directory of the tree's own, masking whatever was there.""" if self.fakechroot: if self.excluded_path(dest): debug(f"not masking {dest} inside a tree fakechroot passes through") return here = self.place(dest) if os.path.lexists(here): self.remove(here) os.mkdir(here, perms if perms is not None else 0o755) return scratch = self.scratch("tmpfs") os.mkdir(scratch, perms if perms is not None else 0o755) self.placeholder(dest, True) self.binds.append((scratch, dest)) def directory(self, dest, perms): if self.fakechroot and self.excluded_path(dest): if not os.path.isdir(dest): debug(f"not creating {dest} inside a tree fakechroot passes through") return here = self.place(dest) if os.path.isdir(here): if perms is not None and not os.path.islink(here): os.chmod(here, perms) return os.mkdir(here, perms if perms is not None else 0o755) def symlink(self, target, dest): if self.fakechroot and self.excluded_path(dest): debug(f"not linking {dest} inside a tree fakechroot passes through") return here = self.place(dest) if os.path.lexists(here): if os.path.islink(here) and os.readlink(here) == target: return self.remove(here) if self.fakechroot and target.startswith("/"): target = self.path + target os.symlink(target, here) def data_file(self, option, fd, dest, perms): """Write an inherited descriptor's content at a container path.""" content = read_fd(int(fd)) os.close(int(fd)) mode = perms if perms is not None else 0o644 if self.fakechroot and self.excluded_path(dest): log(f"cannot place {dest} inside a tree fakechroot passes through") return if self.fakechroot or option == "--file": here = self.place(dest) if os.path.lexists(here): self.remove(here) with open(os.open(here, os.O_WRONLY | os.O_CREAT | os.O_EXCL, mode), "wb") as f: f.write(content) os.chmod(here, mode) return scratch = self.scratch("data") with open(os.open(scratch, os.O_WRONLY | os.O_CREAT | os.O_EXCL, mode), "wb") as f: f.write(content) os.chmod(scratch, mode) self.placeholder(dest, False) self.binds.append((scratch, dest)) def apply(self, ops): """Apply the filesystem operations in order; the rest is for the runner.""" for _, values, _ in ops: for value in values: if value.startswith("/"): parts = [p for p in value.split("/") if p] if len(parts) > 1: self.built_on.add("/" + parts[0]) for option, values, perms in ops: if option in BINDS: self.bind(option, values[0], values[1]) elif option in ("--proc", "--dev"): self.bind("--dev-bind", "/" + option[2:], values[0]) elif option == "--tmpfs": self.tmpfs(values[0], perms) elif option in ("--dir", "--mqueue"): self.directory(values[0], perms) elif option == "--symlink": self.symlink(values[0], values[1]) elif option in DATA_FILES: self.data_file(option, values[0], values[1], perms) elif option == "--chmod": os.chmod(self.place(values[1]), int(values[0], 8)) elif option in ("--overlay", "--tmp-overlay", "--ro-overlay", "--overlay-src"): raise UsageError(f"{option} is not supported here") def prefix_absolute_symlinks(self, directory): """Point a mutable runtime copy's absolute symlinks into the tree. pressure-vessel edits its copy of the runtime in place, linking host libraries as ``/run/host/...`` and runtime files by their container path. The kernel resolves a symlink target on the host, so under fakechroot every absolute target gets the tree prefix, which the preload library strips again when a program reads the link. """ for parent, dirs, files in os.walk(directory): for name in dirs + files: link = os.path.join(parent, name) if not os.path.islink(link): continue target = os.readlink(link) if target.startswith("/") and not target.startswith(self.path + "/"): os.unlink(link) os.symlink(self.path + target, link) def runtime_copy(ops): """The mutable runtime copy bound at /usr, if pressure-vessel made one. pressure-vessel's ``--copy-runtime`` places it at ``var/tmp-XXXXXX`` under its variable directory and edits it in place; the shared runtime under ``files/`` has no absolute symlinks and is never touched. """ for option, values, _ in ops: if option in BINDS and values[1] == "/usr": copy = os.path.dirname(values[0]) if (os.path.basename(copy).startswith("tmp-") and os.path.basename(os.path.dirname(copy)) == "var" and os.access(copy, os.W_OK)): return copy return None def ld_so_conf_dirs(root): """Library directories the container's ld.so.conf names, in order.""" dirs = [] def read(path, depth=0): if depth > 8: return try: with open(path) as f: lines = f.read().splitlines() except OSError: return for line in lines: line = line.split("#", 1)[0].strip() if not line: continue if line.startswith("include "): pattern = line[len("include "):].strip() if not pattern.startswith("/"): pattern = os.path.join(os.path.dirname(path), pattern) import glob for match in sorted(glob.glob(root + pattern)): read(match, depth + 1) elif line.startswith("/") and line not in dirs: dirs.append(line) read(root + "/etc/ld.so.conf") for tuple_ in ("x86_64-linux-gnu", "i386-linux-gnu"): for prefix in ("/lib", "/usr/lib"): path = os.path.join(prefix, tuple_) if path not in dirs: dirs.append(path) for path in ("/lib", "/usr/lib", "/usr/local/lib"): if path not in dirs: dirs.append(path) return dirs def libdl_lib_tokens(helpers): """What the dynamic loader expands ``$LIB`` to, per multiarch tuple. pressure-vessel ships a helper per architecture that reports it; they run on the host here, since pv-adverb's own attempt goes through fakechroot's dlopen, which cannot expand loader tokens. """ import subprocess tokens = {} for tuple_ in ("x86_64-linux-gnu", "i386-linux-gnu"): helper = os.path.join(helpers, tuple_ + "-detect-lib") if not os.access(helper, os.X_OK): continue try: completed = subprocess.run([helper], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=30, check=False) except (OSError, subprocess.SubprocessError): continue token = completed.stdout.decode(errors="replace").strip() if completed.returncode == 0 and token: tokens[tuple_] = token return tokens def merge_preload_modules(modules, root, tokens): """One loader-expanded path per module name, for a set of per-ABI modules. pv-adverb pairs a 32-bit and a 64-bit copy of a module into a single ``${PLATFORM}`` entry once it knows the loader's tokens, which it cannot learn under fakechroot. The same pairing is done here with ``$LIB``: each copy is linked under the tree by its architecture's library directory, and the loader picks the right one for every process. Args: modules: (option, path, tuple) triples from ``--ld-preload=PATH:abi=T`` and ``--ld-audit`` arguments. Returns: The replacement adverb arguments. """ directory = os.path.join(root.stage, "modules") groups = {} for option, path, tuple_ in modules: groups.setdefault((option, os.path.basename(path)), []).append((path, tuple_)) out = [] for (option, base), copies in groups.items(): if len(copies) < 2 or any(t not in tokens for _, t in copies): out.extend(f"{option}={path}:abi={tuple_}" for path, tuple_ in copies) continue for path, tuple_ in copies: where = os.path.join(directory, tokens[tuple_]) os.makedirs(where, exist_ok=True) link = os.path.join(where, base) if os.path.lexists(link): os.unlink(link) os.symlink(root.path + path if path.startswith("/") else path, link) out.append("{}={}".format(option, os.path.join(directory, "$LIB", base))) return out def adjust_adverb(command, root, libraries): """Rewrite pv-adverb's arguments for fakechroot. ldconfig is statically linked, so the cache it would regenerate is neither written nor read here; the search path it stood for becomes a complete LD_LIBRARY_PATH of tree paths instead. Preload modules become paths the dynamic loader can follow on the host, and the fakechroot library itself is added as a module so pv-adverb keeps it in the LD_PRELOAD it composes for the game. Returns: The rewritten command, and the LD_LIBRARY_PATH pv-adverb itself runs with. """ if not command or os.path.basename(command[0]) != "pv-adverb": return command, None tokens = libdl_lib_tokens(root.path + os.path.dirname(command[0])) prefixed = [] aliases = None modules = [] out = [command[0]] i = 1 while i < len(command): arg = command[i] if arg == "--": break if arg == "--regenerate-ld.so-cache": i += 2 continue if arg == "--add-ld.so-path": prefixed.append(command[i + 1]) i += 2 continue if arg == "--set-ld-library-path": aliases = command[i + 1] i += 2 continue if arg.startswith(("--ld-preload=", "--ld-audit=")) and ":abi=" in arg: option, value = arg.split("=", 1) path, tuple_ = value.rsplit(":abi=", 1) modules.append((option, path, tuple_)) i += 1 continue if arg.startswith(("--ld-preload=/", "--ld-audit=/")): option, value = arg.split("=", 1) arg = f"{option}={root.path}{value}" out.append(arg) i += 1 search = list(prefixed) if aliases: search.extend(p for p in aliases.split(":") if p and p not in search) search.extend(p for p in ld_so_conf_dirs(root.path) if p not in search) library_path = ":".join(root.path + p for p in search) out.extend(["--set-ld-library-path", library_path]) out.extend(merge_preload_modules(modules, root, tokens)) out.append("--ld-preload=" + fakechroot_preload(root, libraries)) out.extend(command[i:]) return out, library_path def fakechroot_preload(root, libraries): """The LD_PRELOAD entry loading fakechroot into either architecture. The dynamic loader expands ``$LIB`` to the multiarch library directory, so one path serves 64-bit and 32-bit programs when both libraries sit at the same place under it; otherwise only the primary library is named. """ directory = os.path.join(root.stage, "preload") os.makedirs(directory, exist_ok=True) common = None for tuple_, path in libraries.items(): marker = f"/{tuple_}/" if marker in path: form = path.replace(marker, "/$LIB/", 1).replace("/usr/lib/$LIB/", "/usr/$LIB/", 1) if common is None: common = form elif common != form: common = False if common and len(libraries) > 1: return common return libraries.get("x86_64-linux-gnu") or next(iter(libraries.values())) class Runner: """Executes the command in the tree and supervises it like bwrap would.""" def __init__(self, ops, command, backend, detail, stage): self.ops = ops self.command = command self.backend = backend self.detail = detail self.stage = stage self.env = dict(os.environ) self.cwd = None self.argv0 = None self.new_session = False self.die_with_parent = False self.info_fds = [] self.status_fds = [] self.keep_fds = [] self.lock_files = [] self.locks = [] for option, values, _ in ops: if option == "--setenv": self.env[values[0]] = values[1] elif option == "--unsetenv": self.env.pop(values[0], None) elif option == "--clearenv": self.env = {} elif option == "--chdir": self.cwd = values[0] elif option == "--argv0": self.argv0 = values[0] elif option == "--new-session": self.new_session = True elif option == "--die-with-parent": self.die_with_parent = True elif option == "--info-fd": self.info_fds.append(int(values[0])) elif option == "--json-status-fd": self.status_fds.append(int(values[0])) elif option == "--sync-fd": self.keep_fds.append(int(values[0])) elif option == "--lock-file": self.lock_files.append(values[0]) elif option in ("--seccomp", "--add-seccomp-fd", "--block-fd", "--userns-block-fd", "--userns", "--userns2", "--pidns"): if option == "--block-fd": read_fd(int(values[0])) os.close(int(values[0])) def build(self): """Materialize the tree and return the executor's argv, env, and cwd.""" root = Root(self.stage, self.backend == "fakechroot") root.apply(self.ops) self.repoint_wine_drive(root) for path in self.lock_files: fd = os.open(root.path + path, os.O_RDONLY | os.O_CLOEXEC) fcntl.lockf(fd, fcntl.LOCK_SH) self.locks.append(fd) if self.backend == "proot": return self.build_proot(root) return self.build_fakechroot(root) def build_proot(self, root): argv = [self.detail, "-r", root.path] for source, dest in root.binds: argv.extend(["-b", f"{source}:{dest}"]) if self.cwd: argv.extend(["-w", self.cwd]) argv.extend(self.command) return self.detail, argv, self.env, None def build_fakechroot(self, root): libraries = self.detail copy = runtime_copy(self.ops) if copy: root.prefix_absolute_symlinks(copy) command, library_path = adjust_adverb(self.command, root, libraries) env = dict(self.env) env["FAKECHROOT_BASE"] = root.path if len(root.excluded) >= 99: log("more than 99 pass-through paths; fakechroot ignores the rest") env["FAKECHROOT_EXCLUDE_PATH"] = ":".join([root.path] + root.excluded[:99]) env.pop("FAKECHROOT", None) env["LD_PRELOAD"] = fakechroot_preload(root, libraries) if library_path is not None: env["LD_LIBRARY_PATH"] = library_path elif "LD_LIBRARY_PATH" in env: env["LD_LIBRARY_PATH"] = ":".join( root.path + p if p.startswith("/") else p for p in env["LD_LIBRARY_PATH"].split(":") if p) executable = command[0] if "/" in executable: executable = root.translate(os.path.join(self.cwd or "/", executable)) else: # A bare name is looked up in the tree first and on the host # second, so that a container assembled entirely out of # pass-through binds still finds its command. for translate in (root.translate, lambda path: path): found = None for directory in (env.get("PATH") or os.defpath).split(":"): candidate = translate(os.path.join(directory, executable)) if os.access(candidate, os.X_OK): found = candidate break if found is not None: executable = found break if self.argv0: command = [self.argv0] + command[1:] cwd = root.translate(self.cwd) if self.cwd else None return executable, command, env, cwd def repoint_wine_drive(self, root): """Point a Wine prefix's Z: drive at this container's root. Wine records the container's root as the Unix path behind Z: when it creates a prefix, and every later launch reuses what it recorded. That path is the tree under fakechroot and `/` under proot, so a prefix one executor created leaves the other unable to reach the game. The drive is rewritten to whatever this launch's root is, which is Steam-specific knowledge, as `steam_browser_helper` is. """ data = self.env.get("STEAM_COMPAT_DATA_PATH") if not data: return drive = os.path.join(data, "pfx", "dosdevices", "z:") wanted = root.path if self.backend == "fakechroot" else "/" try: if os.readlink(drive).rstrip("/") == wanted.rstrip("/"): return os.unlink(drive) os.symlink(wanted, drive) except OSError as error: debug(f"leaving {drive} alone: {error}") def run(self): """Fork the executor, wait for it, and mirror its exit like bwrap.""" executable, argv, env, cwd = self.build() env["PWD"] = self.cwd or "/" debug("executing {}: {}".format(executable, " ".join(repr(a) for a in argv))) if self.die_with_parent: set_pdeathsig(signal.SIGKILL) pid = os.fork() if pid == 0: try: if self.new_session: os.setsid() set_pdeathsig(signal.SIGKILL) if cwd: os.chdir(cwd) os.execve(executable, argv, env) except OSError as error: log(f"execve {executable}: {error}") os._exit(127) if self.info_fds or self.status_fds: import json for fd in self.info_fds + self.status_fds: try: os.write(fd, (json.dumps({"child-pid": pid}) + "\n").encode()) except OSError: pass forwarded = (signal.SIGTERM, signal.SIGINT, signal.SIGHUP, signal.SIGUSR1, signal.SIGUSR2) def forward(signum, _frame): try: os.kill(pid, signum) except ProcessLookupError: pass for signum in forwarded: signal.signal(signum, forward) while True: try: _, status = os.waitpid(pid, 0) break except InterruptedError: continue for fd in self.status_fds: try: import json os.write(fd, (json.dumps({"exit-code": os.waitstatus_to_exitcode(status)}) + "\n").encode()) except (OSError, ValueError): pass return status def set_pdeathsig(signum): """Ask the kernel to deliver signum when the parent process dies.""" try: import ctypes libc = ctypes.CDLL(None, use_errno=True) libc.prctl(1, signum, 0, 0, 0) except (OSError, AttributeError): pass def main(argv): if len(argv) == 1 or argv[1] == "--help": sys.stdout.write(HELP) for name in sorted(FLAGS | ONE_ARG | TWO_ARGS | THREE_ARGS): sys.stdout.write(f" {name}\n") return 0 if argv[1] == "--version": print(VERSION) return 0 try: ops, command = parse(argv[1:]) if not command: raise UsageError("COMMAND is required") backend, detail = choose_backend(command) base = runtime_base() except UsageError as error: log(str(error)) return 1 collect_garbage(base) stage = stage_directory(base, ops) debug(f"backend {backend}, tree {stage}") inert = sorted({option for option, _, _ in ops if option in INERT}) if inert: debug(f"no effect without namespaces: {' '.join(inert)}") try: runner = Runner(ops, command, backend, detail, stage) status = runner.run() except UsageError as error: log(str(error)) shutil.rmtree(stage, ignore_errors=True) return 1 finally: pass shutil.rmtree(stage, ignore_errors=True) if os.WIFSIGNALED(status): signal.signal(os.WTERMSIG(status), signal.SIG_DFL) os.kill(os.getpid(), os.WTERMSIG(status)) return 128 + os.WTERMSIG(status) return os.WEXITSTATUS(status) if __name__ == "__main__": sys.exit(main(sys.argv))