diff options
| -rwxr-xr-x | examples/unschroot.py | 288 |
1 files changed, 288 insertions, 0 deletions
diff --git a/examples/unschroot.py b/examples/unschroot.py new file mode 100755 index 0000000..5d847f4 --- /dev/null +++ b/examples/unschroot.py @@ -0,0 +1,288 @@ +#!/usr/bin/python3 +# Copyright 2024 Helmut Grohne <helmut@subdivi.de> +# SPDX-License-Identifier: GPL-3 + +"""Emulate schroot using namespaces sufficiently well that sbuild can deal with +it but not any better. It assumes that ~/.cache/sbuild contains tars suitable +for sbuild --chroot-mode=unshare. Additionally, those tars are expected to +contain the non-essential passwd package. The actual sessions are stored in +~/.cache/unschroot. For using it with sbuild, your sbuildrc should contain: + + $chroot_mode = "schroot"; + $schroot = "/path/to/unschroot"; +""" + + +import argparse +import functools +import grp +import itertools +import os +import pathlib +import pwd +import shutil +import signal +import socket +import sys +import tempfile +import typing + +if __file__.split("/")[-2:-1] == ["examples"]: + sys.path.insert(0, "/".join(__file__.split("/")[:-2])) + +import linuxnamespaces +import linuxnamespaces.tarutils + + +class TarFile( + linuxnamespaces.tarutils.ZstdTarFile, linuxnamespaces.tarutils.XAttrTarFile +): + pass + + +class Chroot: + # Ignore $HOME as sbuild sets to something invalid + home = pathlib.Path(pwd.getpwuid(os.getuid()).pw_dir) + cache_sbuild = home / ".cache/sbuild" + cache_unschroot = home / ".cache/unschroot" + + def __init__(self, path: pathlib.Path, aliases: set[str] | None = None): + self.path = path + self.aliases = set() if aliases is None else aliases + + @functools.cached_property + def namespace(self) -> str: + if self.path.is_file(): + return "Chroot" + if self.path.is_dir(): + return "Session" + raise ValueError("invalid chroot object") + + @functools.cached_property + def name(self) -> str: + suffix = "-sbuild" if self.namespace == "Chroot" else "" + return self.path.name.split(".", 1)[0] + suffix + + def infostr(self) -> str: + lines = [ + f"--- {self.namespace} ---", + f"Name {self.name}", + ] + if self.namespace == "Chroot": + lines.extend(["Type file", f"File {self.path}"]) + if self.namespace == "Session": + lines.append(f"Location {self.path}") + if self.aliases: + lines.append("Aliases " + " ".join(sorted(self.aliases))) + return "".join(map("%s\n".__mod__, lines)) + + @classmethod + def searchchroot(cls, name: str) -> "Chroot": + name = name.removeprefix("chroot:") + name = name.removesuffix("-sbuild") + for path in cls.cache_sbuild.iterdir(): + if path.name.startswith(name + ".t"): + return cls(path) + raise KeyError(name) + + @classmethod + def searchsession(cls, name: str) -> "Chroot": + name = name.removeprefix("session:") + path = cls.cache_unschroot / name + if not path.is_dir(): + raise KeyError(name) + return cls(path) + + @classmethod + def newsession(cls) -> "Chroot": + cls.cache_unschroot.mkdir(parents=True, exist_ok=True) + return Chroot( + pathlib.Path( + tempfile.mkdtemp(prefix="chroot", dir=cls.cache_unschroot) + ), + ) + + @classmethod + def scan_sbuild(cls) -> typing.Iterator["Chroot"]: + if cls.cache_sbuild.is_dir(): + chroots = [] + aliases: dict[str, set[str]] = {} + for path in cls.cache_sbuild.iterdir(): + if path.is_symlink(): + alias = path.name.split(".", 1)[0] + "-sbuild" + aliases.setdefault(str(path.readlink()), set()).add(alias) + elif path.is_file(): + chroots.append(path) + for path in chroots: + yield cls(path, aliases.get(path.name, set())) + + @classmethod + def scan_unschroot(cls) -> typing.Iterator["Chroot"]: + if cls.cache_unschroot.is_dir(): + yield from map(cls, cls.cache_unschroot.iterdir()) + + +def do_info(args: argparse.Namespace) -> None: + """Show information about selected chroots""" + chroots: typing.Iterable[Chroot] + if args.chroot: + try: + chroots = [Chroot.searchchroot(args.chroot)] + except KeyError: + chroots = [Chroot.searchsession(args.chroot)] + else: + chroots = itertools.chain( + Chroot.scan_sbuild(), Chroot.scan_unschroot() + ) + sys.stdout.write("\n".join(chroot.infostr() for chroot in chroots)) + + +def do_begin_session(args: argparse.Namespace) -> None: + """Begin a session; returns the session ID""" + source = Chroot.searchchroot(args.chroot) + session = Chroot.newsession() + uidmap = linuxnamespaces.IDAllocation.loadsubid("uid").allocatemap(65536) + gidmap = linuxnamespaces.IDAllocation.loadsubid("gid").allocatemap(65536) + mainsock, childsock = socket.socketpair() + with TarFile.open(source.path, "r:*") as tarf: + pid = os.fork() + if pid == 0: + mainsock.close() + os.chdir(session.path) + linuxnamespaces.unshare( + linuxnamespaces.CloneFlags.NEWUSER + | linuxnamespaces.CloneFlags.NEWNS, + ) + childsock.send(b"\0") + childsock.recv(1) + childsock.close() + os.setgid(0) + os.setuid(0) + for tmem in tarf: + if not tmem.name.startswith(("dev/", "./dev/")): + tarf.extract(tmem, numeric_owner=True) + sys.exit(0) + childsock.close() + mainsock.recv(1) + linuxnamespaces.newidmaps(pid, [uidmap], [gidmap]) + linuxnamespaces.unshare_user_idmap( + [uidmap, linuxnamespaces.IDMapping(65536, os.getuid(), 1)], + [gidmap, linuxnamespaces.IDMapping(65536, os.getgid(), 1)], + ) + os.chown(session.path, 0, 0) + session.path.chmod(0o755) + mainsock.send(b"\0") + mainsock.close() + _, ret = os.waitpid(pid, 0) + print(session.name) + sys.exit(ret) + + +def do_run_session(args: argparse.Namespace) -> None: + """Run an existing session""" + session = Chroot.searchsession(args.chroot) + uidmap = linuxnamespaces.IDAllocation.loadsubid("uid").allocatemap(65536) + gidmap = linuxnamespaces.IDAllocation.loadsubid("gid").allocatemap(65536) + mainsock, childsock = socket.socketpair() + pid = os.fork() + if pid == 0: + mainsock.close() + os.chdir(session.path) + ns = ( + linuxnamespaces.CloneFlags.NEWUSER + | linuxnamespaces.CloneFlags.NEWNS + | linuxnamespaces.CloneFlags.NEWPID + ) + linuxnamespaces.unshare(ns) + childsock.send(b"\0") + childsock.recv(1) + if os.fork() != 0: + sys.exit(0) + assert os.getpid() == 1 + with linuxnamespaces.FileDescriptor(os.pidfd_open(1, 0)) as pidfd: + socket.send_fds(childsock, [b"\0"], [pidfd]) + os.setgid(0) + os.setuid(0) + linuxnamespaces.bind_mount(".", "/mnt", recursive=True) + os.chdir("/mnt") + linuxnamespaces.populate_sys("/", ".") + linuxnamespaces.populate_proc("/", ".", ns) + linuxnamespaces.populate_dev("/", ".") + linuxnamespaces.pivot_root(".", ".") + linuxnamespaces.umount(".", linuxnamespaces.UmountFlags.DETACH) + os.chdir(args.directory or "/") + if args.user.isdigit(): + spw = pwd.getpwuid(int(args.user)) + else: + spw = pwd.getpwnam(args.user) + supplementary = [ + sgr.gr_gid for sgr in grp.getgrall() if spw.pw_name in sgr.gr_mem + ] + os.setgroups(supplementary) + os.setgid(spw.pw_gid) + os.setuid(spw.pw_uid) + if not args.command: + args.command.append("bash") + linuxnamespaces.prctl_set_pdeathsig(signal.SIGTERM) + if "PATH" not in os.environ: + if spw.pw_uid == 0: + os.environ["PATH"] = "/usr/sbin:/sbin:/usr/bin:/bin" + else: + os.environ["PATH"] = "/usr/bin:/bin" + os.execvp(args.command[0], args.command) + childsock.close() + mainsock.recv(1) + linuxnamespaces.newidmaps(pid, [uidmap], [gidmap]) + linuxnamespaces.prctl_set_child_subreaper(True) + mainsock.send(b"\0") + _data, fds, _flags, _address = socket.recv_fds(mainsock, 1, 1) + pidfd = fds[0] + os.waitpid(pid, 0) + linuxnamespaces.prctl_set_child_subreaper(False) + sys.exit(os.waitid(os.P_PIDFD, pidfd, os.WEXITED).si_status) + + +def do_end_session(args: argparse.Namespace) -> None: + """End an existing session""" + session = Chroot.searchsession(args.chroot) + uidmap = linuxnamespaces.IDAllocation.loadsubid("uid").allocatemap(65536) + gidmap = linuxnamespaces.IDAllocation.loadsubid("gid").allocatemap(65536) + linuxnamespaces.unshare_user_idmap( + [uidmap, linuxnamespaces.IDMapping(65536, os.getuid(), 1)], + [gidmap, linuxnamespaces.IDMapping(65536, os.getgid(), 1)], + ) + shutil.rmtree(session.path) + + +def main() -> None: + parser = argparse.ArgumentParser() + group = parser.add_mutually_exclusive_group(required=True) + for comm in ("info", "begin-session", "run-session", "end-session"): + func = globals()["do_" + comm.replace("-", "_")] + group.add_argument( + f"-{comm[0]}", + f"--{comm}", + dest="command", + action="store_const", + const=func, + help=func.__doc__, + ) + parser.add_argument( + "-c", + "--chroot", + dest="chroot", + action="store", + help="Use specified chroot", + ) + parser.add_argument("-d", "--directory", action="store") + 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=os.getlogin()) + parser.add_argument("command", nargs="*") + args = parser.parse_args() + assert args.command is not None + args.command(args) + + +if __name__ == "__main__": + main() |